Thanks for all the help so far. Got another question! 😄 Hopefully I can explain this okay … Using the feature_engineering.py example as a template. In the flow, there are a number of functions (not tasks) and a class is instantiated too. The last call is a task, but it’s not being attached after the class instantiation. I’m expecting the following three to run in order. Not sure that

upstream_task

will fix this because the

load(clean)

task needs to run after the

DataFrame()

bit.

clean = impute.map(data_cols, replacement_dict=unmapped({np.nan: 0})) # task
clean = DataFrame(clean, column_names) # function, not a task
load(clean) # task

Hi friends, we’re currently switching over from Docker storage to GCS storage. I noticed that every time we call

register

with an

idempotency_key

set, the flow is uploaded to storage despite it not having changed. Is this intended? Command used in the comments

Hello, We are evaluating prefect core to replace our workflow management. As part of security protocol all container images are scanned which included all the 5 docker containers of prefect core (server, Apollo, ui, Hasura/graphql & Postgres:11). All of these containers seem to have vulnerabilities and most of them are medium to low. However, the ones which are critical & high needs to be resolved before we can use them for prod, based on our observation these packages are system libs (like linux kernel, shadow, glibc etc. something which prefect doesn’t directly use). Is it possible to update the base image to say Ubuntu 20.4 (in our tests between Debian and ubuntu, the latter seems to have just about 16 [1 medium rest all low vulnerabilities])? or any other recommended approach we could follow to over come this issue is greatly appreciated. 🙂 PS - Scans have been made using GCP’s vulnerability scan service in GCR

Any tips on how to pass datetime to context?

the task decorator has a task_run_name, which results in an error if

date

becomes a string Like this

@task(task_run_name='{name}-{date:%FT%T}')                         
def say_hello(name):                                               
    logger = prefect.context.get("logger")                         
    <http://logger.info|logger.info>("Hello, %s!", name)                                
    <http://logger.info|logger.info>('Flow run name: %s', prefect.context.flow_run_name)

Hi everyone. I have a question about flows running flows, or event driven flows. I have a flow that will run and poll for some external state in a database. Once that conditional is met, I want to run a set of other flows. I looked at https://docs.prefect.io/core/idioms/flow-to-flow.html but it seems that my two options are either: 1. Add an entry in the upstream flow to call the downstream flows 2. Add a waiter in the downstream flows that could tie up a flow slot or an agent. Are there any other options for how I can tell a flow what its upstream dependencies are without creating a blocking waiter?

Hi there. A couple of questions about "git repo" storage: 1. Has anyone looked at Azure Repos? Any idea of the level of effort to implement a new storage class for Azure Repos? 2. I don't believe the storage classes are designed to work with multiple files (a file that has the flow and other "utility" modules). Am I correct? Is it possible to use the existing classes to implement this pattern or do we have to necessarily create a python package? I have a client who is considering Kubernetes orchestration but doesn't like the idea of baking the flow code in the docker image. They'd like for the flow code (and additional utility modules) to be pulled from a repo every time a flow needs to run.

How can I access LocalRun or LocalDaskRun run_config env variables during task execution? I have read though the relevant pages on docs.prefect.io, searched the examples, even browsed the source, but still no clue.

maybe I have misunderstood something and LocalRun are not meant to be used to pass environment variables to tasks

though this suggests otherwise

I'm getting an error with the following flow:

ValueError: Could not infer an active Flow context.

Details in thread

with Flow("Batch update airtable flow") as f:

    proj, stry = get_selected_bases()
    cmdf = get_communities_2017()

    scrubbed_proj, scrubbed_stry = scrub_dataframes(proj, stry, cmdf)

    debug_task(scrubbed_proj, scrubbed_stry)

I am getting a health check error when registering a flow with Docker storage:

Beginning health checks...
System Version check: OK
Traceback (most recent call last):
  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.tasks'

Registering the same flow locally is passing my check.

from prefect.utilities.debug import is_serializable

if is_serializable(flow): ...

