Hi 😄 I have a quick question is it a good practice to use the prefect api to launch flows ? If you liked answering this question feel free to answer this wonderful question it will help me a lot ! Thanks prefecters 💌

hello how can I cancel a flow in an if condition?

Hi everyone, quick question Can I protect prefect/Orion UI by putting a password on it? Anyone can currently use the URL to access the UI.

Hi i am running prefect using minio as a object storage, i am getting this error while triggering the flow using "create_flow_run_from_deployment" Question is : 1. am i not able to successfully load flow from minio 2. or am i not able to pull SLACK_WEBHOOK_URL from .env file

Hello Everyone, I have some question in using S3 storage with ECSRun configuration with Prefect V1. Currently we are using Docker storage with DockerRun and we can set additional files that our flow needs by

Docker(files={"additional_file.py": "additional_file.py})

Question: • How we can set additional files that our flows needs using S3 storage with ECSRun configuration? • If its not possible to set the additional files, is there any recommended workaround ?

Hi, I use ignore_file to ignore large files does not work

flow = Deployment.build_from_flow(
            flow=fusionist_flow,
            name=f"fusionist-{i}",
            schedule=(IntervalSchedule(
                interval=timedelta(hours=24).total_seconds(),
                timezone="Asia/Shanghai",
                anchor_date=pendulum.now().add(seconds=15),
            )),
            parameters={"account_index": i},
            work_queue_name="browser",
            ignore_file=".prefectignore"
        )
        flow.apply()

cat .prefectignore
venv
web

This message was deleted.

We use prefect 2 and all of our flows use

prefect-shell

to send commands to spin up docker containers. We do get some logs from within the container processes but not all of them - then we have to find the container or process on our VM and look at the logs there - it’d be great to have all logs show up in the prefect dashboard. It’s not clear to me why we get some logs and not others from our docker containers. Any guidance would be appreciated.

Hi ! I'm starting out with prefect 2 and was wondering if it's possible to create a graph from a flow definition ? And the validate this graph ? let's say I have the following flow

@flow
def my_flow():
    a = subflow1(...)
    b = subflow2(...)
    c = subflow3(a)

Would it be possible to create a graph from this and raise an error because subflow2 is a dangling node ?

Hi everyone, I am using this below code to trigger the deployment

from prefect.client import get_client
from datetime import datetime
import asyncio
import random
import uuid

now = datetime.now()
current_time = now.strftime("%H:%M:%S")
print("Current Time =", current_time)

async def trigger_flow_runs():
    client = get_client()
    print(client, ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")
    d_id_list = ["ddf0ae5c-7ed4-48b2-b417-535d767933f2"]
    choosen_queue = random.choice(d_id_list)
    print("Deployment ID ", choosen_queue)
    param = {"my_param": str(uuid.uuid4()), "file_path": "root/flows/tmp_folder"}
    print(param)
    t = await client.create_flow_run_from_deployment(deployment_id=choosen_queue, parameters=param, db: session, name=str(uuid.uuid4))
    print(".....................................................................................................................................................................................")
    print(t)
    print(".....................................................................................................................................................................................")
if __name__ == "__main__":
    asyncio.run(trigger_flow_runs())

but it is giving me this below error, i don't know why it is trying sqllite3

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: deployment
[SQL: SELECT deployment.id AS deployment_id, deployment.created AS deployment_created, deployment.updated AS deployment_updated, deployment.name AS deployment_name, deployment.version AS deployment_version, deployment.description AS deployment_description, deployment.manifest_path AS deployment_manifest_path, deployment.work_queue_name AS deployment_work_queue_name, deployment.infra_overrides AS deployment_infra_overrides, deployment.path AS deployment_path, deployment.entrypoint AS deployment_entrypoint, deployment.schedule AS deployment_schedule, deployment.is_schedule_active AS deployment_is_schedule_active, deployment.tags AS deployment_tags, deployment.parameters AS deployment_parameters, deployment.parameter_openapi_schema AS deployment_parameter_openapi_schema, deployment.created_by AS deployment_created_by, deployment.updated_by AS deployment_updated_by, deployment.flow_id AS deployment_flow_id, deployment.work_pool_queue_id AS deployment_work_pool_queue_id, deployment.infrastructure_document_id AS deployment_infrastructure_document_id, deployment.storage_document_id AS deployment_storage_document_id, work_pool_1.id AS work_pool_1_id, work_pool_1.created AS work_pool_1_created, work_pool_1.updated AS work_pool_1_updated, work_pool_1.name AS work_pool_1_name, work_pool_1.description AS work_pool_1_description, work_pool_1.type AS work_pool_1_type, work_pool_1.base_job_template AS work_pool_1_base_job_template, work_pool_1.is_paused AS work_pool_1_is_paused, work_pool_1.default_queue_id AS work_pool_1_default_queue_id, work_pool_1.concurrency_limit AS work_pool_1_concurrency_limit, work_pool_queue_1.id AS work_pool_queue_1_id, work_pool_queue_1.created AS work_pool_queue_1_created, work_pool_queue_1.updated AS work_pool_queue_1_updated, work_pool_queue_1.name AS work_pool_queue_1_name, work_pool_queue_1.description AS work_pool_queue_1_description, work_pool_queue_1.is_paused AS work_pool_queue_1_is_paused, work_pool_queue_1.concurrency_limit AS work_pool_queue_1_concurrency_limit, work_pool_queue_1.priority AS work_pool_queue_1_priority, work_pool_queue_1.work_pool_id AS work_pool_queue_1_work_pool_id
FROM deployment LEFT OUTER JOIN work_pool_queue AS work_pool_queue_1 ON work_pool_queue_1.id = deployment.work_pool_queue_id LEFT OUTER JOIN work_pool AS work_pool_1 ON work_pool_1.id = work_pool_queue_1.work_pool_id
WHERE deployment.id = :pk_1]
[parameters: {'pk_1': 'ddf0ae5c-7ed4-48b2-b417-535d767933f2'}]
(Background on this error at: <https://sqlalche.me/e/14/e3q8>)

I an using this below docker-compose file

version: "3.3"

services:
  ### Prefect Database
  database:
    image: postgres:15.1-alpine
    restart: always
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=orion
    expose:
      - 5432
    volumes: 
      - db:/var/lib/postgresql/data
  
  ### MinIO for flow storage
  minio:
    image: minio/minio:latest
    entrypoint: ["minio", "server", "--address", "0.0.0.0:9000", "--console-address", "0.0.0.0:9002", "/data"]
    volumes:
      - "minio:/data"
    ports:
      - 9000:9000
      - 9002:9002

  ### Prefect Orion API
  orion:
    image: prefecthq/prefect:2.7.7-python3.11
    restart: always
    volumes:
      - prefect:/root/.prefect
    entrypoint: ["prefect", "orion", "start"]
    working_dir: "/root/flows"
    volumes:
      - "./flows:/root/flows"
    environment:
      - PREFECT_API_URL=<http://172.16.12.17:4200/api>
      - PREFECT_ORION_API_HOST=0.0.0.0
      - PREFECT_ORION_DATABASE_CONNECTION_URL=<postgresql+asyncpg://postgres:postgres@database:5432/orion>
    ports:
      - 4200:4200
    depends_on:
      - database

  ## Prefect Agent
  agent:
    image: prefecthq/prefect:2.7.7-python3.11
    restart: always
    entrypoint: ["prefect", "agent", "start", "-q", "rbfracture", "testing"]
    environment:
      - PREFECT_API_URL=<http://172.16.12.17:4200/api>


  # ### Prefect CLI
  # cli:
  #   image: prefecthq/prefect:2.7.7-python3.11
  #   entrypoint: "bash"
  #   working_dir: "/root/flows"
  #   volumes:
  #     - "./flows:/root/flows"
  #   environment:
  #     - PREFECT_API_URL=<http://orion:4200/api>
      
volumes:
  prefect:
  db:
  minio:


networks:
  default:
    external:
      name: carpl_docker_backend

Can anyone please tell me how i can trigger it

hello! Is there a way to use cache when installing packages in block of docker container?

Hi, I am getting 500 internal server error in deployment api while applying deployment to the cloud , below is attached the screenshot for reference

Hey, how do we upload local files to the S3 bucket using the S3 bucket block? I have tried several times but am getting this kind of error:

KeyError: 'credentials'

Hi everyone. I am trying to deploy KubernetesJob using Kubernetes Agent. Everything is working fine but the problem is our cluster uses Isitio Sidecar I know how to disable it but I am not sure how to do that with perfect deploy. Can someone please share their knowledge in this regard

Hello 😄 I'm currently trying to understand the concept of infrastructure and parallel task runners. My goal is to run potentially high cpu and memory using tasks under a subflow in parallel and in independent "containers/environments". My assumption was that I could e.g. specify to have each task be running in it's own docker container by using the docker infrastructure. But my current results always end up to be running the whole flow in one container. Am I missing something or is this just not in the concept of prefect? Example code but build with

prefect deployment build my_flow.py:do_parallel_stuff -n docker -ib docker-container/docker -sb s3/some-bucket

@flow(task_runner=DaskTaskRunner)
def do_parallel_stuff():
    for x in y:
        result = do_heavy_stuff_in_own_environment.submit(x)

@task()
def do_heavy_stuff_in_own_environment(x: str):
    # Does heavy stuff

Thanks in advance!

prefect agents running on our VM kick off flows, all of which have tasks that execute shell commands that spin up docker containers. We have an issue with two flows that use parallelization inside the container they’re running in using the R programming language. These two flows run just fine if we run the docker containers outside of a prefect context - but when these flows are triggered by a prefect agent the flows start but then hang after a while (e.g., 30 min to 1 hr) with no logs returned, no errors, just hanging. It seems a little surprising if this was due to the prefect python client, but that seems to be a likely source of the problem. Any ideas? Anyone else run into this?

Hi i am running prefect using docker and pushing my flows in S3 storage block, but while flow run, i am not able to download the flows This is the log, after that it get stuck

11:31:19.076 | INFO    | prefect.agent - Submitting flow run 'f8b44688-71ae-464f-ba5a-2220ee44f475'
11:31:19.105 | INFO    | prefect.infrastructure.process - Opening process '<function uuid4 at 0x7fd67974b560>'...
11:31:19.110 | INFO    | prefect.agent - Completed submission of flow run 'f8b44688-71ae-464f-ba5a-2220ee44f475'
<frozen runpy>:128: RuntimeWarning: 'prefect.engine' found in sys.modules after import of package 'prefect', but prior to execution of 'prefect.engine'; this may result in unpredictable behaviour
11:31:20.431 | INFO    | Flow run '<function uuid4 at 0x7fd67974b560>' - Downloading flow code from storage at ''

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 262, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/contextlib.py", line 222, in __aexit__
    await self.gen.athrow(typ, value, traceback)
  File "/usr/local/lib/python3.11/site-packages/prefect/utilities/asyncutils.py", line 247, in asyncnullcontext
    yield
  File "/usr/local/lib/python3.11/site-packages/prefect/client/utilities.py", line 47, in with_injected_client
    return await fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/deployments.py", line 170, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=deployment.path, local_path=".")
  File "/usr/local/lib/python3.11/site-packages/prefect/filesystems.py", line 468, in get_directory
    return await self.filesystem.get_directory(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/filesystems.py", line 313, in get_directory
    return self.filesystem.get(from_path, local_path, recursive=True)
error from daemon in stream: Error grabbing logs: unexpected EOF

Can anyone please help

Hello, my organization has a standard support package with Prefect. I do not believe we have used any of our onboarding sessions. I would like to use these onboarding sessions to get help in building AWS infrastructure needed for running Prefect workflows. How can I find my company's Prefect contact?

Hello, I get a flow crash error when I run the flow via deployment. However, the same flow run successfully via command line. Would appreciate any help. I am running the deployment on a windows server.

Hello, I am taking the DTC course and now that 2nd week homework is over I hope I can get a little more answerable help to the issue I had on one of the questions. My issue was with the GIthub block assignment: On my deployment, the deployment was pulling from my local machine rather than from Github. I used other's advice to put the file in the root of the repo, but that didn't work either:

(zoom) $ prefect deployment build etl_web_to_gcs.py:etl_web_to_gcs -q gh -n "gh-web2gch-2"  -sb github/ghs-webtogcs -o gh-storage-etl-deployment.yaml --apply

I know it is pulling from local repo and not from github because among other reasons, I accidentally left an error in my local etl script that isn't in my github repo etl script and the error is showing up on the deployment. This is from the logs:

Flow run 'gamma2-ixkun-c' - Downloading flow code from storage at ''

Hi there! I'm used to airflow and getting started in Prefect. I'm used to being able to replay individual tasks from the Airflow UI within a DAG/flow run. It looks like there's no way to do that in Prefect, is that correct? If we want to replay anything, we have to replay the entire flow with a new run.

I want to set up a custom docker image for my flows to run in, simply based on the prefect python container with python packages pre-installed to save time at runtime. As a minimal beginning, I'm starting with my Dockerfile looking like this:

FROM prefecthq/prefect:2-python3.10

Here is the error I get when I try and run a flow using that docker image:

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 268, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/usr/local/lib/python3.10/site-packages/prefect/client/utilities.py", line 47, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/deployments.py", line 167, in load_flow_from_flow_run
    storage_block = Block._from_block_document(storage_document)
  File "/usr/local/lib/python3.10/site-packages/prefect/blocks/core.py", line 571, in _from_block_document
    else cls.get_block_class_from_schema(block_document.block_schema)
  File "/usr/local/lib/python3.10/site-packages/prefect/blocks/core.py", line 591, in get_block_class_from_schema
    return lookup_type(cls, block_schema_to_key(schema))
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/dispatch.py", line 186, in lookup_type
    raise KeyError(
KeyError: "No class found for dispatch key 's3-bucket' in registry for type 'Block'."

I'm not sure whether this is a misconfiguration of the flow deployment, the S3 block, or the Docker container/registry blocks, or something else.

In Prefect, what decides where tasks within a single flow are executed? Does the agent pickup an entire flow so that the output of each task can be chained together? Trying to understand this relative to Airflow which states: “Tasks are entirely isolated and may be running on entirely different machines”.

If a flow deployment is configured to use a docker image, and the prefect agent is running in an ECS task, does each flow run within that same ECS task infrastructure? Is there a way to get the agent to spin up a new ECS task for each flow run?

All of our Prefect tasks run shell commands using

prefect-shell

(we’re currently moving from using

shell_run_command

to

ShellOperation

). These shell commands all spin up docker containers. Sometimes a container will hang without any logs for up to 24 hrs. The only thing we’ve found to work is “tickling” the process connected to the container (we’ve done e..g,

tail /proc/<pid>/fd/2

), and then we get logs again. We haven’t seen this problem when we start docker containers outside of a prefect context, so we’re thinking it’s something to do with prefect. it is possible this will go away as we transition to

ShellOperation

, but perhaps not. Curious if anyone has seen this behavior too? (related thread https://prefect-community.slack.com/archives/C048ZHT5U3U/p1675282219967509 )

Hi, Is it somehow possible to start and stop a container in AWS or GCP from a flow from the Prefect Cloud UI? We want to host Metabase in a container but don't want to have it alive 24/7 and would be nice if the container could be started/stoped somehow from the Prefect UI. Also any alternatives are welcome thanks!

I am unable to install prefect in windows using "pip install prefect" command in CLI. It is throwing the following error message: Could not install packages due to an OSError: [Errno 2] No such file or directory: 'C:\\Users\\Bhnimda\\AppData\\Local\\Packages\\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\\LocalCache\\local-packages\\Python310\\site-packages\\prefect\\orion\\database\\migrations\\versions\\postgresql\\2022_04_23_132803_d38c5e6a9115_rename_block_to_blockbasis_and_.py'

I have created sample python code with flow and task in VS code editor. Also I have started server using "prefect orion start" using CLI in windows. I am not seeing the running flow in the prefect UI (http://127.0.0.1:4200/). Could any one help me on this? Thanks in Advance.

Hi I am following this tutorial to install prefect in google cloud https://medium.com/the-prefect-blog/prefect-server-101-deploying-to-google-cloud-platform-47354b16afe2, but when I get to the step

cd prefect && python3 -m pip install .

I get an error that prefect needs python 3.7, I try to upgrade to python 3.7 but I think (the repo was made with a lower version of python? I could be wrong just starting to understand docker) is there a more updated version of this article? or somebody can point me on how to install prefect on google cloud?