How can I fit in these Gitlab's sheetload script in the Prefect workflow? Is there similar example for this use case?

Hi! I want to make sure that only one run of a flow runs at a time, is there a way to restrict the paralell runs of a flow?

Hi, i have a task function which retries multiple times to check if a file is present and if successful starts the downstream task. This same task has different parameters going into it, the ones in orange color are working correctly (going into retries) but few of them are in yellow color (going in submitted state) and causing downstream task to fail. What is this submitted task? And why is it happening ?

Hi, Maybe someone can help on how to do best options in such case 1) We have a linux docker image as Prefect Agent with mounted S3 storage acting as primary storage for Flows, configs and some resources used for tasks. 2) Each Flow is executed in separate container using DockerRun(image=prefecthq/prefect:0.15.13-python3.8, host_config = host_config). So we have isolated Flow Containers inside Agent Container :). host_config is used to mount Agents S3 into Flow runtime, so that Flow, when running can pick up configs and resources mounted to Agent. Tricky thing is how to pass connection strings inside Flow Container? 1) I thought that we can pass them as arguments of DockerRun(.... environment = environment) according to (https://docker-py.readthedocs.io/en/stable/api.html#module-docker.api.container), but we need to prepare the list and it means that it will be "plain text". Not good. 2) Another option is to define some dictionary in the Flow and replace variables in CI/CD tool during deployment, but in this case it will be stored as plain text on storage inside Flow. This might be an option in case if serialized Prefect flow can be encrypted. 3) ??? maybe some other ideas?

Hi there !

I have noticed that my flow got Permission denied for write in home directory. Does anyone knows about this error? File "/usr/lib/python3.9/os.py", line 225, in makedirs mkdir(name, mode) PermissionError: [Errno 13] Permission denied: '/MboiTui' I have changed the permission for my home '/MboiTui' as 777, with chmod.

Hi there, Trying to save a csv to GCS using pandas in a Prefect cloud flow... I have set the GCP_CREDENTIALS key on prefect cloud, which I am beginning to realize is for prefect GCP tasks only, but I am getting the GCS 401 error that suggests the credentials are missing. Is there a way to pass the credentials into the pandas operation using the saved credentials on Prefect Cloud or is there another way using Prefect tasks that could work?

I tried to get the logs of my run to analyse failed runs:

prefect get logs -n name_of_the_run

but they seemed to be incomplete. I get only 2636 lines, for less than an hour of the flow. The logs are there since I see them in the UI. Any suggestions?

Hello. We are working with

prefect[aws]==0.15.3

. We have a flow that generates a large number of batch jobs, then submit them, splits them to chunks of 100 and generates an AWS waiter object per 100. We have one very large jobs with over 450 batch calls. This job seem to be stuck even though it progressed until the end. When tracking the progress we can see that a few tasks are stuck as mapped even though all child tasks are done. i.e. 472 successful definitions created in

create_ive_analysis

then all of them were submitted and chunked in

submit_jobs_analysis

and then all 5

AWSClientWait

jobs were done (we call it with map on the chunked job ids). The parent block is still in mapped mode. Also, sometimes an

AWSClientWait

tasks fails and the job doesn’t fail, it just stays there (again this is with a mapped

AWSClientWait

) Wait code

wait_res = AWSClientWait(client='batch', waiter_name='JobComplete').map(waiter_kwargs=batched_waits)

Where batched_waits is created by

@task
def submit_jobs_and_batch_job_awaits(jobs: List, batched_num=BATCH_CLIENT_WAIT_MAX_SIZE):
    submitted_jobs = [BatchSubmit().run(**job) for job in jobs]
    waits = []
    for i in range(0, len(submitted_jobs), batched_num):
        waits.append(
            {
                'jobs': submitted_jobs[i : i + batched_num],
                'WaiterConfig': {
                    'Delay': 10,
                    'MaxAttempts': 10000,
                },
            }
        )
    return

What could cause that?

Hi guys, How can I get the working directory or file path for Docker Run with GitHub Storage? I need to import some utilities from my repo, but I keep getting import errors.

Hello, I am running an Azure DevOps pipeline with service connection set up to access my AWS account. As a pipeline task, I run a sagemaker processing job that includes a Prefect flow.

- task: AWSShellScript@1
              inputs:
                # awsCredentials: 'xxxxx'
                regionName: 'xxx'
                scriptType: 'inline'
                inlineScript: python3 cli.py -remote S3 -p "['a','2020-09-01', '2020-09-02']"

I get this error:

[2022-02-03 21:29:07+0000] ERROR - prefect.S3Result | Unexpected error while reading from S3: TypeError('expected string or bytes-like object')
Traceback (most recent call last):
  File "/miniconda3/lib/python3.7/site-packages/prefect/engine/results/s3_result.py", line 142, in exists
    self.client.get_object(Bucket=self.bucket, Key=location.format(**kwargs))
  File "/miniconda3/lib/python3.7/site-packages/botocore/client.py", line 391, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/miniconda3/lib/python3.7/site-packages/botocore/client.py", line 692, in _make_api_call
    api_params, operation_model, context=request_context)
  File "/miniconda3/lib/python3.7/site-packages/botocore/client.py", line 738, in _convert_to_request_dict
    api_params, operation_model, context)
  File "/miniconda3/lib/python3.7/site-packages/botocore/client.py", line 770, in _emit_api_params
    params=api_params, model=operation_model, context=context)
  File "/miniconda3/lib/python3.7/site-packages/botocore/hooks.py", line 357, in emit
    return self._emitter.emit(aliased_event_name, **kwargs)
  File "/miniconda3/lib/python3.7/site-packages/botocore/hooks.py", line 228, in emit
    return self._emit(event_name, kwargs)
  File "/miniconda3/lib/python3.7/site-packages/botocore/hooks.py", line 211, in _emit
    response = handler(**kwargs)
  File "/miniconda3/lib/python3.7/site-packages/botocore/handlers.py", line 238, in validate_bucket_name
    if not VALID_BUCKET.search(bucket) and not VALID_S3_ARN.search(bucket):
