https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P>, can I trigger deployments in prefect...
# ask-marvin
m
@Marvin, can I trigger deployments in prefect cloud using tags? E.g. prefect deployment run <tag>
m
I couldn't find a direct answer to your question about triggering Prefect Cloud deployments using tags in the provided documentation. However, from what I know, Prefect Cloud generally allows you to define and run deployments. If tagging refers to Docker image tags or similar identifiers within deployment specifications, it's plausible that you can trigger specific deployment runs based on those identifiers. It may be that this feature is not described directly in easily accessible documentation, and could require some additional setup or use of Prefect's CLI or API. To get information tailored to Prefect 2.x (which you mentioned), I recommend referring directly to the Prefect documentation on deployments or querying the GitHub repository for issues or discussions related to this topic. If you need assistance with this, please let me know, and I can search the GitHub issues for you.
m
@Marvin, what's the python code for a prefect task to, trigger another deployment and set a start time for that deployment?
m
It seems there may have been a slight mix-up, as the code example provided below describes a basic Prefect task for performing an addition and doesn't directly address how to trigger another deployment and set a start time for it. However, triggering a deployment and setting a start time require using Prefect's deployments and scheduling API. You would typically need to use the Prefect client within a task to interface with Prefect Cloud or Prefect Server. This way, you can programmatically create deployments and schedule runs. I'll provide you with an illustrative code snippet on how you might achieve this. Please note that this code is hypothetical and for illustrative purposes; it might require adjustments to fit your specific circumstances and the Prefect version you're using:
Copy code
python
from prefect import flow, task
from prefect.deployments import Deployment
from datetime import timedelta, datetime

@task
def trigger_deployment():
    # Assuming you have a deployment name or id
    deployment_id = "your-deployment-id"
    
    # Set a start time for the deployment (e.g., start in 1 hour)
    start_time = datetime.utcnow() + timedelta(hours=1)

    # Use the Deployment API to trigger a flow run at a specific time
    # Note: This API might differ in Prefect 2.0, consult the relevant documentation
    deployment = Deployment.load(deployment_id)
    deployment.create_flow_run(run_config={'start_time': start_time})
    
@flow
def my_flow():
    # This flow can be used to trigger another deployment
    trigger_deployment()

if __name__ == "__main__":
    # Run the flow
    my_flow()
Since this functionality depends on Prefect's API and the infrastructure you have set up, I recommend checking the official Prefect documentation for the most accurate and up-to-date usage. If you need further assistance with this, I can look up specific details or code examples for you.
m
@Marvin, how can I schedule a flow to run from another flow? My logic is as follows: Every day process X triggers a sync. This sync takes quite long, so I don't want to continuously check the status. Instead I want flow X to schedule process Y to start an hour later. Then, every hour, process Y triggers and checks whether the sync has completed. if so, it will stop scheduling new checks for itself.
m
To schedule a flow to run from another flow, you may create subflows or use scheduling features provided by Prefect. Here's a hypothetical approach that captures your logic: 1. Process X triggers a sync and schedules Process Y to start checking an hour later. 2. Process Y runs every hour to check if the sync has completed. 3. If the sync is complete, Process Y stops scheduling new runs. Here's an illustrative Python code snippet that demonstrates how you might implement this logic using Prefect:
Copy code
python
from prefect import flow, task
from prefect.tasks import wait_for_flow_run, create_flow_run
from datetime import datetime, timedelta

@task
def initiate_sync():
    # Replace this with your actual sync logic
    print("Starting Sync Process...")
    return "Sync Process Initiated"

@task
def check_sync_status():
    # Replace this with the actual check for sync completion
    return True  # Assuming sync is complete

@flow
def process_y():
    sync_complete = check_sync_status()
    if not sync_complete:
        # Schedule this flow again in an hour if the sync is not complete
        create_flow_run(flow_name="process_y", scheduled_start_time=datetime.utcnow() + timedelta(hours=1))

@flow
def process_x():
    # Initiate sync process
    initiate_sync()
    
    # Schedule process Y to start an hour later
    create_flow_run(flow_name="process_y", scheduled_start_time=datetime.utcnow() + timedelta(hours=1))
    
    # Wait for the sync to complete (if needed)
    # Example of waiting for a different flow run to complete:
    # flow_run_id = initiate_sync()
    # wait_for_flow_run(flow_run_id)

if __name__ == "__main__":
    # Start Process X
    process_x()
In the code above: - 
process_x
is the flow initiating the sync. - 
process_y
checks if the sync is complete and reschedules itself if the sync is not done. - We use 
create_flow_run
to schedule another flow run for 
process_y
. - 
scheduled_start_time
is set to current UTC time plus one hour. Please adjust the code to fit the actual logic of your sync check and how you wish to initiate the sync process. Remember that this illustrative code is generic; you need to replace placeholders with actual implementation details fitting your infrastructure and Prefect version-specific functionality. Consult the Prefect Docs for more information on composing flows and the Schedules Docs for scheduling flows.
m
@Marvin, thanks that's a good template. how can I pass a list into the new flow?
m
To pass a list or any other parameters into a new flow run, you can take advantage of the parameterization feature that Prefect flows support. Parameters allow you to run the same group of tasks with different data by simply passing different argument values to the flow. Here's an example that demonstrates how to modify the previous template to pass a list as a parameter to 
process_y
flow:
Copy code
python
from prefect import flow, task
from prefect.tasks import create_flow_run
from datetime import datetime, timedelta

@task
def initiate_sync():
    # Replace this with your actual sync logic
    print("Starting Sync Process...")
    return ["item1", "item2", "item3"]  # The sync process returns a list to pass to the next flow

@flow
def process_y(list_param):
    # list_param is used within this flow as needed, for example:
    print(f"Received list: {list_param}")
    # Check sync status and logic...

# Change the process_x flow to:
@flow
def process_x():
    # Initiate sync process
    sync_list = initiate_sync()
    
    # Assuming sync_list is available immediately and you want to pass it to process_y
    # Schedule process Y to start an hour later, and pass the list as a parameter
    create_flow_run(
        flow_name="process_y",  # Make sure to use the correct flow name
        parameters={"list_param": sync_list.result()},  # Pending the actual result
        scheduled_start_time=datetime.utcnow() + timedelta(hours=1)
    )

if __name__ == "__main__":
    # Start Process X
    process_x()
In this example, the 
initiate_sync
task generates a list and returns it to the 
process_x
flow, which then includes the list as a parameter, 
list_param
, to the new 
process_y
flow run. When using the 
create_flow_run
task, you specify the parameters using a dictionary. When you're using such a pattern, make sure that the task returning the list (initiate_sync in this example) completes before passing its result to 
create_flow_run
. Also, ensure that the flow and task definitions properly accept and handle the parameters being passed. Executions and passes of parameters may vary depending on the exact version and setup of Prefect that you are using. For the most accurate and detailed instructions, consult the Prefect documentation on flows and parameters.
m
@Marvin, how do i pass a tuple in prefect.yml as parameter?
m
Passing a tuple as a parameter within a 
prefect.yaml
file is not explicitly documented in the excerpts provided. However, considering 
prefect.yaml
follows the YAML format and syntax, you should be able to define a tuple (or any list-like structure) as a sequence. For instance:
Copy code
yaml
parameters:
  my_tuple: !!python/tuple [1, 2, 3]
In the above example, 
!!python/tuple
is used to explicitly declare a tuple in YAML, and 
[1, 2, 3]
defines the elements of the tuple. When Prefect reads this configuration, it should understand that 
my_tuple
is a tuple with the values 
(1, 2, 3)
. As documentation and handling of YAML structures can vary, please verify that this processing of tuples is supported by your version of Prefect. If this does not work or you need further guidance, you may need to transform the tuple into a list in the 
prefect.yaml
and then cast it back to a tuple within the flow if necessary. For detailed and specific instructions or syntax, always refer to the Prefect Documentation, especially since syntax and features may evolve over time. If you run into any trouble or need further assistance, do not hesitate to ask.
m
@Marvin, if i use .submit() for a concurrent task, can I pass a variable with the name for each task? If so, how should this look?
m
Prefect does not provide an explicit mention in the documentation about passing a variable with the name for each concurrent task submitted with 
.submit()
. However, you can achieve a similar outcome by defining your task with a parameter that includes the name and then submitting concurrent task runs with different names. Below is an example of how you can define a task that accepts a name parameter and submit multiple instances of the task concurrently, each with a different name:
Copy code
python
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@task
def my_task(task_name):
    print(f"Task {task_name} running.")

@flow(task_runner=ConcurrentTaskRunner())
def my_flow():
    names = ["alpha", "beta", "gamma"]  # List of names to pass to tasks
    tasks = [my_task.submit(task_name=name) for name in names]  # Submits tasks concurrently with different names

if __name__ == "__main__":
    my_flow()
In this example, we define a task called 
my_task
that expects a 
task_name
parameter. When creating the flow 
my_flow
, we loop through a list of names and submit 
my_task
with these names using list comprehension. The 
.submit()
method is called for each task, and 
ConcurrentTaskRunner()
is used to ensure that tasks run concurrently. The key here is to use the 
.submit()
method with different argument values (names) within the flow definition. Variables can be passed to the 
.submit()
method just as you would pass arguments to a function call. Please note that the exact implementation details may depend on the version of Prefect that you are using, and I encourage you to refer to the Prefect documentation on tasks for the most accurate and up-to-date information.
m
@Marvin, why can a flow not receive .submit() for concurrent task runs? How can I run two flow concurrently?
m
In Prefect, a flow itself is not designed to be submitted for concurrent execution using 
.submit()
because 
.submit()
is a method used for submitting individual tasks for concurrent execution within a flow. However, you can still run multiple flows concurrently, but the approach differs from running tasks concurrently within a flow. To run two flows concurrently, you can execute each flow in its own thread or process, or use asynchronous flow definitions. If you are using Python's asyncio library, you can define your flows with 
async def
and then use 
asyncio.gather
to run them concurrently. Here's an example of how you can run two asynchronous flows concurrently:
Copy code
python
import asyncio
from prefect import flow

@flow
async def flow_one():
    # Your logic for flow one goes here
    print("Running flow one...")
    await asyncio.sleep(1)  # Simulate an async operation
    print("Flow one done.")

@flow
async def flow_two():
    # Your logic for flow two goes here
    print("Running flow two...")
    await asyncio.sleep(1)  # Simulate an async operation
    print("Flow two done.")

if __name__ == "__main__":
    asyncio.run(asyncio.gather(flow_one(), flow_two()))
In this example, 
flow_one
and 
flow_two
are defined as async flows. They are then run concurrently using 
asyncio.gather
. This approach requires that your flows be compatible with asynchronous execution. For synchronous execution, you could use threading or multiprocessing modules to achieve concurrency. For the specifics of concurrent flow execution in Prefect, always consult the Prefect documentation. The implementation details and best practices may vary depending on the version of Prefect you're using.
m
@Marvin, Why is in the code below the 
determine_period
function not executed concurrently? They happen sequentially: @flow( flow_run_name="Customer & Relation Data Fetch", log_prints=True, task_runner=ConcurrentTaskRunner(), ) def fetch_data(apis): print("Retrieving changes from 1TIS API") for api in apis: api_name = api["name"] print("Starting " + api_name + " API") date_range = determine_period.submit(api, api_name) retrieve_changed_record_ids(date_range, api) print("Completed " + api_name + " API")
m
The 
determine_period
task within your 
fetch_data
flow is not executed concurrently because the execution is happening inside a 
for
loop, which iterates sequentially. When you use 
my_task.submit()
, the task is submitted to the task runner for execution, but if this submission is inside a synchronous for loop, the next iteration (and thus the next task submission) will not happen until the previous one is completed. In order to achieve concurrency with 
ConcurrentTaskRunner
, you would need to collect all the task futures (or coroutine objects) first and then await their completion outside of the for loop. If you are not using async/await syntax, you can use asyncio.gather with launched coroutine objects. For synchronous code, you can collect the futures in a list and then resolve them after the for loop. Here's an example of how you might achieve concurrency using the `ConcurrentTaskRunner`:
Copy code
python
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@task
def determine_period(api):
    # task code here
    pass

@task
def retrieve_changed_record_ids(date_range, api):
    # task code here
    pass

@flow(
    flow_run_name="Customer & Relation Data Fetch",
    log_prints=True,
    task_runner=ConcurrentTaskRunner(),
)
def fetch_data(apis):
    print("Retrieving changes from 1TIS API")
    task_futures = []
    for api in apis:
        api_name = api["name"]
        print("Starting " + api_name + " API")
        date_range_future = determine_period.submit(api, api_name)
        task_futures.append((date_range_future, api))
    
    # Now all the tasks have been submitted, let's loop and await their results
    for date_range_future, api in task_futures:
        date_range = date_range_future.result()  # This will await the future
        retrieve_changed_record_ids(date_range, api)
        print("Completed " + api_name + " API")

if __name__ == "__main__":
    fetch_data(apis=[{'name': 'API1'}, {'name': 'API2'}])
In the revised example, 
task_futures
collects the futures of all 
determine_period
tasks as they are submitted. After submitting all the tasks, we loop over the futures to get their results. This allows the 
ConcurrentTaskRunner
to run the tasks concurrently. The 
result()
method on a future is blocking and will wait until the future is resolved. Please note that this hypothetical code is a general example and may not directly work with your specific flow implementation without adjustments. Also, you should consult the latest Prefect documentation to ensure that your synchronous or asynchronous task definitions align with Prefect's best practices and capabilities.
m
Hi @Marvin, could you please help me setting up a Cloud Run V2 work pool using a GCP VPC?
m
To set up a Cloud Run V2 work pool using a GCP VPC, follow these steps based on the information provided: 1. Step 1: Create a service account in your Google Cloud Project that will be used to run the Cloud Run jobs. 2. Step 2: Once you have the service account, create a new work pool of type 
cloud-run
using the Prefect UI or CLI. The CLI command would look like this (make sure to replace 
<WORK-POOL-NAME>
with your desired name for the work pool):
Copy code
shell
   prefect work-pool create --type cloud-run <WORK-POOL-NAME>
3. After the work pool is created, go to the Prefect UI and edit the created work pool. There will be options to customize the base job template for the work pool, which influences the infrastructure configuration for flow runs submitted to the work pool. 4. Specify the region for the Cloud Run job within the work pool configuration. Make sure that the region you choose supports VPC Connector, so your Cloud Run services can communicate with your GCP VPC network resources. 5. Specify your service account, so the work pool uses it for Cloud Run jobs. 6. Depending on your requirements, you might need to edit additional settings related to VPC such as the VPC Connector settings to ensure the Cloud Run instances can communicate with your VPC network. Please note that while the excerpts from the documentation provide an overview, they do not contain detailed information on how to use VPC with Cloud Run V2. This level of detail would typically be available in GCP documentation or more specific instructions provided by Prefect. Thus, you may need to consult GCP documentation about VPC settings for Cloud Run. For up-to-date instructions and more detailed steps, please visit the Prefect GCP Worker Guide and be sure to follow the latest best practices and any specific instructions about integrating Cloud Run V2 work pools with GCP VPC.
m
Hi @Marvin, How can i use the api to run a deployment and override the workpool?
m
To run a deployment via the Prefect API and override the work pool, you need to utilize the Prefect REST API for creating flow runs and specify your desired work pool in the request. Here's a step-by-step guide to achieve this: 1. Get Deployment ID: First, make sure you have the deployment ID for the deployment you want to run. You can list deployments using Prefect CLI.
Copy code
sh
    prefect deployment ls
2. Create Flow Run with Override: Use the 
create_flow_run
REST API to trigger a deployment and override the work pool. Below is an example using Python with the 
requests
library.
Copy code
python
    import requests

    # Set your Prefect Cloud or Prefect Server API URL and API Key
    PREFECT_API_URL = "<https://api.prefect.io/api/accounts/<ACCOUNT-ID>/workspaces/<WORKSPACE-ID>>"
    PREFECT_API_KEY = "<YOUR-PREFECT-API-KEY>"

    # Define headers
    headers = {
        "Authorization": f"Bearer {PREFECT_API_KEY}",
        "Content-Type": "application/json",
    }

    # Deployment ID
    deployment_id = "<YOUR-DEPLOYMENT-ID>"

    # Define the payload with the work pool override
    payload = {
        "deployment_id": deployment_id,
        "parameters": {
            # Any parameters for your flow
        },
        "work_pool": {
            "name": "your-new-work-pool-name"
        }
    }

    # Trigger the flow run
    response = <http://requests.post|requests.post>(
        f"{PREFECT_API_URL}/flow_runs/",
        json=payload,
        headers=headers
    )

    # Check response
    if response.status_code == 201:
        print("Flow run created successfully")
    else:
        print(f"Failed to create flow run: {response.status_code}")
        print(response.text)
