@Marvin, how can I use asyncio.create_task with a task.submit() to use asyncio.wait function?

Hi everyone, I'm experiencing an error in my prefect.yaml deployment in (what I think is) the Deployment action step...

prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '<http://10.10.4.100:4200/api/deployments/>'
Response: {'detail': 'Block not found.'}

I don't have a block declared for this URL so I'm not sure what is generating it. The only thing that I see is my github-credential block but that doesn't have a URL associated in the block document. I can hit the URL from my PC but I land on a browser page that says "Method Not Found" Here's my prefect.yaml, can anyone point me in the right direction? Banging my head here. :)

# Welcome to your prefect.yaml file! You can use this file for storing and managing
# configuration for deploying your flows. We recommend committing this file to source
# control along with your flow code.

# Generic metadata about this project
name: marketing
prefect-version: 3.0.11

# build section allows you to manage and build docker images
build:
- prefect.deployments.steps.run_shell_script:
    id: get-commit-hash
    script: git rev-parse --short HEAD
    stream_output: false
- prefect.deployments.steps.run_shell_script:
    id: get-prefect-version
    script: prefect --version
    stream_output: false
- prefect_docker.deployments.steps.build_docker_image:
    id: build-image
    requires: prefect-docker>=0.3.0
    image_name: 10.10.4.100:5000/something_get_app_list_landing
    tag: '{{ get-prefect-version.stdout }}-{{ get-commit-hash.stdout }}'
    # TODO:  Change this full directory before you check in to Git
    dockerfile: 
      src/prefect/deployment_image/Dockerfile
    buildargs:
      EXTRA_PIP_PACKAGES: psycopg2-binary psutil
- prefect.deployments.steps.run_shell_script:
    id: tag-image-with-registry-latest
    script: docker tag '{{build-image.image_name}}:{{build-image.tag}}' '{{build-image.image_name}}:latest'
    stream_output: true
- prefect_docker.deployments.steps.push_docker_image:
    id: push-image-version-tag
    requires: prefect-docker>=0.3.0
    image_name: '{{ build-image.image_name }}'
    tag: '{{ build-image.tag }}'
- prefect_docker.deployments.steps.push_docker_image:
    id: push-image-latest-tag
    requires: prefect-docker>=0.3.0
    image_name: '{{ build-image.image_name }}'
    tag: latest

# push section allows you to manage if and how this project is uploaded to remote locations
push:

# pull section allows you to provide instructions for cloning this project in remote locations
pull:
- prefect.deployments.steps.git_clone:
    id: clone-step
    repository: <http://10.10.4.100:3302/myusername/marketing.git>
    branch: main
    credentials: '{{ prefect.blocks.github-credentials.gitea-api-token }}'

# the deployments section allows you to provide configuration for deploying flows

deployments:
- name: something-app-list-landing-etl
  version: '{{ build-image.tag }}'
  tags:
  - latest
  description: Pulls something App List into landing tables
  entrypoint: 
    src/etl/landing/something/something_get_app_list_landing_etl_prefect.py:run_something_get_app_list_landing_etl
  parameters: {}
  work_pool:
    name: test_pool
    work_queue_name: default
    job_variables: 
      image: "{{ build-image.image }}"
  concurrency_limit:
  enforce_parameter_schema: true
schedules:
  - rrule: RRULE:FREQ=DAILY;INTERVAL=1;BYDAY=MO,TU,WE,TH,FR;BYHOUR=0;BYMINUTE=0;BYSECOND=0
    timezone: America/Phoenix
    active: true
    max_active_runs:
    catchup: false

How can I programmatically create deployments of flows located in a S3 bucket within a python process ? I am using django to upload flows into an S3 bucket ( django is used for much more then just storing meta data about a flow). Whenever a new flow is uploaded, a django signal should trigger the deployment (to a execute the flow on a local process worker. I do this with the following function that is called when the prefect flow was uploaded to the S3 storage (and a django model instance with more metadata has been created) :

def deploy_prefect_process_loc(sender, instance, **kwargs):
    """
    deploy prefect process
    """

  s3_bucket_block = S3Bucket.load("minio-prefect-flows-lara")

  print(f"******* - --- deploying prefect process {instance.name} - {prefect_workpool} ")

  flow.from_source(
        source=s3_bucket_block,
        entrypoint="robot_workflow_sila.py:science_robotic_process"
   ).deploy(
        name=f"minio-s3-deployment-lara-{instance.name}-{instance_id}",
        work_pool_name="local-process-worker-1", #prefect_workpool,
        job_variables={"env": {"EXTRA_PIP_PACKAGES": "prefect-aws"} },
        tags=["lara", "sila"],
        description="Deployment of a SiLA workflow.",
        version="v0.0.1",
  )

This programmatic deployment works fine as standalone code, but when I run it within the django process / event loop or async with celery, strange things happen - already at the stage of retrieving the S3_bucket_block: ( I added the error message in the thread) Any idea, why this is not working and how I can achieve the deployment ? Thanks 🙏

@Marvin how to trigger a flow with api webhook in prefect server?

@Marvin what's the recommended way for removing old log entries written by Prefect in a PostgreSQL database?

@Marvin give me an example how to retrieve flow id , who run the flow, and its all states and send this infor to snowflake

Hi, is there a way to export the flow logs from Prefect? I would like to save the logs for at least 90 days, preferably with a copy of the flow logs imported our own logging infrastructure (Loki) for searchability.

Hi, How to install prefect CLI on mac ?

Or how to make a task that runs after a new blob arrives in a google cloud storage bucket?

@Marvin what's the best way to run child flows in parallel?