Q from <@U02U61Q47QR> - Hi Prefect community - I'm...
# prefect-community
a
Q from @Chris Gunderson - Hi Prefect community - I'm having an issue with the parameters used in a task ๐Ÿงต
โœ… 1
part of a question from @Chris Gunderson: When logging the parameter, I see this: [2022-08-22 123106-0500] INFO - prefect.send API request | Process: <Parameter: process_name>
Copy code
@task(name = 'send API request')
def post_request_process(process_name:str, arguments:str):
    logger = prefect.context.get('logger')

    url = "<https://XX.XXX.X.XX>:YYYYY/Process/"
    <http://logger.info|logger.info>(f'The url is: {url}')
    current_date = pendulum.now("America/Chicago").to_datetime_string()
    if process_name:
        <http://logger.info|logger.info>(f'Process: {process_name}')
    else:
        <http://logger.info|logger.info>(f'Who knows?')

with Flow('SOD Loaders CHILD - ',
          run_config=run_config) as child_flow:
    loggerFlow = prefect.context.get('logger')
    procName = Parameter("process_name", default = None)
    arguments = Parameter("arguments", default = None)
    <http://loggerFlow.info|loggerFlow.info>(f'Child Flow called with procName: {procName}')
    run_alloc = post_request_process(process_name=procName, arguments=arguments, upstream_tasks = [procName,arguments])
with Flow('SOD Loaders PARENT',
          schedule = Schedule(
                  clocks = [
                          CronClock("30 6 * * 1-5",
                                    start_date = pendulum.now("America/Chicago"),
                                    parameter_defaults = {
                                            'process_name':'custodianname',
                                    }),
                          CronClock("32 6 * * 1-5",
                                    start_date = pendulum.now("America/Chicago"),
                                    parameter_defaults = {
                                            'process_name':'custodianname2'
                                    }),
                  ]
          ),
        run_config = run_config,
        storage = CodeCommit( repo = 'repositoryname',
                  path = 'filename of script',
                  commit = 'branch',
                  secrets = ["XYZ"],
                  client_options = { "region_name":"us-east-2" })
          ) as parent_flow:

    process_name = Parameter("process_name", default=None)
    arguments = Parameter("arguments", default=None)

    child_flow.name = process_name
    loggerFlow = prefect.context.get('logger')
    <http://loggerFlow.info|loggerFlow.info>(f'Process Name: {process_name.run()}')
    <http://loggerFlow.info|loggerFlow.info>(f'Args: {arguments.run()}')
    <http://loggerFlow.info|loggerFlow.info>(f'Called child flow from parent: {child_flow.name}')
    child_flow.run(parameters = {
            "process_name": process_name,
            "arguments":arguments }
    )

if __name__ == "__main__":
    parent_flow.run(parameters = {"process_name":'Fidelity',"arguments":''}, run_on_schedule = False)
Answer: parent-child flows work a little differently. In Prefect 1.0, there is no concept of subflows. If you're just getting started with Prefect, I would recommend implementing this using subflows in Prefect 2.0: https://docs.prefect.io/tutorials/first-steps/?h=subflow#flows-tasks-and-subflows
but if you have some constraint that you can only use Prefect 1.0, I'd recommend checking out the flow of flows orchestrator pattern as described here https://www.prefect.io/guide/blog/flow-of-flows-orchestrating-elt-with-prefect-and-dbt/
c
Thank you Anna. I'll take a look.
@Anna Geller It worked! Thank you!
๐Ÿ™ 1
๐Ÿ™Œ 1