DEBUG:agent:Querying for ready flow runs...
[xx] DEBUG - agent | Creating namespaced job prefect-job-c4a440b5
DEBUG:agent:Creating namespaced job prefect-job-c4a440b5
[xx] DEBUG - agent | Job prefect-job-c4a440b5 created
DEBUG:agent:Job prefect-job-c4a440b5 created
[xx] INFO - agent | Completed deployment of flow run 549ced42-7840-43d6-89da-6f1a1f9cb378
INFO:agent:Completed deployment of flow run 549ced42-7840-43d6-89da-6f1a1f9cb378

HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"services \"foobar\" already exists","reason":"AlreadyExists","details":{"name":"foobar","kind":"services"},"code":409}

prefect.utilities.notifications.notifications.slack_notifier()

ResourceManager

default

LATEST_FLOW_BY_NAME = gql(
    '''
        query LatestFlowByName($name: String) {
          flow(
            where: {name: {_eq: $name}},
            order_by: {version: desc},
            limit: 1,
          )
          {
            id
          }
        }
    ''',
)

variable_values={'name': name}

CREATE_FLOW_RUN = gql(
    '''
        mutation CreateFlowRun($input: create_flow_run_input!) {
            create_flow_run(input: $input) {id}
        }
    ''',
)

variable_values={
    'input': {
        'flow_id': flow_id,
        'parameters': parameters,
    },
},

flow_runs

flow_runs(where:{parameters: {_has_key: "process_date"}}

process_date

person-build

person_build_flow = StartFlowRun(flow_name="person-build", project_name=get_stage(), wait=True)
release_flow = StartFlowRun(flow_name="release", project_name=get_stage(), wait=True)

with Flow(
    "build-then-release",
    executor=LocalDaskExecutor(num_workers=8),
    result=PrefectResult(),
) as flow:
    release_flow(upstream_tasks=[person_build_flow])

def handler(task, old_state, new_state):
    if new_state.is_failed():
        // do stuff here
        return Skipped()
    return new_state

@task(name="final check", state_handlers=[final_check_handler])
def final_check():
    pass

def final_check_handler(task, old_state, new_state):
    if new_state.is_skipped():
        return Failed()

flow_result = flow.run()

Succeed: all reference tasks succeeded

flow_result.result

get_duplicated_pairs_info

task_run

flow.run()

with Flow(
    name=flow_name,
    # schedule=schedule,
    state_handlers=[slack_notifier(only_states=[Success]), flow_state_handler],
) as flow:
    ---some logic---
        for sync_type, cmd_parameter in sync_commands:
            ---some logic --
            sync_command = (
                f"{command_prefix} {source_bucket} {destination_bucket} {options}"
            )
            s3_sync_task = ShellTask(
                command=sync_command,
                return_all=True,
                name=task_id,
                slug=task_id,
                timeout=timedelta(minutes=task_execution_timeout),
            )()

            # Create a task to validate sync commands.
            validate_s3_sync_task = FunctionTask(
                fn=validate_s3_sync,
                name="".join(["validate_", task_id]),
                slug="".join(["validate_", task_id]),
            )(
                timeout=timedelta(minutes=task_execution_timeout),
                s3_sync_command=sync_command,
            )

            # Define dependencies between tasks.
            s3_sync_task.set_downstream(validate_s3_sync_task)
            
# flow.run()
flow.register(project_name="demo-project", labels=tags)