There are several ways how you can accomplish that.
Run the agent directly in the virtual environment from which you want to execute the flow runs
Assuming that your agent is already running in the virtual environment that has all required dependencies. In that case, you don’t need to do anything else on your infrastructure block or deployment - the default Process block will already work out of the box.
Adjust the YAML file
Another option is to adjust the YAML deployment file before applying the deployment.
You can modify the command that is used for the flow run entrypoint to use Python in the virtual environment as follows:
Example flow
from prefect import task, flow
from prefect import get_run_logger
from typing import Any
@task
def say_hi(user_name: str, question: str, answer: Any) -> None:
logger = get_run_logger()
logger.info("Hello from Prefect, %s! 👋", user_name)
logger.info("The answer to the %s question is %s! 🤖", question, answer)
@flow
def parametrized(
user: str = "Marvin", question: str = "Ultimate", answer: Any = 42
) -> None:
say_hi(user, question, answer)
if __name__ == "__main__":
parametrized(user="World")
Corresponding deployment YAML manifest
Here, we assume a virtual environment is named dataplatform - adjust it to the name of your Conda environment:
###
### A complete description of a Prefect Deployment for flow 'parametrized'
###
name: dev
description: null
version: 77ecff5259bc026121213109401965ae
# The work queue that will handle this deployment's runs
work_queue_name: dev
tags: []
parameters: {}
schedule: null
infra_overrides: {}
###
### DO NOT EDIT BELOW THIS LINE
###
flow_name: parametrized
manifest_path: null
infrastructure:
type: process
env: {}
labels: {}
name: null
command:
- conda
- run
- -n
- dataplatform
- python
- -m
- prefect.engine
stream_output: true
working_dir: null
_block_document_id: fe9361e7-68d2-45d9-818d-9f3d77660ea1
_block_document_name: anonymous-e3b61f4f-c424-4d4e-b3c7-e3c311f77d6c
_is_anonymous: true
block_type_slug: process
_block_type_slug: process
storage: null
path: /Users/anna/repos/prefect-deployment-patterns/__create_deployment
entrypoint: flows/parametrized.py:parametrized
parameter_openapi_schema:
title: Parameters
type: object
properties:
user:
title: user
default: Marvin
type: string
question:
title: question
default: Ultimate
type: string
answer:
title: answer
default: 42
required: null
definitions: null
Note that this deployment replaces the default command (shown below) with a command that runs the same in a conda environment:
command:
- python
- -m
- prefect.engine
Adjust the infrastructure Block from Code or from the UI
From code
Here is how you can create a Process infra block that will run the flow in a conda environment named dataplatform:
from prefect.infrastructure import Process
process_block = Process(
["conda", "run", "-n", "conda_environment_name", "python", "-m", "prefect.engine"]
)
process_block.save("default", overwrite=True)
From the UI
Or you could do the same from the UI:
