https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> how to get deployment_id using prefec...
# ask-marvin
z
@Marvin how to get deployment_id using prefect library imported
m
✅ thought for 20.0 seconds
Here's how you can get the deployment ID in your code:
Copy code
python
from prefect.runtime.deployment import id as deployment_id

# Access the deployment ID
current_deployment_id = deployment_id
A few important notes: 1. This only works when your flow is running as part of a deployment. It won't work if you're just running a flow locally. 2. The deployment ID is only available during runtime within the deployment's execution context. If you need to get deployment IDs outside of a flow run (e.g., for administrative purposes), you can use the Prefect REST API client:
Copy code
python
from prefect.client import get_client

async def get_deployment_info(deployment_name: str):
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(deployment_name)
        return deployment.id
You can find more details about deployment-related APIs in the Prefect documentation.
z
@Marvin I run this example auto_to_create = Automation( name="woodchonk", trigger=EventTrigger( expect={"animal.walked"}, match={ "genus": "Marmota", "species": "monax", }, posture="Reactive", threshold=3, within=timedelta(seconds=10), ), actions=[CancelFlowRun()] ) created_automation = auto_to_create.create() and then I get this error Traceback (most recent call last): File "/Users/zvalutyte/Barbora_projects/prefect-v3/automations.py", line 31, in <module> aut = Automation( ^^^^^^^^^^^ File "/Users/zvalutyte/Barbora_projects/prefect-v3/.venv/lib/python3.12/site-packages/pydantic/main.py", line 253, in init validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ pydantic_core._pydantic_core.ValidationError: 16 validation errors for Automation actions.0.DoNothing Input should be a valid dictionary or instance of DoNothing [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.function-after[selected_deployment_requires_id(), RunDeployment] Input should be a valid dictionary or instance of RunDeployment [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.function-after[selected_deployment_requires_id(), PauseDeployment] Input should be a valid dictionary or instance of PauseDeployment [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.function-after[selected_deployment_requires_id(), ResumeDeployment] Input should be a valid dictionary or instance of ResumeDeployment [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.CancelFlowRun Input should be a valid dictionary or instance of CancelFlowRun [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.ChangeFlowRunState Input should be a valid dictionary or instance of ChangeFlowRunState [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.function-after[selected_work_queue_requires_id(), PauseWorkQueue] Input should be a valid dictionary or instance of PauseWorkQueue [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.function-after[selected_work_queue_requires_id(), ResumeWorkQueue] Input should be a valid dictionary or instance of ResumeWorkQueue [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.SendNotification Input should be a valid dictionary or instance of SendNotification [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.CallWebhook Input should be a valid dictionary or instance of CallWebhook [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.function-after[selected_automation_requires_id(), PauseAutomation] Input should be a valid dictionary or instance of PauseAutomation [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.function-after[selected_automation_requires_id(), ResumeAutomation] Input should be a valid dictionary or instance of ResumeAutomation [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.SuspendFlowRun Input should be a valid dictionary or instance of SuspendFlowRun [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.PauseWorkPool Input should be a valid dictionary or instance of PauseWorkPool [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.ResumeWorkPool Input should be a valid dictionary or instance of ResumeWorkPool [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type actions.0.DeclareIncident Input should be a valid dictionary or instance of DeclareIncident [type=model_type, input_value=RunDeployment(type='run-d...']}, job_variables=None), input_type=RunDeployment] For further information visit https://errors.pydantic.dev/2.11/v/model_type
m
Your message was too long, here's your message at the allowed limit: I run this example auto_to_create = Automation( name= woodchonk trigger=EventTrigger( expect={ animal walked } match={ genus Marmota species monax } posture= Reactive threshold=3 within=timedelta(seconds=10) ) actions=[CancelFlowRun()] ) created_automation = auto_to_create create() and then I get this error Traceback (most recent call last) File /Users/zvalutyte/Barbora_projects/prefect-v3/automations py line 31 in <module> aut = Automation( ^^^^^^^^^^^ File /Users/zvalutyte/Barbora_projects/prefect-v3/ venv/lib/python3 12/site-packages/pydantic/main py line 253 in init validated_self = self __pydantic_validator__ validate_python(data self_instance=self) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ pydantic_core _pydantic_core ValidationError 16 validation errors for Automation actions 0 DoNothing Input should be a valid dictionary or instance of DoNothing [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 function-after[selected_deployment_requires_id() RunDeployment] Input should be a valid dictionary or instance of RunDeployment [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 function-after[selected_deployment_requires_id() PauseDeployment] Input should be a valid dictionary or instance of PauseDeployment [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 function-after[selected_deployment_requires_id() ResumeDeployment] Input should be a valid dictionary or instance of ResumeDeployment [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 CancelFlowRun Input should be a valid dictionary or instance of CancelFlowRun [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 ChangeFlowRunState Input should be a valid dictionary or instance of ChangeFlowRunState [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 function-after[selected_work_queue_requires_id() PauseWorkQueue] Input should be a valid dictionary or instance of PauseWorkQueue [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 function-after[selected_work_queue_requires_id() ResumeWorkQueue] Input should be a valid dictionary or instance of ResumeWorkQueue [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 SendNotification Input should be a valid dictionary or instance of SendNotification [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 CallWebhook Input should be a valid dictionary or instance of CallWebhook [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 function-after[selected_automation_requires_id() PauseAutomation] Input should be a valid dictionary or instance of P
auseAutomation [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 function-after[selected_automation_requires_id() ResumeAutomation] Input should be a valid dictionary or instance of ResumeAutomation [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 SuspendFlowRun Input should be a valid dictionary or instance of SuspendFlowRun [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 PauseWorkPool Input should be a valid dictionary or instance of PauseWorkPool [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 ResumeWorkPool Input should be a valid dictionary or instance of ResumeWorkPool [type=model_type input_value=RunDeployment(type='run-d ']} job_variables=None) input_type=RunDeployment] For further information visit <https //errors pydantic dev/2 11/v/model_type> actions 0 DeclareIncident Input should be a valid dictionary or instance
z
@Marvin I run this example auto_to_create = Automation( name="woodchonk", trigger=EventTrigger( expect={"animal.walked"}, match={ "genus": "Marmota", "species": "monax", }, posture="Reactive", threshold=3, within=timedelta(seconds=10), ), actions=[CancelFlowRun()] ) created_automation = auto_to_create.create() and then I get this error Traceback (most recent call last): File "/Users/zvalutyte/Barbora_projects/prefect-v3/automations.py", line 31, in <module> aut = Automation( ^^^^^^^^^^^ File "/Users/zvalutyte/Barbora_projects/prefect-v3/.venv/lib/python3.12/site-packages/pydantic/main.py", line 253, in init validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ pydantic_core._pydantic_core.ValidationError: 16 validation errors for Automation
m
✅ thought for 49.0 seconds
Looking at your error and the research, I notice a few potential issues with the automation creation. Here's the correct way to create an automation with an EventTrigger and CancelFlowRun action:
Copy code
python
from prefect.server.schemas.automations import Automation
from prefect.server.schemas.events import EventTrigger
from prefect.server.schemas.actions import CancelFlowRun
from datetime import timedelta

auto_to_create = Automation(
    name="woodchonk",
    description="Cancel flow runs when specific animal events occur",  # Optional but recommended
    enabled=True,  # Optional, defaults to True
    trigger=EventTrigger(
        match_related={  # Use match_related instead of match for object properties
            "genus": "Marmota",
            "species": "monax",
        },
        event_names=["animal.walked"],  # Use event_names instead of expect
        after=None,  # Optional
        within_timedelta=timedelta(seconds=10),  # Use within_timedelta instead of within
        count_within_timedelta=3,  # Use count_within_timedelta instead of threshold
    ),
    actions=[CancelFlowRun()]
)

# Create the automation
created_automation = await auto_to_create.save()  # Note: This needs to be awaited
The key corrections from your original code are: 1. The 
expect
parameter should be 
event_names
and should be a list 2. Use 
match_related
instead of 
match
for object properties 3. 
within
should be 
within_timedelta
4. 
threshold
should be 
count_within_timedelta
5. The 
posture
parameter is not a valid field for EventTrigger 6. The creation method is 
save()
not 
create()
, and it needs to be awaited Also, make sure you're running this code in an async context. If you're running it in a synchronous context, you'll need to wrap it in an async function and use an async runner:
Copy code
python
async def create_automation():
    # ... automation creation code here ...
    return await auto_to_create.save()

# Then run it using:
import asyncio
created_automation = asyncio.run(create_automation())
This should resolve the validation errors you're seeing. The errors occurred because some of the field names and structures have changed in Prefect 3.x compared to earlier versions.
Open in Slack
PreviousNext
Powered by