Ryan Kladar
07/09/2025, 3:21 PMserve()
to deploy multiple flows?Marvin
07/09/2025, 3:23 PMMarvin
07/09/2025, 3:24 PMserve()
in Prefect 3.x:
1. Basic Multiple Flow Deployment
python
from prefect import flow, serve
@flow
def flow_1():
print("Running flow 1")
@flow
def flow_2():
print("Running flow 2")
if __name__ == "__main__":
serve(
flow_1,
flow_2,
)
2. Multiple Flows with Different Schedules
python
from prefect import flow, serve
from prefect.deployments import Deployment
@flow
def flow_1():
print("Running flow 1")
@flow
def flow_2():
print("Running flow 2")
if __name__ == "__main__":
# Create deployments with different schedules
deployment_1 = flow_1.to_deployment(
name="flow-1-deployment",
interval=60 # Runs every 60 seconds
)
deployment_2 = flow_2.to_deployment(
name="flow-2-deployment",
cron="0 * * * *" # Runs hourly
)
serve(
deployment_1,
deployment_2
)
Key points to remember:
- The serve()
command is a long-running process that keeps your deployments active
- You can mix and match flows and deployments in the same serve()
call
- Each deployment can have its own schedule and configuration
- The script must remain running for the deployments to continue working
For more details, check out:
- Schedule your code quickly with flow.serve
- Run flows in local processes
Would you like me to explain any specific aspect of serving multiple flows in more detail?Ryan Kladar
07/09/2025, 3:44 PMMarvin
07/09/2025, 3:44 PMMarvin
07/09/2025, 3:46 PMpython
from prefect import flow
from prefect.deployments import deploy
@flow
def flow_1(name: str = "default"):
print(f"Running flow 1 with {name}")
@flow
def flow_2(count: int = 1):
print(f"Running flow 2 with {count}")
if __name__ == "__main__":
# Create and deploy multiple flows at once
deploy(
flow_1.to_deployment(name="flow-1", tags=["demo"]),
flow_2.to_deployment(name="flow-2", tags=["demo"]),
work_pool_name="my-work-pool" # Specify your work pool
)
2. Advanced Deployment with Docker and Different Configurations
python
from prefect import flow
from prefect.deployments import deploy
from prefect.deployments.runner import RunnerDeployment
from prefect.runner.storage import DeploymentImage
# Define your flows
@flow
def flow_1():
print("Running flow 1")
@flow
def flow_2():
print("Running flow 2")
@flow
def flow_3():
print("Running flow 3")
if __name__ == "__main__":
# Create a shared image configuration
image = DeploymentImage(
name="my-flows-image",
tag="latest",
dockerfile="Dockerfile" # Path to your Dockerfile
)
# Create deployments with different configurations
deployments = [
flow_1.to_deployment(
name="scheduled-flow-1",
interval=300, # Run every 5 minutes
image=image
),
flow_2.to_deployment(
name="cron-flow-2",
cron="0 * * * *", # Run hourly
image=image
),
flow_3.to_deployment(
name="manual-flow-3",
image=image
)
]
# Deploy all flows at once
deploy(
*deployments,
work_pool_name="my-docker-pool",
build=True, # Build the Docker image
push=True # Push the image to registry
)
Key points to remember:
1. Work Pool Configuration:
- Specify your work pool name using work_pool_name
- Make sure the work pool exists before deploying
2. Deployment Options:
- Each flow can have its own schedule (interval, cron)
- You can add tags, parameters, and other metadata
- All flows can share the same Docker image if needed
3. Docker Settings (if using containers):
- Set build=True
to build Docker images
- Use push=True
if pushing to a registry
- All flows can share the same DeploymentImage
configuration
4. Best Practices:
- Group related flows in the same script
- Use consistent naming conventions
- Add relevant tags for better organization
- Consider using environment variables for sensitive configuration
For more details, check out the deployment via Python documentation.
Would you like me to explain any specific aspect of deploying multiple flows in more detail?Ryan Kladar
07/09/2025, 4:46 PMMarvin
07/09/2025, 4:46 PMMarvin
07/09/2025, 4:47 PMpython
from prefect.blocks.system import Secret
# Load the existing secret by its name
github_token = Secret.load("github-token")
# Access the stored value
token = github_token.get()
A few important points:
- Replace "github-token"
with whatever name you used when creating the Secret block
- The .get()
method returns the actual secret value
- If the Secret doesn't exist, load()
will raise a ValueError
If you haven't created the Secret block yet, you can do so either through the UI or programmatically:
python
from prefect.blocks.system import Secret
# Create and save a new secret
github_secret = Secret(value="your-github-token")
github_secret.save(name="github-token", overwrite=True)
For more information about managing secrets, check out the documentation on storing secrets.