In this example: - Replace 
<ACCOUNT-ID>
, 
<WORKSPACE-ID>
, 
<YOUR-PREFECT-API-KEY>
, and 
<YOUR-DEPLOYMENT-ID>
with your actual Prefect Cloud or Prefect Server credentials and deployment ID. - The URL 
<https://api.prefect.io/api/>...
should be adjusted based on whether you're using Prefect Cloud or an on-premise Prefect Server. - The payload includes a 
"work_pool"
field to specify the work pool to be used for this run. 3. Additional Settings: Ensure your Prefect YAML configuration or equivalent deployment settings support the dynamic override of job variables and work pools as per your infrastructure requirements. For the latest and more detailed API specifications, refer to the Prefect API documentation.
m
@Marvin,i am trying to pass the image as a variable when calling the 
Create Flow Run
api. I am passing the following body: { "state": { "type": "SCHEDULED" }, "tags": ["development"], "job_variables": {"image":"eu.gcr.io/abc/prefect/prefect-pricing:development"} } And the template of the work pool is as follows: "image": { "type": "string", "title": "Image Name", "default": "{{ env('image') }}", "description": "The image to use for the Cloud Run job. If not provided the latest Prefect image will be used." }, Yet i get the error : "returned "template.template.containers[0].image: Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or docker.io but obtained {{ env('image') }}". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'template.template.containers[0].image', 'description': "Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or docker.io but obtained {{ env('image') }}"}]}]">" Why is the variable not picked up?
m
Your message was too long by 87 tokens - please shorten it and try again. For reference, here's your message at the allowed limit:
,i am trying to pass the image as a variable when calling the 
Create Flow Run
api. I am passing the following body:
{ "state": { "type": "SCHEDULED" }, "tags": ["development"], "job_variables": {"image":"eu.gcr.io/abc/prefect/prefect-pricing:development"} } And the template of the work pool is as follows: "image": { "type": "string", "title": "Image Name", "default": "{{ env('image') }}", "description": "The image to use for the Cloud Run job. If not provided the latest Prefect image will be used." }, Yet i get the error : "returned "template.template.containers[0].image: Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or docker.io but obtained {{ env('image') }}". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'template.template.containers[0].image',
m
@Marvin,i am trying to pass the image as a variable when calling the 
Create Flow Run
api. I am passing the following body: { "state": { "type": "SCHEDULED" }, "tags": ["development"], "job_variables": {"image":"eu.gcr.io/abc/prefect/prefect-pricing:development"} } And the template of the work pool is as follows: "image": { "type": "string", "title": "Image Name", "default": "{{ env('image') }}", "description": "The image to use for the Cloud Run job. If not provided the latest Prefect image will be used." }, Expected an image path like [host/]repo-path[:tag and/or @digest], but obtained {{ env('image') }}". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'template.template.containers[0].image', 'description': "Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or docker.io but obtained {{ env('image') }}"}]}]">" Why is the variable not picked up?
m
Your message was too long by 26 tokens - please shorten it and try again. For reference, here's your message at the allowed limit:
,i am trying to pass the image as a variable when calling the 
Create Flow Run
api. I am passing the following body:
{ "state": { "type": "SCHEDULED" }, "tags": ["development"], "job_variables": {"image":"eu.gcr.io/abc/prefect/prefect-pricing:development"} } And the template of the work pool is as follows: "image": { "type": "string", "title": "Image Name", "default": "{{ env('image') }}", "description": "The image to use for the Cloud Run job. If not provided the latest Prefect image will be used." }, Expected an image path like [host/]repo-path[:tag and/or @digest], but obtained {{ env('image') }}". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'template.template.containers[0].image', 'description': "Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or <http://docker.io|
m
@Marvin,i am trying to pass the image as a variable when calling the 
Create Flow Run
api. I am passing the following body: { "state": { "type": "SCHEDULED" }, "tags": ["development"], "job_variables": {"image":"eu.gcr.io/abc/prefect/prefect-pricing:development"} } And the template of the work pool is as follows: "image": { "type": "string", "title": "Image Name", "default": "{{ env('image') }}", "description": "The image to use for the Cloud Run job." }, Expected an image path like [host/]repo-path[:tag and/or @digest], but obtained {{ env('image') }}". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'template.template.containers[0].image', 'description': "Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or docker.io but obtained {{ env('image') }}"}]}]">" Why is the variable not picked up?
m
Your message was too long by 14 tokens - please shorten it and try again. For reference, here's your message at the allowed limit:
,i am trying to pass the image as a variable when calling the 
Create Flow Run
api. I am passing the following body:
{ "state": { "type": "SCHEDULED" }, "tags": ["development"], "job_variables": {"image":"eu.gcr.io/abc/prefect/prefect-pricing:development"} } And the template of the work pool is as follows: "image": { "type": "string", "title": "Image Name", "default": "{{ env('image') }}", "description": "The image to use for the Cloud Run job." }, Expected an image path like [host/]repo-path[:tag and/or @digest], but obtained {{ env('image') }}". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'template.template.containers[0].image', 'description': "Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or docker.io but obtained {{ env('image') }}"}
m
@Marvin,i am trying to pass the image as a variable when calling the 
Create Flow Run
api. I am passing the following body: { "state": { "type": "SCHEDULED" }, "tags": ["development"], "job_variables": {"image":"eu.gcr.io/abc/prefect/prefect:development"} } And the template of the work pool is : "image": { "type": "string", "title": "Image Name", "default": "{{ env('image') }}", "description": "The image for the Cloud Run job." }, Expected an image path like [host/]repo-path[:tag and/or @digest], but obtained {{ env('image') }}". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'template.template.containers[0].image', 'description': "Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or docker.io but obtained {{ env('image') }}"}]}]">" Why is the variable not picked up?
m
Your message was too long by 6 tokens - please shorten it and try again. For reference, here's your message at the allowed limit:
,i am trying to pass the image as a variable when calling the 
Create Flow Run
api. I am passing the following body:
{ "state": { "type": "SCHEDULED" }, "tags": ["development"], "job_variables": {"image":"eu.gcr.io/abc/prefect/prefect:development"} } And the template of the work pool is : "image": { "type": "string", "title": "Image Name", "default": "{{ env('image') }}", "description": "The image for the Cloud Run job." }, Expected an image path like [host/]repo-path[:tag and/or @digest], but obtained {{ env('image') }}". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'template.template.containers[0].image', 'description': "Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or docker.io but obtained {{ env('image') }}"}]}]">" Why is
m
@Marvin,i am trying to pass the image as a variable when calling the 
Create Flow Run
api. I am passing the following body: { "state": { "type": "SCHEDULED" }, "tags": ["development"], "job_variables": {"image":"eu.gcr.io/abc/prefect/prefect:development"} } And the template of the work pool is : "image": { "type": "string", "title": "Image Name", "default": "{{ env('image') }}", "description": "The image for the Cloud Run job." }, Expected an image path like [host/]repo-path[:tag and/or @digest], but obtained {{ env('image') }}". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'template.template.containers[0].image', 'description': "Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or docker.io but obtained {{ env('image') }}"}]}]">" Why is the variable not picked up?
m
Your message was too long by 6 tokens - please shorten it and try again. For reference, here's your message at the allowed limit:
,i am trying to pass the image as a variable when calling the 
Create Flow Run
api. I am passing the following body:
{ "state": { "type": "SCHEDULED" }, "tags": ["development"], "job_variables": {"image":"eu.gcr.io/abc/prefect/prefect:development"} } And the template of the work pool is : "image": { "type": "string", "title": "Image Name", "default": "{{ env('image') }}", "description": "The image for the Cloud Run job." }, Expected an image path like [host/]repo-path[:tag and/or @digest], but obtained {{ env('image') }}". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'template.template.containers[0].image', 'description': "Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or docker.io but obtained {{ env('image') }}"}]}]">" Why is
m
@Marvin,i am trying to pass the image as a variable when calling the 
Create Flow Run
api. I am passing the following body: { "state": { "type": "SCHEDULED" }, "job_variables": {"image":"eu.gcr.io/abc/prefect/prefect:development"} } And the template of the work pool is : "image": { "type": "string", "title": "Image Name", "default": "{{ env('image') }}", "description": "The image for the Cloud Run job." }, Expected an image path like [host/]repo-path[:tag and/or @digest], but obtained {{ env('image') }}". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'template.template.containers[0].image', 'description': "Expected an image path like [host/]repo-path[:tag and/or @digest], where host is one of [region.]gcr.io, [region-]docker.pkg.dev or docker.io but obtained {{ env('image') }}"}]}]">" Why is the variable not picked up?
m
The issue you're encountering is related to how variables are referenced in your work pool template and how they are overridden via the API. It seems that the placeholder 
{{ env('image') }}
is not being correctly replaced by the provided value in 
job_variables
. Here's an updated approach to handling this, ensuring that the 
image
value is properly passed and picked up: 1. Update your work pool template: Ensure that your work pool template is correctly set up to accept 
image
as a variable. Specifically, you should define it directly in your job_variables without relying on 
env()
.
Copy code
yaml
   variables:
     image:
       type: string
       title: "Image Name"
       default: "<http://eu.gcr.io/abc/prefect/prefect:development|eu.gcr.io/abc/prefect/prefect:development>"
       description: "The image for the Cloud Run job."
2. Use the Prefect API correctly: When passing the 
image
variable in the 
Create Flow Run
API, do so directly via 
parameters.job_variables
. Here is a revised example of your API call:
Copy code
python
   import requests

   # Set your Prefect Cloud or Prefect Server API URL and API Key
   PREFECT_API_URL = "<https://api.prefect.io/api/accounts/<ACCOUNT-ID>/workspaces/<WORKSPACE-ID>>"
   PREFECT_API_KEY = "<YOUR-PREFECT-API-KEY>"

   # Define headers
   headers = {
       "Authorization": f"Bearer {PREFECT_API_KEY}",
       "Content-Type": "application/json",
   }

   # Deployment ID
   deployment_id = "<YOUR-DEPLOYMENT-ID>"

   # Define the payload with job_variables override
   payload = {
       "deployment_id": deployment_id,
       "parameters": {},
       "state": {
           "type": "SCHEDULED"
       },
       "job_variables": {
           "image": "<http://eu.gcr.io/abc/prefect/prefect:development|eu.gcr.io/abc/prefect/prefect:development>"
       }
   }

   # Trigger the flow run
   response = <http://requests.post|requests.post>(
       f"{PREFECT_API_URL}/flow_runs/",
       json=payload,
       headers=headers
   )

   # Check response
   if response.status_code == 201:
       print("Flow run created successfully")
   else:
       print(f"Failed to create flow run: {response.status_code}")
       print(response.text)
### Explanation: 1. Work Pool Template: Ensure that the template directly defines 
image
as a variable with a default value. This circumvents issues with potentially unsupported placeholders like 
{{ env('image') }}
. 2. API Request: Update the JSON payload in the API request, ensuring 
job_variables
references 
image
correctly. The API should now correctly capture and apply the 
image
variable for the flow run. This approach ensures that the variable 
image
is passed correctly when creating a flow run, and the work pool's job template picks it up without any placeholder interpretation issues. For the latest specifications on Prefect API and job variable overrides, refer to the Prefect API documentation.
m
@Marvin, i don't really see what is different in your example vs my use case. The body of the API call has not changed and the job template just changes the default value. how will my api job_variables override the default value in the workpool?
m
It appears the confusion arises from how job variables are overridden when creating a flow run. Let's clarify: ### Key Points: 1. Work Pool Template: - The work pool template defines the job variables and their defaults. - Job variables can be overridden on a per-deployment or per-flow run basis. 2. Overriding Job Variables: - Job variables must be correctly specified in the request body when creating a flow run. - Ensure the structure and naming in the API payload align with how job variables are defined in the work pool template. 3. Prefect API Call: - Ensure the correct fields are used to override job variables. Let’s rewrite the approach with more clarity: ### Work Pool Template Here’s how your work pool template should look, ensuring job variables can be dynamically overridden.
Copy code
yaml
variables:
  image:
    type: string
    title: "Image Name"
    default: "<http://eu.gcr.io/abc/prefect/prefect:development|eu.gcr.io/abc/prefect/prefect:development>"
    description: "The image for the Cloud Run job."
### Creating Flow Run with API Make sure you correctly reference the job variable 
image
in your request to override it.
Copy code
python
import requests