TypeError: expected string or bytes-like object

the same task runs good when I just run it on an Azure Devops agent:

- task: AWSShellScript@1
              inputs:
                # awsCredentials: 'xxxxx'
                regionName: 'xxx'
                scriptType: 'inline'
                inlineScript: python3 cli.py -local S3 -p "['a','2020-09-01', '2020-09-02']"

has anyone had this issue before?

Hello, hope this would be the right channel for this question... • I already have my tasks working in my local machine, actually right now running my tasks manually, and I wanted to start deploying this tasks to a production env and prefect cloud • This is an ETL common process but with storing files locally to not waste too much RAM among tasks and because the requests I'm doing to extract data are really heavy • So, in the extract process, after I just finish a smaller task, I store the result in a local file • all of this is working in my local machine with no issue with flow.run() • But when I register the flow for prefect cloud the issues start ◦ right now I'm dealing with

FileNotFoundError: [Errno 2] No such file or directory: 'data/file.csv'

◦ this in prefect cloud ◦ mine is basic config. local agent in a server ◦ my code in a directory where I want to create the file with a directory ◦ the line of code is a

pathlib.mkdir

sentence I'm sure there is something I dont understand about prefect, because I cannot manipulate the local filesystem in the server with prefect cloud, any ideas?

Hi there, I’m trying to setup a demo flow package inspired from here, the idea is to keep Gitlab as a source of truth and as a

STORAGE

for the flows and have the Gitlab CI/CD register the flows to the Prefect Cloud. However, I receive an error when I try to run the dummy flow taken from the official docs :

from prefect import task, Flow
from prefect.storage import Gitlab


@task(log_stdout=True)
def hello_task():
    text = f"hello World"
    print(text)
    return text


