Are there any example flows involving pulling from different branches of the same repo using mapping?

Is there a way just to execute a git pull command at the start of a run, assuming the local machine has authentication already set up? My company uses Azure Devops, and as discussed in https://github.com/PrefectHQ/prefect/issues/4850, I set up a storage to pull from this via a hard-coded URL. Unfortunately the personal access token (PAT) in that URL keeps expiring in only 30 days (think I'm hitting some corporate policy limit). The standard Azure approach is to use the Git authentication manager https://docs.microsoft.com/en-us/azure/devops/repos/git/set-up-credential-managers?view=azure-devops, not SSH. I have that set up on the machines already, and a git pull would use that more reliably. Possible?

Nesting multiple asynchronous flows in Prefect can be managed using the concept of subflows or nested flows. Here's how it works, based on both Prefect 2.x and 3.x: ### Prefect 2.x In Prefect 2.x, you can create subflows by calling a flow function inside another flow. The parent flow is the main flow, while the child or "subflow" is executed within it. Subflows behave like regular flow runs and have their own task runners. Importantly, subflows block the parent flow's execution until completion, but with asynchronous subflows, you can run them concurrently using

asyncio.gather

. For more on this, you can check the Prefect 2.x documentation on nested flows. ### Prefect 3.x In Prefect 3.x, nested flows offer several benefits, such as using different task runners and enabling conditional execution of grouped tasks. They provide first-class observability, meaning you can track the status of nested flows easily within the Prefect UI. Nested flows also support asynchronous execution, allowing them to be awaited or run concurrently, adhering to async Python standards. For a deeper dive into creating asynchronous nested flows in Prefect 3.x, review the Prefect 3.x documentation on writing flows. In essence, whether you're using Prefect 2.x or 3.x, the key to nesting asynchronous flows lies in leveraging subflows with async capabilities and managing their execution so they can handle concurrent workloads effectively. Ah, the joys of managing asynchronous tasks—always making you feel like a juggler who needs just one more hand!

@Marvin i want to set a timeout for a flow and then re-run the flow automatically if the flow timeouts. what are my options here? thanks

Hey all - my prefect flow that calls a dbt cloud job failed when checking on the status of it (even though the dbt cloud job finished successfully). Here's the error:

State message: Flow run encountered an exception. json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

Looks like there was some kind of issue with the json response from dbt cloud (maybe it was empty?) We just upgraded dbt cloud to 1.5 - has anyone else ran into something similar? Is there any way to prevent a flow from failing because of a failed status API call? Maybe some way to let it retry? Flow code in comments Prefect version: 2.8.7

Hi, I have a pretty simple issue that I can't seem to find any documentation about. I set up an

orion

+

agent

container setup locally, and everything is working as expected when I try to run the flows manually. I was able to successfully create and apply a deployment as well (from within the

cli

container). However, when trying to run the deployment I am faced with a

File not found error

. Here's the traceback:

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 879, in exec_module
  File "<frozen importlib._bootstrap_external>", line 1016, in get_code
  File "<frozen importlib._bootstrap_external>", line 1073, in get_data
FileNotFoundError: [Errno 2] No such file or directory: '/tmp/tmp_zx6qpglprefect/flows/fetch_etfs.py'

My

docker-compose

is as follows:

version: "3.9"
services:
  db:
    image: timescale/timescaledb:latest-pg14
    volumes:
      - $PWD/data/db:/var/lib/postgresql/data
      - $PWD/config/db:/docker-entrypoint-initdb.d/
    healthcheck:
      test: [ "CMD-SHELL", "pg_isready" ]
      interval: 10s
      timeout: 5s
      retries: 5
    environment:
      - POSTGRES_DB=$POSTGRES_DB
      - POSTGRES_USER=$POSTGRES_USER
      - POSTGRES_PASSWORD=$POSTGRES_PASSWORD
    networks:
      - db_network
  orion:
    image: prefecthq/prefect:2.6.3-python3.10
    restart: always
    volumes:
      - $PWD/prefect:/root/.prefect
    entrypoint: [ "prefect", "orion", "start" ]
    environment:
      - PREFECT_ORION_API_HOST=0.0.0.0
      - PREFECT_ORION_DATABASE_CONNECTION_URL=$PREFECT_DB_URL
    ports:
      - 4200:4200
    depends_on:
      - db
    networks:
      - prefect_network
      - db_network

  agent:
    build: ./prefect
    restart: always
    entrypoint: [ "prefect", "agent", "start", "-q", "main_queue" ]
    environment:
      - PREFECT_API_URL=<http://orion:4200/api>
    networks:
      - prefect_network
      - db_network

  cli:
    build: ./prefect
    entrypoint: "bash"
    working_dir: "/root/flows"
    volumes:
      - "$PWD/prefect:/root"
    environment:
      - PREFECT_API_URL=<http://orion:4200/api>
    networks:
      - prefect_network
      - db_network

networks:
  db_network:
  web_network:
  prefect_network:

To create and apply the deployment(s):

> docker compose run cli
$ prefect deployment create ./flows/fetch_etf.py:flow_name --name flow_name
$ prefect deployment apply deployment.yaml

I have been running prefect server for the last month and my containers and services have stayed up and healthy. I am not sure what changed in the last few hours but my agents can no longer connect to the server and the UI no longer renders. I am not sure how to debug this. From what I can see, the services are still up. For the agents, Timeout exception, the UI tries to load but doesn't render. I can shut down and restart all containers but I would like to keep the metadata stored in the database. Can anyone help?

Also, if I want to fully reset the database (deleting all data and recreating everything), is the correct command:

prefect server database reset -y

? Thanks!

Hi Everyone, How can I pass an the current date (must update daily) to a deployment script parameter? This particular deployment needs to automatically update the "calendar_date" parameter to the latest date. I tried using the "from datetime import date" and the creating an internal variable that returns the latest date based the function date.today() but the argument does not change once the script is deployed. I also tried to use a helper function to return the latest date, but again the parameter does not change. How can I solve this problem?

from helper_functions import calc_date

from prefect.deployments import Deployment

from prefect.server.schemas.schedules import CronSchedule

from datetime import date, timedelta

_yesterday_str = calc_date.fn()

deploy_flow = Deployment.build_from_flow(

flow=my_test_deployment,

name="Test",

storage=,

path="",

entrypoint=f"xxx/yyy",

schedule=CronSchedule(cron="0 10 * * *", timezone="America/New_York"),

work_queue_name="Hawk1",

tags=["ETL", 'JSON'],

description="This Prefect flow performs ingests, loads, and Transforms data for ...",

parameters=dict(calendar_date=f'{_yesterday_str}'

)

)

if __name__ == "__main__":

deploy_flow.apply()

hi @Marvin. how to tag a flow? is it something that should be done in the code?