import random
from prefect import task, Flow


class Generator:
    def __init__(self, data):
        self.particle = data


@task
def run_generator():
    return Generator(random.random())


@task
def compute(particle):
    print(particle**2)


with Flow('flow1') as flow:
    gen = run_generator()
    compute(gen.particle)

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
/var/folders/7g/2_wk8ddx7h5_4r12flp9031m0000gn/T/ipykernel_3073/2988601640.py in <module>
      1 with Flow('flow1') as flow:
      2     gen = run_generator()
----> 3     val = compute(gen.particle)

AttributeError: 'FunctionTask' object has no attribute 'particle'

compute

Map a function that adds tasks to a flow elementwise across one or more tasks. Arguments that should not be mapped over should be wrapped with prefect.unmapped.

users = ["ben", "kevin"]
CONSTANT_DATA = "something"

with Flow():
   apply_map(notify, CONSTANT_DATA)

latest

from prefect.agent.ecs.agent import ECSAgent

AGENT = ECSAgent(cluster="prefect-fargate-cluster", labels=['ecs', 'dev'])

AGENT.start()

import prefect

from prefect.storage import Docker

from prefect.run_configs import ECSRun

from prefect import task, Flow, Parameter

STORAGE = Docker(registry_url='<http://037961805145.dkr.ecr.ap-southeast-2.amazonaws.com/|037961805145.dkr.ecr.ap-southeast-2.amazonaws.com/>',

image_name='prefect-etl-flow',

image_tag='latest',

dockerfile='Dockerfile.txt')

RUN_CONFIG = ECSRun(run_task_kwargs={'cluster': 'prefect-fargate-cluster'},

execution_role_arn='arn:aws:iam::037961805145:role/prefect-ecs',

labels=['ecs', 'dev'])

@task

def testfunc():

print("Hello prefect!")

with Flow('prefect-etl-flow', storage=STORAGE, run_config=RUN_CONFIG) as flow:

testfunc()

flow.register('emmi-etl')

with Flow("impact partners",schedule=schedule) as flow:
    
    #setting default date to today - 1 to get the records from previous day
    default_date = (datetime.today() - timedelta(days=1)).strftime('%Y-%m-%d')

    start_date = Parameter('start_date', default=default_date)
    end_date = Parameter('end_date', default=default_date)

    records = get_impact_records(start_date,end_date)
    impact_partners = rename_columns(records)
    upload(data=impact_partners)
    
if __name__ == "__main__":
    main(flow)

Failed to load and execute Flow's environment: UnpicklingError("invalid load key, '{'.")

{"flow": "gASVPgk....aFtdlHViLg==", "versions": {"cloudpickle": "2.0.0", "prefect": "0.15.6", "python": "3.7.10"}}

LocalDaskExecutor

threads

mount_glue_catalog.run(schema_name=schema_name, glue_db=glue_db_name, upstream_tasks=[create_catalog])

create_catalog

TypeError: mount_glue_catalog() got an unexpected keyword argument 'upstream_tasks'

mount_glue_catalog

task(result=OurGCSResult(bucket='our-bucket'))

OurGCSResult

GCSResult

OurGCSResult

GCSResult

map_index

{
              "id": "eba36675-c15a-43ea-ad5b-2540936477b5",
              "map_index": 0,
              "name": null,
              "serialized_state": {
                "type": "Success",
                "_result": {
                  "type": "Result",
                  "location": null,  // WAT?!
                  "__version__": "0.14.22+9.g61192a3ee"
                },
                "context": {
                  "tags": []
                },
                "message": "Task run succeeded.",
                "__version__": "0.14.22+9.g61192a3ee",
                "cached_inputs": {}
              }
            }

ValueError: Local Secret "CRITEO_SECRET_KEEPS" was not found.

>>> prefect.client.Secret('CRITEO_SECRET_KEEPS').get()
'1stxxxxg'

#flows/smaple_flow.py

@task
def load():

from flows.sample_flow import load
from flows import sample_flow as flow_file

def test_flow_composition():
    
    load_task = load
    flow_tasks = flow_file.flow.tasks

    assert flow_file.flow.terminal_tasks() == set([load_task])

    # assert load_task in flow_tasks    - same things happens here , load_task is clearlya n element of the flow_tasks set but it asserts false

>       assert flow_file.flow.terminal_tasks() == set([load_task])
E       assert {<Task: load>} == {<Task: load>}
E         Extra items in the left set:
E         <Task: load>
E         Extra items in the right set:
E         <Task: load>
E         Full diff:
E           {<Task: load>}

client = docker.from_env()
host_config = client.api.create_host_config(binds=[
    '/psql/data/:/data'])
print(host_config)

container_id = CreateContainer(command="cat /data/test",
                      host_config=host_config)

container_id = CreateContainer(command="cat /data/test", host_config={'binds': ['/psql/data/:/data']}

job_template.yaml

job_template.yaml

prefect register

--network=host

GOOGLE_APPLICATION_CREDENTIALS

GOOGLE_APPLICATION_CREDENTIALS