# Prefect API URL and API Key
PREFECT_API_URL = "<https://api.prefect.io/api/accounts/<ACCOUNT-ID>/workspaces/<WORKSPACE-ID>>"
PREFECT_API_KEY = "<YOUR-PREFECT-API-KEY>"

# Define headers
headers = {
    "Authorization": f"Bearer {PREFECT_API_KEY}",
    "Content-Type": "application/json",
}

# Deployment ID
deployment_id = "<YOUR-DEPLOYMENT-ID>"

# Define the payload to override the image variable
payload = {
    "deployment_id": deployment_id,
    "state": {
        "type": "SCHEDULED"
    },
    "parameters": {},
    "work_pool_name": "your-pool-name",  # Replace with your actual work pool name
    "job_variables": {"image": "<http://eu.gcr.io/abc/prefect/prefect:development|eu.gcr.io/abc/prefect/prefect:development>"}
}

# Trigger the flow run
response = <http://requests.post|requests.post>(
    f"{PREFECT_API_URL}/flow_runs/",
    json=payload,
    headers=headers
)

# Check response
if response.status_code == 201:
    print("Flow run created successfully")
else:
    print(f"Failed to create flow run: {response.status_code}")
    print(response.text)
### Important Details: - 
job_variables
: Ensure you are using the correct key and structure. In this case, the 
"job_variables"
key holds overrides. - Work Pool Name: Replace 
"your-pool-name"
with the actual name of the work pool where the job is to be run. ### Additional Notes Refer to the latest Prefect documentation for any changes in API structure, required fields, or additional options. The Overriding Job Variables Guide provides detailed usage and examples. If this still doesn't resolve your issue, verifying the exact formatting and payload structure against the Prefect API specs is crucial.
m
@Marvin, your template is in YAML, but my template is in JSON: Is it correct like this? { "variables": { "type": "object", "required": [ "credentials" ], ... "env": { "type": "object", "title": "Environment Variables", "description": "Environment variables to set when starting a flow run.", "additionalProperties": { "type": "string" } }, "image": { "type": "string", "title": "Image Name", "default": "eu.gcr.io/victoriacamperholidays/prefect/prefect-pricing:development", "description": "The image to use for the Cloud Run job. If not provided the latest Prefect image will be used." }, ...?
m
Yes, your JSON template for the work pool looks correct. JSON and YAML are effectively interchangeable formats; the key point is the structure and the adherence to the predefined schema. Here’s how your JSON template should look to define the 
image
job variable:
Copy code
json
{
  "variables": {
    "type": "object",
    "required": [
      "credentials"
    ],
    "env": {
      "type": "object",
      "title": "Environment Variables",
      "description": "Environment variables to set when starting a flow run.",
      "additionalProperties": {
        "type": "string"
      }
    },
    "image": {
      "type": "string",
      "title": "Image Name",
      "default": "<http://eu.gcr.io/victoriacamperholidays/prefect/prefect-pricing:development|eu.gcr.io/victoriacamperholidays/prefect/prefect-pricing:development>",
      "description": "The image to use for the Cloud Run job. If not provided the latest Prefect image will be used."
    }
  }
}
### Create Flow Run with API Ensure that your API call to create the flow run correctly references the job variable `image`:
Copy code
python
import requests

# Prefect API URL and API Key
PREFECT_API_URL = "<https://api.prefect.io/api/accounts/<ACCOUNT-ID>/workspaces/<WORKSPACE-ID>>"
PREFECT_API_KEY = "<YOUR-PREFECT-API-KEY>"

# Define headers
headers = {
    "Authorization": f"Bearer {PREFECT_API_KEY}",
    "Content-Type": "application/json",
}

# Deployment ID
deployment_id = "<YOUR-DEPLOYMENT-ID>"

# Define the payload to override the image variable
payload = {
    "deployment_id": deployment_id,
    "state": {
        "type": "SCHEDULED"
    },
    "parameters": {},
    "work_pool_name": "your-pool-name",  # Replace with your actual work pool name
    "job_variables": {
        "image": "<http://eu.gcr.io/victoriacamperholidays/prefect/prefect-pricing:development|eu.gcr.io/victoriacamperholidays/prefect/prefect-pricing:development>"
    }
}

# Trigger the flow run
response = <http://requests.post|requests.post>(
    f"{PREFECT_API_URL}/flow_runs/",
    json=payload,
    headers=headers
)

# Check response
if response.status_code == 201:
    print("Flow run created successfully")
else:
    print(f"Failed to create flow run: {response.status_code}")
    print(response.text)
### Explanation: - Work Pool Template: Ensure that 
image
is defined appropriately in the work pool template JSON. - API Request: The 
job_variables
key in the payload correctly overrides the 
image
defined in your template. Make sure that you correctly replace 
<ACCOUNT-ID>
, 
<WORKSPACE-ID>
, 
<YOUR-PREFECT-API-KEY>
, and 
<YOUR-DEPLOYMENT-ID>
with actual values from your Prefect environment. Additionally, make sure that 
"your-pool-name"
matches the name of the work pool where the job should be run. This approach ensures that the 
image
variable is properly overridden for the created flow run. If you still encounter issues, checking Prefect's documentation for any recent changes or detailed error handling would be beneficial. Feel free to ask if you need further assistance!
m
@Marvin, the job configuration in my job template is as follows: "job_configuration": { "env": "{{ env }}", ... "job_body": { "client": "prefect", "template": { "template": { "timeout": "{{ timeout }}", "containers": [ { "env": [], "args": "{{ args }}", "image": "{{ image }}", "command": "{{ command }}", "resources": { "limits": { "cpu": "{{ cpu }}", "memory": "{{ memory }}" } } } ], "maxRetries": "{{ max_retries }}", "serviceAccount": "{{ service_account_name }}" } }, "launchStage": "{{ launch_stage }}" }, "keep_job": "{{ keep_job }}", "credentials": "{{ credentials }}" } } Is this correct with what i stated before and how I am trying to override {{image}}?
m
From what you’ve shared, it looks like your job template configuration uses Jinja template placeholders (e.g., 
{{ image }}
) for various job variables. These placeholders should be correctly replaced with the provided job variables when triggering the flow run. Your approach should work if the API correctly references and overrides these job variables. Let’s combine everything and ensure that your work pool template and API call match correctly. ### Work Pool Template (JSON) Ensure the work pool template includes the 
image
job variable:
Copy code
json
{
  "variables": {
    "type": "object",
    "required": ["credentials"],
    "env": {
      "type": "object",
      "title": "Environment Variables",
      "description": "Environment variables to set when starting a flow run.",
      "additionalProperties": {
        "type": "string"
      }
    },
    "image": {
      "type": "string",
      "title": "Image Name",
      "default": "<http://eu.gcr.io/victoriacamperholidays/prefect/prefect-pricing:development|eu.gcr.io/victoriacamperholidays/prefect/prefect-pricing:development>",
      "description": "The image to use for the Cloud Run job. If not provided the latest Prefect image will be used."
    }
  },
  "job_configuration": {
    "env": "{{ env }}",
    "job_body": {
      "client": "prefect",
      "template": {
        "template": {
          "timeout": "{{ timeout }}",
          "containers": [
            {
              "env": [],
              "args": "{{ args }}",
              "image": "{{ image }}",
              "command": "{{ command }}",
              "resources": {
                "limits": {
                  "cpu": "{{ cpu }}",
                  "memory": "{{ memory }}"
                }
              }
            }
          ],
          "maxRetries": "{{ max_retries }}",
          "serviceAccount": "{{ service_account_name }}"
        }
      },
      "launchStage": "{{ launch_stage }}"
    },
    "keep_job": "{{ keep_job }}",
    "credentials": "{{ credentials }}"
  }
}
### API Call to Override 
image
Ensure your API call correctly overrides the 
image
variable in the job template:
Copy code
python
import requests

