How to deploy flows with Python

​

When to use flow.deploy instead of flow.serve

1. Cost optimization: Dynamic infrastructure can help reduce costs by scaling resources up or down based on demand.
2. Resource scarcity: If you have limited persistent infrastructure, dynamic provisioning can help manage resource allocation more efficiently.
3. Varying workloads: For workflows with inconsistent resource needs, dynamic infrastructure can adapt to changing requirements.
4. Cloud-native deployments: When working with cloud providers that offer serverless or auto-scaling options.

​

Prerequisites

1. A running Prefect server or Prefect Cloud workspace: You can either run a Prefect server locally or use a Prefect Cloud workspace. To start a local server, run prefect server start. To use Prefect Cloud, sign up for an account at app.prefect.cloud and follow the Connect to Prefect Cloud guide.
2. A Prefect flow: You should have a flow defined in your Python script. If you haven’t created a flow yet, refer to the Write Workflows guide.
3. A work pool: You need a work pool to manage the infrastructure for running your flow. If you haven’t created a work pool, you can do so through the Prefect UI or using the Prefect CLI. For more information, see the Work Pools guide. For examples in this guide, we’ll use a Docker work pool created by running:
   Copy

   prefect work-pool create --type docker my-work-pool
   

4. Docker: Docker will be used to build and store the image containing your flow code. You can download and install Docker from the official Docker website. If you don’t want to use Docker, you can see other options in the Use remote code storage section.
5. (Optional) A Docker registry: While not strictly necessary for local development, having an account with a Docker registry (such as Docker Hub) is recommended for storing and sharing your Docker images.

​

Deploy a flow with flow.deploy

from prefect import flow


@flow(log_prints=True)
def my_flow(name: str = "world"):
    print(f"Hello, {name}!")

from prefect import flow


@flow(log_prints=True)
def my_flow(name: str = "world"):
    print(f"Hello, {name}!")


if __name__ == "__main__":
    my_flow.deploy(
        name="my-deployment",
        work_pool_name="my-work-pool",
        image="my-registry.com/my-docker-image:my-tag",
        push=False # switch to True to push to your image registry
    )

python example.py

​

Trigger a run

prefect worker start --pool my-work-pool

prefect deployment run 'my-flow/my-deployment'

​

Deploy with a schedule

• interval Defines the interval at which the flow should run. Accepts an integer or float value representing the number of seconds between runs or a datetime.timedelta object.

  Show Example

  interval.py

  Copy

  from datetime import timedelta
  from prefect import flow
  
  
  @flow(log_prints=True)
  def my_flow(name: str = "world"):
      print(f"Hello, {name}!")
  
  
  if __name__ == "__main__":
      my_flow.deploy(
          name="my-deployment",
          work_pool_name="my-work-pool",
          image="my-registry.com/my-docker-image:my-tag",
          push=False,
          # Run once a minute
          interval=timedelta(minutes=1)
      )
  

• cron Defines when a flow should run using a cron string.

  Show Example

  cron.py

  Copy

  from prefect import flow
  
  
  @flow(log_prints=True)
  def my_flow(name: str = "world"):
      print(f"Hello, {name}!")
  
  
  if __name__ == "__main__":
      my_flow.deploy(
          name="my-deployment",
          work_pool_name="my-work-pool",
          image="my-registry.com/my-docker-image:my-tag",
          push=False,
          # Run once a day at midnight
          cron="0 0 * * *"
      )
  

• rrule Defines a complex schedule using an rrule string.

  Show Example

  rrule.py

  Copy

  from prefect import flow
  
  
  @flow(log_prints=True)
  def my_flow(name: str = "world"):
      print(f"Hello, {name}!")
  
  
  if __name__ == "__main__":
      my_flow.deploy(
          name="my-deployment",
          work_pool_name="my-work-pool",
          image="my-registry.com/my-docker-image:my-tag",
          push=False,
          # Run every weekday at 9 AM
          rrule="FREQ=WEEKLY;BYDAY=MO,TU,WE,TH,FR;BYHOUR=9;BYMINUTE=0"
      )
  

• schedules Defines multiple schedules for a deployment. This option provides flexibility for:
  • Setting up various recurring schedules
  • Implementing complex scheduling logic
  • Applying timezone offsets to schedules

  Show Example

  schedules.py

  Copy

  from datetime import datetime, timedelta
  from prefect import flow
  from prefect.schedules import Interval
  
  
  @flow(log_prints=True)
  def my_flow(name: str = "world"):
      print(f"Hello, {name}!")
  
  
  if __name__ == "__main__":
      my_flow.deploy(
          name="my-deployment",
          work_pool_name="my-work-pool",
          image="my-registry.com/my-docker-image:my-tag",
          push=False,
          # Run every 10 minutes starting from January 1, 2023
          # at 00:00 Central Time
          schedules=[
              Interval(
                  timedelta(minutes=10),
                  anchor_date=datetime(2023, 1, 1, 0, 0),
                  timezone="America/Chicago"
              )
          ]
      )
  

  Learn more about schedules here.

​

Use remote code storage

1. Faster iterations: You can update your flow code without rebuilding Docker images.
2. Reduced storage requirements: You don’t need to store large Docker images for each code version.
3. Flexibility: You can use different storage backends based on your needs and infrastructure.

1. Your code is already there.
2. You can deploy to multiple environments via branches and tags.
3. You can roll back to previous versions of your flows.

from prefect import flow


if __name__ == "__main__":
    flow.from_source(
        source="https://github.com/username/repository.git",
        entrypoint="path/to/your/flow.py:your_flow_function"
    ).deploy(
        name="my-deployment",
        work_pool_name="my-work-pool",
    )

• Git repositories
• S3 buckets (using the s3:// scheme)
• Google Cloud Storage buckets (using the gs:// scheme)
• Azure Blob Storage (using the az:// scheme)

​

Set default parameters

from prefect import flow


@flow
def my_flow(name: str = "world"):
    print(f"Hello, {name}!")


if __name__ == "__main__":
    my_flow.deploy(
        name="my-deployment",
        work_pool_name="my-work-pool",
        # Will print "Hello, Marvin!" by default instead of "Hello, world!"
        parameters={"name": "Marvin"},
        image="my-registry.com/my-docker-image:my-tag",
        push=False,
    )

​

Set job variables

import os
from prefect import flow


@flow
def my_flow():
    name = os.getenv("NAME", "world")
    print(f"Hello, {name}!")


if __name__ == "__main__":
    my_flow.deploy(
        name="my-deployment",
        work_pool_name="my-work-pool",
        # Will print "Hello, Marvin!" by default instead of "Hello, world!"
        job_variables={"env": {"NAME": "Marvin"}},
        image="my-registry.com/my-docker-image:my-tag",
        push=False,
    )

​

Deploy multiple flows

from prefect import flow, deploy


@flow
def my_flow_1():
    print("I'm number one!")


@flow
def my_flow_2():
    print("Always second...")


if __name__ == "__main__":
    deploy(
        # Use the `to_deployment` method to specify configuration 
        #specific to each deployment
        my_flow_1.to_deployment("my-deployment-1"),
        my_flow_2.to_deployment("my-deployment-2"),

        # Specify shared configuration for both deployments
        image="my-registry.com/my-docker-image:my-tag",
        push=False,
        work_pool_name="my-work-pool",  
    )

• Saves time and resources by avoiding redundant image builds.
• Simplifies management by maintaining a single image for multiple flows.

​

Further reading

• Work Pools
• Store Flow Code
• Customize Infrastructure
• Schedules
• Write Workflows