docker.errors.APIError: 500 Server Error for <https://35.229.119.149:2376/v1.32/build?t=xxx-xxx.jfrog.io%2Fprefect%2Fprefect_qisvgvrv-compare_sqlserver_and_snowflake%3A0.0.0&q=False&nocache=False&rm=False&forcerm=True&pull=False&dockerfile=Dockerfile>: Internal Server Error ("Syntax error - can't find = in "Driver". Must be of the form: name=value")

v 0.13.12

references_task = StartFlowRun(flow_name='references-etl', project_name='etl', wait=True)
facts_task = StartFlowRun(flow_name='facts-etl', project_name='etl', wait=True)

with Flow('main-etl') as main_flow:
    files = get_files()
    references_task.map(files)
    facts_task.map(files)


@task
def get_group(file)
    # get files "group" name from the file (20200101-facts.csv -> group=20200101)

@task
def check_references_done(group, file):
    # get references tasks from Perfect DB for the group and check they are done
    # or check if wait timeout reached (e.g. get time of the file creation and check if now - created_at > timeout)

@task
def check_no_another_facts_running(group, file):
    # check if no "do_etl" tasks from Perfect DB with "running" state exist

@task
def do_etl(group, file):
    ...

with Flow('facts-etl') as facts_flow:
    file = Parameter('file')
    group = get_group(file)
    check_references_done(group, file)
    check_no_another_facts_running(group, file)
    do_etl(group, file, upstream_tasks=[check_references_done, check_no_another_facts_running])

@task(log_stdout=True)
def sample_print_task():
    print("hello")

task_shell = ShellTask(return_all=True, log_stdout=True, log_stderr=True, stream_output=True)
with Flow("test") as flow:
    print_test = task_shell(command="echo hi", task_args={"name": "hi"})

log_to_cloud

[2020-12-16 14:48:37,015] ERROR - agent | 400 Client Error: Bad Request for url: <http://prefect-apollo:4200/>

The following error messages were provided by the GraphQL server:

    INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at
        "input.tenant_id"; Expected non-nullable type UUID! not to be null.

The GraphQL query was:

    mutation($input: get_runs_in_queue_input!) {
            get_runs_in_queue(input: $input) {
                flow_run_ids
        }
    }

The passed variables were:

    {"input": {"before": "2020-12-16T14:48:36.911186+00:00", "labels": ["talynuc", "any"], "tenant_id": null}}

The repository with name 'staging-prefect-insurance_export' does not exist in the registry with id '524279393077'

Docker(registry_url='<account_id>.<http://dkr.ecr.us-west-2.amazonaws.com|dkr.ecr.us-west-2.amazonaws.com>',
                          image_name="dev-prefect-<flow-name>",
                          python_dependencies=python_dependencies,
                          files={f: os.path.join('/modules/', os.path.basename(f))
                                                for f in get_files_in_directory(current_module_directory)},
                          env_vars = {"PYTHONPATH": "$PYTHONPATH:modules/:modules/utils/"})

None

KubernetesRun

Error: ErrImagePull

custom_confs = {
    "run_config": KubernetesRun(
        image="drtools/prefect:aircraft-etl", 
        image_pull_secrets=["regcred"], 
    ),   
    "storage": S3(bucket="dr-prefect"),
} 

with Flow("Aircraft-ETL", **custom_confs) as flow:
    airport = Parameter("airport", default = "IAD")
    radius = Parameter("radius", default = 200)
    
    reference_data = extract_reference_data()
    live_data = extract_live_data(airport, radius, reference_data)

    transformed_live_data = transform(live_data, reference_data)

    load_reference_data(reference_data)
    load_live_data(transformed_live_data)

mutation {
  delete_flow(input: 
  
    query {
  flow 
  (where: {name: {_eq: "MyFlowToDelete" }})
    {

    id
  }
}
    ) {
    success,
    error
  }
}

StartFlowRun

create_flow_run