# Prefect API URL and API Key
PREFECT_API_URL = "<https://api.prefect.io/api/accounts/<ACCOUNT-ID>/workspaces/<WORKSPACE-ID>>"
PREFECT_API_KEY = "<YOUR-PREFECT-API-KEY>"

# Define headers
headers = {
    "Authorization": f"Bearer {PREFECT_API_KEY}",
    "Content-Type": "application/json",
}

# Deployment ID
deployment_id = "<YOUR-DEPLOYMENT-ID>"

# Define the payload to override the image variable
payload = {
    "deployment_id": deployment_id,
    "state": {
        "type": "SCHEDULED"
    },
    "parameters": {},
    "work_pool_name": "your-pool-name",  # Replace with your actual work pool name
    "job_variables": {
        "image": "<http://eu.gcr.io/victoriacamperholidays/prefect/prefect-pricing:development|eu.gcr.io/victoriacamperholidays/prefect/prefect-pricing:development>"
    }
}

# Trigger the flow run
response = <http://requests.post|requests.post>(
    f"{PREFECT_API_URL}/flow_runs/",
    json=payload,
    headers=headers
)

# Check response
if response.status_code == 201:
    print("Flow run created successfully")
else:
    print(f"Failed to create flow run: {response.status_code}")
    print(response.text)
### Explanation: 1. Work Pool Template: - The work pool template defines the job's base configuration. - Placeholders like 
{{ image }}
are defined in 
job_configuration
and are to be replaced at runtime. 2. API Call: - The API call overrides the 
image
variable in 
job_variables
with the desired value during the flow run creation. ### Troubleshooting: If the above setup still does not work, check the following: - Correct Values: Ensure that the placeholders (e.g., 
{{ image }}
) in your job configuration exactly match the keys used in 
job_variables
(case-sensitive). - Valid JSON: Validate your JSON configurations to ensure there are no syntax errors. - API Endpoint and Authentication: Ensure that the API endpoint and authentication details are correct and have necessary permissions.
- Logs and Errors: Review the response from the API call and Prefect logs to identify any errors or misconfigurations. If issues persist, please share any specific error messages or logs you encounter for further assistance.
m
@Marvin, i am requesting a deployment flow run using the following job variables: "job_variables": { "image": "eu.gcr.io/abc/prefect/prefect-pricing:production" } I can also see the image job variable in the UI, but the work pool does not utilize this variable in its configuration or job template. Why not?
@Marvin, i am requesting a deployment flow run using the following job variables: "job_variables": { "image": "eu.gcr.io/abc/prefect/prefect-pricing:production" } I can also see the image job variable in the UI, but the work pool does not utilize this variable in its configuration or job template. Why not?
m
If you're seeing the 
image
job variable set in the UI but the work pool is not utilizing it in its job configuration or job template, there are a few potential issues to consider: ### 1. Incorrect Placeholder Referencing The placeholder in JSON configuration must perfectly match the variable name used in the job_variables override. Ensure that the placeholder used within job configuration uses the correct Jinja templating placeholder. ### 2. Variable Context Make sure the variable is referenced properly in the context where it's needed. Sometimes the context in which the variable is substituted might be misplaced or misconfigured. ### Potential Issues and Solutions: ### Validate job configuration referencing Here's how an example of how the job configuration should properly reference the variable: #### Job Configuration in Template:
Copy code
json
{
  "job_configuration": {
    "env": "{{ env }}",
    "job_body": {
      "client": "prefect",
      "template": {
        "template": {
          "timeout": "{{ timeout }}",
          "containers": [
            {
              "env": [],
              "args": "{{ args }}",
              "image": "{{ image }}",
              "command": "{{ command }}",
              "resources": {
                "limits": {
                  "cpu": "{{ cpu }}",
                  "memory": "{{ memory }}"
                }
              }
            }
          ],
          "maxRetries": "{{ max_retries }}",
          "serviceAccount": "{{ service_account_name }}"
        }
      },
      "launchStage": "{{ launch_stage }}"
    },
    "keep_job": "{{ keep_job }}",
    "credentials": "{{ credentials }}"
  }
}
Ensure that 
{{ image }}
is placed exactly where it's needed. ### 3. Validate API request Make sure the 
job_variables
are correctly passed in the API request. The variable name should exactly match the one used in the job configuration. ### Example Proper API Usage:
Copy code
python
import requests