with Flow(
    "hello-flow",
    storage=Gitlab(
        repo="predicthq/data-engineering/prefect-server",
        path="flows/hello-world/flow_hello_flow.py",
        access_token_secret="",
    ),
) as flow:
    hello_task()

The error:

(.venv) ➜  prefect-server git:(main) ✗ python flows/hello-world/flow_hello_flow.py
Traceback (most recent call last):
  File "/Users/farid/WorkSpace/Github_Repos/dataobs-poc/prefect-server/flows/hello-world/flow_hello_flow.py", line 2, in <module>
    from prefect.storage import Gitlab
ImportError: cannot import name 'Gitlab' from 'prefect.storage' (/Users/farid/WorkSpace/Github_Repos/dataobs-poc/prefect-server/.venv/lib/python3.9/site-packages/prefect/storage/__init__.py)

Looking at the source code, I don’t see anything that could cause a dependency circle. Any idea what it could be causing this? Needles to say that I have installed the gitlab package:

pip install 'prefect[gitlab]'

With Orion will we have the option to say we want this flow to run all the time ? The idea would be for that one to listen to event and start other child flow. However this flow will be in a continuous loop.The main question will be how do you release a new version cleanly as the current event loop will need to be stopped and a new instance starter

Hi, I'm looking for a workflow management solution that we can use now to replace our homebrew scripts. I've seen Orion which looks like it would tick our boxes, but I'm aware it's not ready for production use yet. So I have two questions about prefect core: 1. Is it possible to have cyclic workflows? I have a flow which transforms input records into a smaller number of output records. I'd like to then process the output records with the same flow again until it outputs no records. 2. Is it possible to have event-driven flows? We've got records coming into a queue which ideally I'd process when the queue size reaches a certain size. I have read the PINs and suspect the answer is no, but just wanted to check. I can work around it by periodically scheduling a flow that checks the size of the queue before exiting if it's too small

Hello everyone. We have a prefect-server instance running in our cluster, and we were wondering what would be the best way to organize the projects? Is it possible to create directory-like structure of the form

{DOMAIN}/{PROJECT_NAME}

(or longer) or similar? For example

iot_devices/site-temperature-management

I'm lazy to write a pull request, so I shall point out here that the link to the docs in

README.md

is broken: it links to

/core/

, which means accessing it from Github causes it to link to

<https://github.com/PrefectHQ/prefect/blob/master/core|https://github.com/PrefectHQ/prefect/blob/master/core>

instead of

<https://docs.prefect.io/core/|https://docs.prefect.io/core/>

I'd suggest linking directly to

<https://docs.prefect.io/|https://docs.prefect.io/>

as per the previous behaviour. @Anna Geller git blame says you made the change 😛

Hey everyone I am Amar. I am new to Prefect and am happy to be part of the community. Looking forward to learn from all of you. 👋

Hello everyone, tell me how it works, I do not understand. Is there an example available? here is the link

Hi there, I have this task with a trigger:

@task(trigger=any_failed)
def alert_email(email, message):
    sendmail(email, '[Prefect] ADCheck - [FAILED]', message)

with Flow("AD Check") as flow:

    # Parameters
    admin_email = Parameter("admin_email", default="<mailto:patrick@xyz.br|patrick@xyz.br>")

    # Tasks
    message = "some message"

    try:
        ad_conn = connect_active_directory(ad_server, ad_user, ad_pass)

    except Exception as e:
        message = f'Erro ao executar workflow: {e}'

    # If all tasks are successful
    notify_email(admin_email, message)

    # If any task fails
    alert_email(admin_email, message)

When the task fails, I would like to get the Exception error and save it o message variable, but I am getting

Trigger was "all_successful" but some of the upstream tasks failed

@Anna Geller, any tips? I've tried to use the state_handlers, but I need to be able to change parameters of the alert function for each workflow (email, subject, etc.) So triggers seems better. But I could not get the error.

