flow not found

run_config

image_name = f"{git_repo}/prefect-{Path(flow_file).stem.lower()}"
        image_tag = f"githash-{git_commit}"
        # built by cicd agent
        base_image = f"{REGISTRY_URL}/{git_repo}/flows:{image_tag}"
        # second image built
        base_image_plus_prefect = f"{REGISTRY_URL}/{image_name}:{image_tag}"

         flow.storage = Docker(
            registry_url=REGISTRY_URL,
            base_image=base_image,
            image_name=image_name,
            image_tag=image_tag,
            env_vars=fetch_env_vars(os.environ["GIT_BRANCH"]),
            # here be the trouble
            # same path as my Dockerfile???
            path=f"/repo/{flow_file}",
            stored_as_script=True,
        )
        flow.storage.build()

        flow.run_config = DockerRun(
            image=base_image_plus_prefect,
            labels=fetch_agent_labels(os.environ["GIT_BRANCH"]),
        )
        ... cicd stuff ...
        flow.register()

ValueError('Flow is not contained in this Storage')

Docker(…, files={…})

files

File "/opt/prefect/healthcheck.py", line 151, in <module>
    flows = cloudpickle_deserialization_check(flow_file_paths)
  File "/opt/prefect/healthcheck.py", line 44, in cloudpickle_deserialization_check
    flows.append(cloudpickle.loads(flow_bytes))
ModuleNotFoundError: No module named 'flows.<my flow name>'

client.create_flow_run(flow_id)

client.create_flow_run(project_name, flow_name)

def create_flow_run(flow_id=None, project_name=None, flow_name=None):
	
	if not flow_id and not (project_name and flow_name):
		raise ValueError("Either a flow_id or a project_name+flow_name must be provided")
	
	# then add if/else to handle proper graphql mutation

schedule = IntervalSchedule(start_date=datetime.datetime.now() + datetime.timedelta(seconds=1), interval=datetime.timedelta(minutes=1),)

Access denied (403) Current session has been terminated. For further information, do not hesitate to contact us.

flow = load('{project_name}.{flow})

flow.run()

PandasSerializer

GCSResults

pandas_serializer = PandasSerializer(
    file_type="parquet",
    serialize_kwargs=dict(
        engine="pyarrow", partition_cols=["date"]
    ),
)

gcs_result = GCSResult(
    bucket="bucket-name", serializer=pandas_serializer
)

class Extract(Task):
    def __init__(self, **kwargs):

        super().__init__(**kwargs, result=gcs_result, target="target_file_location")

    def run(self, df: pd.DataFrame) -> pd.DataFrame:
        # Some processing steps
        return df

SKIP signal raised: SKIP('Provided value "True" did not match "False"')

case

[2021-02-22 08:49:04-0600] INFO - prefect.TaskRunner | Task 'case(True)': Starting task run...
[2021-02-22 08:49:04-0600] INFO - prefect.TaskRunner | Task 'case(True)': Finished task run for task with final state: 'Mapped'
[2021-02-22 08:49:04-0600] INFO - prefect.TaskRunner | Task 'case(True)[0]': Starting task run...
[2021-02-22 08:49:04-0600] INFO - prefect.TaskRunner | Task 'case(True)[0]': Finished task run for task with final state: 'Success'
[2021-02-22 08:49:04-0600] INFO - prefect.TaskRunner | Task 'case(False)': Starting task run...
[2021-02-22 08:49:04-0600] INFO - prefect.TaskRunner | Task 'case(False)': Finished task run for task with final state: 'Mapped'
[2021-02-22 08:49:04-0600] INFO - prefect.TaskRunner | Task 'case(False)[0]': Starting task run...
[2021-02-22 08:49:04-0600] INFO - prefect.TaskRunner | SKIP signal raised: SKIP('Provided value "True" did not match "False"')
[2021-02-22 08:49:04-0600] INFO - prefect.TaskRunner | Task 'case(False)[0]': Finished task run for task with final state: 'Skipped'

StartFlowRun

task

state = adder.run(parameters=dict(num_lst=num_lst))
# # Get results
param_tsk = adder.get_tasks("add_num")
num_lst = state.result[param_tsk[0]]._result.value

register_task_definition()

aws ecr get-login-password --region regionname | docker login --username AWS --password-stdin <http://xxxxxx.dkr.ecr.us-east-2.amazonaws.com|xxxxxx.dkr.ecr.us-east-2.amazonaws.com>

register()

denied: Your authorization token has expired. Reauthenticate and try again.

#!/usr/bin/env python
# coding: utf-8

import pyodbc
from prefect.storage import GCS
import json
import time
import os
import datetime
from google.cloud import pubsub_v1
import pymysql
from prefect import task, Flow, Parameter



@task(log_stdout=True)
def extract_metadata(source_system,source_database_id):
    
    #provides a list of metadata to loop through and extract delta
    return metadata


def delta_push(metadata):
    # This function extract data from every metadata detail provided and pushes it further

    


@task
def delta_push_wrapper(metadata):
#Looping through ever metadata row and calling the push function 
    for metadata_row in metadata:
        delta_push(metadata_row)



with Flow("data_flow") as flow:
    flow.storage = GCS(bucket="bucketname")
    parameter1 = Parameter("paramater1",default="default")
    parameter2 = Parameter("parameter2",default=1)
    metadata = extract_metadata(parameter1,parameter2)
    delta_push_wrapper(metadata)
    

flow.register(project_name="test_project")