# Prefect API URL and API Key
PREFECT_API_URL = "<https://api.prefect.io/api/accounts/<ACCOUNT-ID>/workspaces/<WORKSPACE-ID>>"
PREFECT_API_KEY = "<YOUR-PREFECT-API-KEY>"

# Define headers
headers = {
    "Authorization": f"Bearer {PREFECT_API_KEY}",
    "Content-Type": "application/json",
}

# Deployment ID
deployment_id = "<YOUR-DEPLOYMENT-ID>"

# Define the payload to override the image variable
payload = {
    "deployment_id": deployment_id,
    "state": {
        "type": "SCHEDULED"
    },
    "parameters": {},
    "work_pool_name": "your-pool-name",  # Replace with your actual work pool name
    "job_variables": {
        "image": "<http://eu.gcr.io/abc/prefect/prefect-pricing:production|eu.gcr.io/abc/prefect/prefect-pricing:production>"
    }
}

# Trigger the flow run
response = <http://requests.post|requests.post>(
    f"{PREFECT_API_URL}/flow_runs/",
    json=payload,
    headers=headers
)

# Check response
if response.status_code == 201:
    print("Flow run created successfully")
else:
    print(f"Failed to create flow run: {response.status_code}")
    print(response.text)
### 4. Review Workflow Configuration Check Prefect UI and work pool settings to ensure the correct configuration. Here're some checks: - Verify in the Prefect UI that the 
image
variable is set and applied. - Ensure there are no conflicts or overrides that might prevent the new setting from being picked up. ### Final Checks: - Ensure consistency: The names must consistently match between template and API payload. - Debug output: Use 
print
statements or logging within your job configuration to debug what values are being substituted. - Check Prefect logs: Sometimes the error messages or logs in Prefect might provide insights into why the variables are not being set. If everything looks right but the variable is still not utilized, you might want to reach out to Prefect support or community for more specific assistance, as there may be subtleties in how variable substitution is managed.
m
@Marvin, can the 
create_flow_run
API call, which starts a flow run from a deployment override the 
job_configuration
values of a work pool using the 
job_variables
parameter in the API
m
Yes, the 
create_flow_run
API call can override 
job_configuration
values of a work pool using the 
job_variables
parameter. By providing 
job_variables
in your API request, you can dynamically set or override values specified in your work pool's job configuration, such as the Docker image. ### Key Points: 1. Job Variables in Work Pool Template: - Ensure your work pool template is set up to include placeholders for the job variables, such as 
{{ image }}
. 2. API Call: - Passing 
job_variables
in the API call allows you to override these placeholders. ### Example Work Pool Configuration (JSON): Ensure that the 
image
variable is properly set up in the job configuration:
Copy code
json
{
  "job_configuration": {
    "env": "{{ env }}",
    "job_body": {
      "client": "prefect",
      "template": {
        "template": {
          "timeout": "{{ timeout }}",
          "containers": [
            {
              "env": [],
              "args": "{{ args }}",
              "image": "{{ image }}",
              "command": "{{ command }}",
              "resources": {
                "limits": {
                  "cpu": "{{ cpu }}",
                  "memory": "{{ memory }}"
                }
              }
            }
          ],
          "maxRetries": "{{ max_retries }}",
          "serviceAccount": "{{ service_account_name }}"
        }
      },
      "launchStage": "{{ launch_stage }}"
    },
    "keep_job": "{{ keep_job }}",
    "credentials": "{{ credentials }}"
  }
}
### API Call with Job Variables Here's how to correctly call the 
create_flow_run
API and override the 
image
variable:
Copy code
python
import requests

# Prefect API URL and API Key
PREFECT_API_URL = "<https://api.prefect.io/api/accounts/<ACCOUNT-ID>/workspaces/<WORKSPACE-ID>>"
PREFECT_API_KEY = "<YOUR-PREFECT-API-KEY>"

# Define headers
headers = {
    "Authorization": f"Bearer {PREFECT_API_KEY}",
    "Content-Type": "application/json",
}

# Deployment ID
deployment_id = "<YOUR-DEPLOYMENT-ID>"

# Define the payload to override the image variable
payload = {
    "deployment_id": deployment_id,
    "state": {
        "type": "SCHEDULED"
    },
    "parameters": {},
    "work_pool_name": "your-pool-name",  # Replace with your actual work pool name
    "job_variables": {
        "image": "<http://eu.gcr.io/abc/prefect/prefect-pricing:production|eu.gcr.io/abc/prefect/prefect-pricing:production>"
    }
}

# Trigger the flow run
response = <http://requests.post|requests.post>(
    f"{PREFECT_API_URL}/flow_runs/",
    json=payload,
    headers=headers
)

# Check response
if response.status_code == 201:
    print("Flow run created successfully")
else:
    print(f"Failed to create flow run: {response.status_code}")
    print(response.text)
### Troubleshooting Steps: 1. Confirm Placeholder Matching: - Ensure that the placeholders (e.g., 
{{ image }}
) in your job configuration exactly match the keys used in 
job_variables
. 2. Variable Context: - Confirm that the context where the variable substitution takes place in the job configuration is correct and that the variables are being recognized and replaced properly. 3. Logs and API Response: - Check Prefect logs and the response from the API call for any signs indicating why the override might not be taking effect. ### Final Validation: After making these adjustments, the variable 
image
should be overridden by the value provided in the 
job_variables
parameter of your API call, and the job configuration in the work pool should utilize this value during the flow run execution. If the issue persists, checking Prefect community forums, support, or detailed documentation might provide additional insights.
2 Views
Open in Slack
PreviousNext
Powered by