My PYTHONPATH is the root of the directory and when SSH'ing into the container I am able to import the "missing" module... Any pointers?

Hi, community, quick question: I have a parametrized (with Prefects' Parameters) flow and i would like to schedule the same flow to run, lets say, at 14 o'clock each day. But, the same flow should run multiple times with different parameters. As i was looking into it, it seems there is no way to do it through the UI, only on the flow itself, unless i re-register the flow with different names. Is this correct? I was unable to find any guidelines regarding this on the docs. Thanks in advance.

Hi guys, I have been wondering about k8s and job executions, since it's widely known that Kubernetes cron jobs can create none or two executions of its schedule. My concern is of course that if something out of the ordinary happens in our cluster and our job either don't run or runs twice instead of the single run we would like to have. What guarantees is there for a flow when using the k8s agent that a job only runs once? Another concern is if we use CreateNamespacedJob from the task library, wouldn't that have the same problems as k8s cron jobs?

I imagine this is a fairly simple thing to figure out, but I'm not seeing it in the docs. If I have a task that takes 2 parameters, but also needs to wait until the completion of a task that doesn't return anything, how I do signal that dependency?

with Flow as f:
   x, y = func_1():
   no_return_func():
   res = final_func(x,y)

Using the example above, how do I signal for

final_func

to wait on

no_return_func

? Do I just add a third argument to the function and have

no_return_func

return

True

? Or is there a cleaner way?

hello, I was wondering if prefect had a way to interact with an AWS SNS queue. We are trying to detect the presence of a file to kick off a job. One solution is to have the flow check every 30 min or so and then delete the processed file, but I was wondering if there was a more elegant solution out there

has anyone worked with using a generic web cloud hook before? I'd like to configure the JSON payload that gets sent to the provided URL but it doesn't seem like there's a way to do that just by looking at the cloud interface?

Hi! I am wondering if anyone is able to use ShellTask on Windows? I'm running a script from tutorial and getting an error:

from prefect import Flow
from prefect.tasks.shell import ShellTask

task = ShellTask(helper_script="cd ~")

with Flow("My Flow") as f:
    contents = task(command='ls')
        
out = f.run()

[2021-02-26 10:32:28+1100] INFO - prefect.FlowRunner | Beginning Flow run for 'My Flow'
[2021-02-26 10:32:28+1100] INFO - prefect.TaskRunner | Task 'ShellTask': Starting task run...
[2021-02-26 10:32:28+1100] ERROR - prefect.TaskRunner | Unexpected error: FileNotFoundError(2, 'The system cannot find the file specified', None, 2, None)
Traceback (most recent call last):
.....
....
python\python38-32\lib\subprocess.py", line 1307, in _execute_child
    hp, ht, pid, tid = _winapi.CreateProcess(executable, args,
FileNotFoundError: [WinError 2] The system cannot find the file specified

I do have bash installed, also specifying path doesn't help

task = ShellTask(shell="C:\Windows\System32\bash.exe", helper_script="cd ~")

Trying to run a Flow of Flows but I'm getting this:

Unexpected error: ClientError('400 Client Error: Bad Request for url: <http://prefect-apollo:4200/graphql/graphql>\n\nThe following error messages were provided by the GraphQL server:\n\n GRAPHQL_VALIDATION_FAILED: Cannot query field "user" on type "Query".\n\nThe GraphQL query was:\n\n query {\n user {\n default_membership {\n tenant {\n slug\n }\n }\n }\n }\n\nThe passed variables were:\n\n null\n',)

Does anyone know how to use

case

with

mypy

? When I try to add a case to the flow, mypy throws

Module not callable

with Flow("flow") as flow:
  case(task_result, True):
    other_task()

I swear I have seen this somewhere in the docs, but for the life of me I can’t find it now. I am looking to use prefect to spin up a docker container, clone a github repo, and then run the flow defined there. I see the ability to pull the container from our registry but I cannot find how to pull the github repo

What's the correct way to store&pass environment information (like API endpoint URLs or database IPs) to flows when using Prefect Cloud? Approaches I considered (but none of them feels just right): • have multiple projects (one per enviorment), using parameters with different defaults for each environment • using agent environment variables • using secrets • storing in docker image used for flow runs

hi folks, how to set the docker storage configuration right? i have this example:

Docker(

    env_vars={
        # append modules directory to PYTHONPATH
        "PYTHONPATH": "C://Users/laura.vaida.000/anaconda3/envs/prefect/Lib/site-packages"
    },
    stored_as_script=True
)

Hi all, here’s a weird case we’re running into today. A flow running on K8s that hasn’t changed in a few wks has started failing consistently with:

Container 'flow' state: terminated
		Exit Code:: 2
		Reason: Error

The job shows up submitted, then ~30 seconds later we get 4 of those errors. Other nearly identical flows are working. The pods don’t hang around long enough to get logs (if they even start). Has anyone else run into this?

When using Github Storage I have two issues. The first is that it exposes the secret in the

KeyError

. The second is that it is showing me the correct key and yet stating that it was not found. I have the secrets defined in the Prefect UI

Failed to load and execute Flow's environment: KeyError('The secret <secret> was not found.  Please ensure that it was set correctly in your tenant: <https://docs.prefect.io/orchestration/concepts/secrets.html>')

I followed the docs to establish the secret and am using the same variable name in both the UI and the code.

GH_TOKEN

GH_TOKEN = Secret("GH_TOKEN").get()

STORAGE = GitHub(
    repo="<my_repo_name",
    path=f"flows/my_flow.py",
    access_token_secret=GH_TOKEN
)

Need help for this. This is on server as the backend. On exception I am using "raise SystemExit(0)", but the flow is not quitting immediately and takes 15 minutes or so for the flow to fail. What is the issue?

@task()
def mainprocess(a):
    global df_single_record
    try:
        for indx in df_data_extracted.index + 1:
            df_single_record = df_data_extracted.iloc[indx - 1:indx]
            converttojson(df_single_record)
            postjsontoqapi(df_json_data)
    except Exception as e:
        logger = get_logger()
        logger.error(str(get_pst_time())
                     + '==========ERROR IN mainprocess() WHILE CONVERTING TO JSON/POSTING TO Q-API: '
                      + str(e))
        raise SystemExit(0)

Is there a way to import a custom model when using

GitHub

storage? Reason for this is two-fold. Since we would be using the same block of code for just about every flow, it would help keep it semi-DRY. Second, it would allow us to pass different configs depending on environment variables (

dev

vs

prod

)

from prefect_utils import (
  RUN_CONFIG,
  STORAGE
)
flow.storage=STORAGE
flow.run_config=RUN_CONFIG

It works with the

S3

storage. But when trying to use

GitHub

I get the below.

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named 'prefect_utils'")

The module is in the same folder as the flow (

flows/

). Is this an issue because when the flow is ran it is being ran from the parent directory? thus an issue with relative paths?

Hi. In this example, if all the sub flows had the same parameters, how do I create a parameter at the parent flow to thread thru to the dependencies?

flow_a = StartFlowRun(flow_name="A", project_name="examples", wait=True)
flow_b = StartFlowRun(flow_name="B", project_name="examples", wait=True)
flow_c = StartFlowRun(flow_name="C", project_name="examples", wait=True)
flow_d = StartFlowRun(flow_name="D", project_name="examples", wait=True)

with Flow("parent-flow", schedule=weekday_schedule) as flow:
    b = flow_b(upstream_tasks=[flow_a])
    c = flow_c(upstream_tasks=[flow_a])
    d = flow_d(upstream_tasks=[b, c])

On Prefect-Cloud, I had 1000 flow runs and after roughly 150 of them completed, the remaining all failed with "_Failed after exceeding scheduled work SLA_". Is there some unwritten limit to Prefect Cloud that I'm missing? My best guess is a limit on task duration or maybe a limit on how far "behind schedule" a flow run is allowed to be.

Hi! We are developing an OCR service. I would like to know how good the prefect is for this task?