Hi folks, I'm receiving this error everytime I try to run a Prefect script. For context I'm trying to receive Slack's status notifications so I added the SLACK_WEBHOOK_URL in the config.toml and I've also opened prefect's init.py out of curiosity. I don't think I changed anything in it but I might have miscliccked something.

File "C:\Users\andrea.nerla\AppData\Local\Programs\Python\Python39\lib\site-packages\box\box.py", line 516, in __getattr__

value = object.__getattribute__(self, item)

AttributeError: 'Config' object has no attribute 'datefmt'

My flow A has a StartFlowRun task that runs flow B. If I’m looking at a flow run of flow A in the UI, how can I navigate to the corresponding run of flow B?

Has anyone seen issues like this?

$ prefect run -n example --watch
Looking up flow metadata...Done
Creating run for flow 'example'...Done
└── Name: eccentric-groundhog
└── UUID: bc79f3ec-5525-4be6-b327-375624abb387
└── Labels: ['caretaker', 'input', 'output', 'prefect-agent-556bd57fdf-v74zj']
└── Parameters: {}
└── Context: {}
└── URL: <http://localhost:8080/default/flow-run/bc79f3ec-5525-4be6-b327-375624abb387>
Watching flow run execution...
└── 20:33:26 | INFO    | Entered state <Scheduled>: Flow run scheduled.
── 20:33:43 | WARNING | It has been 15 seconds and your flow run has not been submitted by an agent. Agent 93e9ff4d-5fce-4b1d-ad1b-59925fd32f92 (agent) has matching labels and last queried a few seconds ago. It should deploy your flow run.
└── 20:34:16 | WARNING | It has been 50 seconds and your flow run has not been submitted by an agent. Agent 93e9ff4d-5fce-4b1d-ad1b-59925fd32f92 (agent) has matching labels and last queried a few seconds ago. It should deploy your flow run.

No agent is picking up any of our flows, and flow runs just stay in the "scheduled" state even though on a CLI run, it states that there is an agent with matching labels.

Hi folks, a question about prefect cloud automations. What does the “does not start” condition really mean? Does it wait to check until the flow run is in a “Running” state; or does a “Submitted” state equate to a “start”?

Hi people, a question about environments. How do you support different environments (development / testing / production) in your code, not only env variables, but also taking into account things like different environments using different executors?

Hi I have installed the 2.0a9 . But unable to exeute prefect cmd like prefect auth login --key **********8 etc .I am getting no such cmd auth . Error: Prefect Server does not have authentication. Change your backend to Prefect Cloud with

prefect backend cloud

to log in.

Hello Myself Rishika, Product and Business Developer at COMETLABS. We are organizing our flagship hackathon for which we would like to collaborate with Perfect for partnership/sponsorship. Could someone connect me to the relevant person. Thanks and Regards Rishika Product and Growth COMETLABS

Hello everyone! During the past few weeks I have been learning how to use Prefect locally. However, now I am trying to deploy a flow using 4 OpenStack machines and I have some questions about the possible implementation: • Is it possible to make a flow of flows in which each flow executes on a specific machine of the 4 I have (without using dask)? Machines would share volumes between them. • In one of the machines I want to run a flow (triggered by other flow on a different machine) that must run a docker with a custom image when certain criteria is meet. Could I use the Docker Agent with the Local Executor for that flow? Thank you in advance! 🙏

Hello! Is there a way to have concurrency limit on the agent? (Prefect Server)

Hey there, maybe you can help me out here with some Prefect code and the rendered schematics. Context: I’m developing a flow that creates and submits simple spark apps to EMR. To my surprise there are no built-in tasks for EMR (docs). Nonetheless, I am trying to use

boto3

and

awswrangler

APIs (as I’ve seen others users asking for it on the community channel). My current issue, is on the lack of understanding on how the Flows are rendered on the Schematic view. It’s related to tasks dependencies and inter-task communication (passing values).