from prefect.tasks.prefect import StartFlowRun
from prefect import Flow, task
from prefect.engine.results.local_result import LocalResult

graph_building = StartFlowRun(
      flow_name="test_flow",
      project_name="test_project",
      wait=True,
      result=LocalResult(".")
)
with Flow("Call Flow") as flow:
    end_flow = graph_building()

state = flow.run()
state.result[endflow].result #nothing here

prefect agent ecs start --cluster arn:aws:ecs:us-west-2:xxx:cluster/staging-cluster --label staging --task-role-arn arn:aws:iam::xxx:role/staging-prefect-agent-flow --log-level DEBUG --launch-type FARGATE

FARGATE

[2020-12-18 00:34:44,559] ERROR - agent | Error while deploying flow: InvalidParameterException('An error occurred (InvalidParameterException) when calling the RunTask operation: Task definition does not support launch_type FARGATE.')

{'_schema': 'Invalid data type: None'}

Flow_a: Created 3 chunks
Flow_b: Got result from previous flow None

import prefect as pf
from prefect import Client, task, Parameter, Flow
from prefect.engine.results import LocalResult
from prefect.environments.storage import Local
from prefect.utilities.configuration import set_temporary_config
from prefect.tasks.prefect import FlowRunTask
from prefect.environments import LocalEnvironment
import socket

def get_server_config(server, port):
    return {
                "cloud.api": "http://{}:{}".format(server, port),
                "cloud.graphql": "http://{}:{}/graphql".format(server, port),
                "backend": "server",
            }

def get_logger():
    return pf.context.get('logger')

@task()
def create_chunks(inputs):
    logger = get_logger()
    chunks = ['a', 'b', 'c']
    <http://logger.info|logger.info>('Created %d chunks', len(chunks))
    return chunks

@task()
def accept_results(result):
    logger = get_logger()
    <http://logger.info|logger.info>('Got result from previous flow %s', result)
    return result

def main():
    hostname = socket.gethostname()
    labels = [hostname]
    env = LocalEnvironment(labels=labels)
    with set_temporary_config(get_server_config('xxx', '4200')):
        with Flow('flow_a', storage=Local(), environment=env,
                  result=LocalResult(dir='c:/temp/flows/flow_a',
                                     location='{flow_run_id}_{task_name}_{map_index}.txt')
                  ) as flow_a:
            inputs = Parameter('inputs', required=True)
            create_chunks(inputs)

        with Flow('flow_b', storage=Local(), environment=env,
                  result=LocalResult(dir='c:/temp/flows/flow_b',
                                     location='{flow_run_id}_{task_name}_{map_index}.txt')
                  ) as flow_b:
            result = Parameter('result', required=True)
            accept_results(result)

        fa = FlowRunTask(flow_name="flow_a", project_name="Test", wait=True)
        fb = FlowRunTask(flow_name="flow_b", project_name="Test", wait=True)

        with Flow('flow_c', storage=Local(), environment=env,
                  result=LocalResult(dir='c:/temp/flows/flow_c',
                                     location='{flow_run_id}_{task_name}_{map_index}.txt')
                  ) as flow_c:
            a = fa(parameters={'inputs': {}})
            b = fb(upstream_tasks=[a], parameters={'result': a.result})

        client = Client()
        id_a = client.register(flow_a, project_name="Test")
        id_b = client.register(flow_b, project_name="Test")
        id_c = client.register(flow_c, project_name="Test")
        client.create_flow_run(flow_id=id_c)

if __name__ == "__main__":
    main()

FlowRunTask

StartFlowRun

/opt/project/dwt/extract/extract.py:43: UserWarning: A Task was passed as an argument to BookmarkEnd, you likely want to first initialize BookmarkEnd with any static (non-Task) arguments, then call the initialized task with any dynamic (Task) arguments instead. For example:

  my_task = BookmarkEnd(...)  # static (non-Task) args go here
  res = my_task(...)  # dynamic (Task) args go here