Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/events/actions.py: 58%
732 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 02:04 +0000
1"""
2The actions consumer watches for actions that have been triggered by Automations
3and carries them out. Also includes the various concrete subtypes of Actions
4"""
6from __future__ import annotations
8import abc
9import asyncio
10import copy
11from base64 import b64encode
12from contextlib import asynccontextmanager
13from datetime import datetime, timedelta, timezone
14from typing import (
15 TYPE_CHECKING,
16 Any,
17 AsyncGenerator,
18 Awaitable,
19 Callable,
20 ClassVar,
21 Coroutine,
22 Dict,
23 List,
24 Literal,
25 MutableMapping,
26 Optional,
27 Tuple,
28 Type,
29 Union,
30 cast,
31)
32from uuid import UUID
34import jinja2
35import orjson
36from cachetools import TTLCache
37from httpx import Response
38from pydantic import (
39 Field,
40 PrivateAttr,
41 ValidationInfo,
42 field_validator,
43 model_validator,
44)
45from typing_extensions import Self, TypeAlias
47from prefect._internal.uuid7 import uuid7
48from prefect.blocks.abstract import NotificationBlock, NotificationError
49from prefect.blocks.core import Block
50from prefect.blocks.webhook import Webhook
51from prefect.logging import get_logger
52from prefect.server.events.clients import (
53 PrefectServerEventsAPIClient,
54 PrefectServerEventsClient,
55)
56from prefect.server.events.schemas.events import Event, RelatedResource, Resource
57from prefect.server.events.schemas.labelling import LabelDiver
58from prefect.server.schemas.actions import DeploymentFlowRunCreate, StateCreate
59from prefect.server.schemas.core import (
60 BlockDocument,
61 ConcurrencyLimitV2,
62 Flow,
63 TaskRun,
64 WorkPool,
65)
66from prefect.server.schemas.responses import (
67 DeploymentResponse,
68 FlowRunResponse,
69 OrchestrationResult,
70 StateAcceptDetails,
71 WorkQueueWithStatus,
72)
73from prefect.server.schemas.states import Scheduled, State, StateType, Suspended
74from prefect.server.utilities.http import should_redact_header
75from prefect.server.utilities.messaging import Message, MessageHandler
76from prefect.server.utilities.schemas import PrefectBaseModel
77from prefect.server.utilities.user_templates import (
78 TemplateRenderError,
79 TemplateSecurityError,
80 matching_types_in_templates,
81 maybe_template,
82 register_user_template_filters,
83 render_user_template,
84 validate_user_template,
85)
86from prefect.types import DateTime, NonNegativeTimeDelta, StrictVariableValue
87from prefect.types._datetime import now, parse_datetime
88from prefect.utilities.schema_tools.hydration import (
89 HydrationContext,
90 HydrationError,
91 Placeholder,
92 ValidJinja,
93 WorkspaceVariable,
94 hydrate,
95)
96from prefect.utilities.text import truncated_to
98if TYPE_CHECKING: # pragma: no cover 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true
99 import logging
101 from prefect.server.api.clients import OrchestrationClient
102 from prefect.server.events.schemas.automations import TriggeredAction
104 Parameters: TypeAlias = dict[str, Any | dict[str, Any] | list[Any | dict[str, Any]]]
106logger: "logging.Logger" = get_logger(__name__)
109class ActionFailed(Exception):
110 def __init__(self, reason: str):
111 self.reason = reason
114class Action(PrefectBaseModel, abc.ABC):
115 """An Action that may be performed when an Automation is triggered"""
117 type: str
119 # Captures any additional information about the result of the action we'd like to
120 # make available in the payload of the executed or failed events
121 _result_details: Dict[str, Any] = PrivateAttr(default_factory=dict)
122 _resulting_related_resources: List[RelatedResource] = PrivateAttr(
123 default_factory=list
124 )
126 @abc.abstractmethod
127 async def act(self, triggered_action: "TriggeredAction") -> None:
128 """Perform the requested Action"""
130 async def fail(self, triggered_action: "TriggeredAction", reason: str) -> None:
131 from prefect.server.events.schemas.automations import EventTrigger
133 automation = triggered_action.automation
134 action = triggered_action.action
135 action_index = triggered_action.action_index
137 automation_resource_id = f"prefect.automation.{automation.id}"
139 action_details = {
140 "action_index": action_index,
141 "action_type": action.type,
142 "invocation": str(triggered_action.id),
143 }
144 resource = Resource(
145 {
146 "prefect.resource.id": automation_resource_id,
147 "prefect.resource.name": automation.name,
148 "prefect.trigger-type": automation.trigger.type,
149 }
150 )
151 if isinstance(automation.trigger, EventTrigger): 151 ↛ 154line 151 didn't jump to line 154 because the condition on line 151 was always true
152 resource["prefect.posture"] = automation.trigger.posture
154 logger.warning(
155 "Action failed: %r",
156 reason,
157 extra={**self.logging_context(triggered_action)},
158 )
160 async with PrefectServerEventsClient() as events:
161 triggered_event_id = uuid7()
162 # Link to the triggering event if available and recent to establish causal chain.
163 # Only set follows if timing is tight (within 5 minutes) to avoid unnecessary
164 # waiting at CausalOrdering when events arrive >15 min after their follows event.
165 follows_id = None
166 if (
167 triggered_action.triggering_event
168 and triggered_action.triggering_event.occurred
169 ):
170 time_since_trigger = (
171 triggered_action.triggered
172 - triggered_action.triggering_event.occurred
173 )
174 TIGHT_TIMING = timedelta(minutes=5)
175 if abs(time_since_trigger) < TIGHT_TIMING:
176 follows_id = triggered_action.triggering_event.id
178 # Build related resources including automation.triggered and triggering event
179 related_resources = list(self._resulting_related_resources)
180 if triggered_action.automation_triggered_event_id: 180 ↛ 189line 180 didn't jump to line 189 because the condition on line 180 was always true
181 related_resources.append(
182 RelatedResource(
183 {
184 "prefect.resource.id": f"prefect.event.{triggered_action.automation_triggered_event_id}",
185 "prefect.resource.role": "automation-triggered-event",
186 }
187 )
188 )
189 if triggered_action.triggering_event:
190 related_resources.append(
191 RelatedResource(
192 {
193 "prefect.resource.id": f"prefect.event.{triggered_action.triggering_event.id}",
194 "prefect.resource.role": "triggering-event",
195 }
196 )
197 )
198 await events.emit(
199 Event(
200 occurred=triggered_action.triggered,
201 event="prefect.automation.action.triggered",
202 resource=resource,
203 related=related_resources,
204 payload=action_details,
205 id=triggered_event_id,
206 follows=follows_id,
207 )
208 )
209 await events.emit(
210 Event(
211 occurred=now("UTC"),
212 event="prefect.automation.action.failed",
213 resource=resource,
214 related=related_resources,
215 payload={
216 **action_details,
217 "reason": reason,
218 **self._result_details,
219 },
220 follows=triggered_event_id,
221 id=uuid7(),
222 )
223 )
225 async def succeed(self, triggered_action: "TriggeredAction") -> None:
226 from prefect.server.events.schemas.automations import EventTrigger
228 automation = triggered_action.automation
229 action = triggered_action.action
230 action_index = triggered_action.action_index
232 automation_resource_id = f"prefect.automation.{automation.id}"
234 action_details = {
235 "action_index": action_index,
236 "action_type": action.type,
237 "invocation": str(triggered_action.id),
238 }
239 resource = Resource(
240 {
241 "prefect.resource.id": automation_resource_id,
242 "prefect.resource.name": automation.name,
243 "prefect.trigger-type": automation.trigger.type,
244 }
245 )
246 if isinstance(automation.trigger, EventTrigger): 246 ↛ 249line 246 didn't jump to line 249 because the condition on line 246 was always true
247 resource["prefect.posture"] = automation.trigger.posture
249 async with PrefectServerEventsClient() as events:
250 triggered_event_id = uuid7()
251 # Link to the triggering event if available and recent to establish causal chain.
252 # Only set follows if timing is tight (within 5 minutes) to avoid unnecessary
253 # waiting at CausalOrdering when events arrive >15 min after their follows event.
254 follows_id = None
255 if (
256 triggered_action.triggering_event
257 and triggered_action.triggering_event.occurred
258 ):
259 time_since_trigger = (
260 triggered_action.triggered
261 - triggered_action.triggering_event.occurred
262 )
263 TIGHT_TIMING = timedelta(minutes=5)
264 if abs(time_since_trigger) < TIGHT_TIMING:
265 follows_id = triggered_action.triggering_event.id
267 # Build related resources including automation.triggered and triggering event
268 related_resources = list(self._resulting_related_resources)
269 if triggered_action.automation_triggered_event_id: 269 ↛ 278line 269 didn't jump to line 278 because the condition on line 269 was always true
270 related_resources.append(
271 RelatedResource(
272 {
273 "prefect.resource.id": f"prefect.event.{triggered_action.automation_triggered_event_id}",
274 "prefect.resource.role": "automation-triggered-event",
275 }
276 )
277 )
278 if triggered_action.triggering_event:
279 related_resources.append(
280 RelatedResource(
281 {
282 "prefect.resource.id": f"prefect.event.{triggered_action.triggering_event.id}",
283 "prefect.resource.role": "triggering-event",
284 }
285 )
286 )
287 await events.emit(
288 Event(
289 occurred=triggered_action.triggered,
290 event="prefect.automation.action.triggered",
291 resource=resource,
292 related=related_resources,
293 payload=action_details,
294 id=triggered_event_id,
295 follows=follows_id,
296 )
297 )
298 await events.emit(
299 Event(
300 occurred=now("UTC"),
301 event="prefect.automation.action.executed",
302 resource=resource,
303 related=related_resources,
304 payload={
305 **action_details,
306 **self._result_details,
307 },
308 id=uuid7(),
309 follows=triggered_event_id,
310 )
311 )
313 def logging_context(self, triggered_action: "TriggeredAction") -> Dict[str, Any]:
314 """Common logging context for all actions"""
315 return {
316 "automation": str(triggered_action.automation.id),
317 "action": self.model_dump(mode="json"),
318 "triggering_event": (
319 {
320 "id": triggered_action.triggering_event.id,
321 "event": triggered_action.triggering_event.event,
322 }
323 if triggered_action.triggering_event
324 else None
325 ),
326 "triggering_labels": triggered_action.triggering_labels,
327 }
330class DoNothing(Action):
331 """Do nothing when an Automation is triggered"""
333 type: Literal["do-nothing"] = "do-nothing"
335 async def act(self, triggered_action: "TriggeredAction") -> None:
336 logger.info(
337 "Doing nothing",
338 extra={**self.logging_context(triggered_action)},
339 )
342class EmitEventAction(Action):
343 async def act(self, triggered_action: "TriggeredAction") -> None:
344 event = await self.create_event(triggered_action)
346 self._result_details["emitted_event"] = str(event.id)
348 async with PrefectServerEventsClient() as events:
349 await events.emit(event)
351 @abc.abstractmethod
352 async def create_event(self, triggered_action: "TriggeredAction") -> "Event":
353 """Create an event from the TriggeredAction"""
356class ExternalDataAction(Action):
357 """Base class for Actions that require data from an external source such as
358 the Orchestration API"""
360 async def orchestration_client(
361 self, triggered_action: "TriggeredAction"
362 ) -> "OrchestrationClient":
363 from prefect.server.api.clients import OrchestrationClient
365 return OrchestrationClient(
366 additional_headers={
367 "Prefect-Automation-ID": str(triggered_action.automation.id),
368 "Prefect-Automation-Name": (
369 b64encode(triggered_action.automation.name.encode()).decode()
370 ),
371 },
372 )
374 async def events_api_client(
375 self, triggered_action: "TriggeredAction"
376 ) -> PrefectServerEventsAPIClient:
377 return PrefectServerEventsAPIClient(
378 additional_headers={
379 "Prefect-Automation-ID": str(triggered_action.automation.id),
380 "Prefect-Automation-Name": (
381 b64encode(triggered_action.automation.name.encode()).decode()
382 ),
383 },
384 )
386 def reason_from_response(self, response: Response) -> str:
387 error_detail = None
388 if response.status_code in {409, 422}: 388 ↛ 389line 388 didn't jump to line 389 because the condition on line 388 was never true
389 try:
390 error_detail = response.json().get("detail")
391 except Exception:
392 pass
394 if response.status_code == 422 or error_detail:
395 return f"Validation error occurred for {self.type!r}" + (
396 f" - {error_detail}" if error_detail else ""
397 )
398 else:
399 return f"Conflict (409) occurred for {self.type!r} - {error_detail or response.text!r}"
400 else:
401 return (
402 f"Unexpected status from {self.type!r} action: {response.status_code}"
403 )
406def _first_resource_of_kind(event: "Event", expected_kind: str) -> Optional["Resource"]:
407 for resource in event.involved_resources:
408 kind, _, _ = resource.id.rpartition(".")
409 if kind == expected_kind:
410 return resource
412 return None
415def _kind_and_id_from_resource(
416 resource: Resource,
417) -> tuple[str, UUID] | tuple[None, None]:
418 kind, _, id = resource.id.rpartition(".")
420 try:
421 return kind, UUID(id)
422 except ValueError:
423 pass
425 return None, None
428def _id_from_resource_id(resource_id: str, expected_kind: str) -> Optional[UUID]:
429 kind, _, id = resource_id.rpartition(".")
430 if kind == expected_kind: 430 ↛ 435line 430 didn't jump to line 435 because the condition on line 430 was always true
431 try:
432 return UUID(id)
433 except ValueError:
434 pass
435 return None
438def _id_of_first_resource_of_kind(event: "Event", expected_kind: str) -> Optional[UUID]:
439 resource = _first_resource_of_kind(event, expected_kind)
440 if resource:
441 if id := _id_from_resource_id(resource.id, expected_kind): 441 ↛ 443line 441 didn't jump to line 443 because the condition on line 441 was always true
442 return id
443 return None
446WorkspaceVariables: TypeAlias = Dict[str, StrictVariableValue]
447TemplateContextObject: TypeAlias = Union[PrefectBaseModel, WorkspaceVariables, None]
450class JinjaTemplateAction(ExternalDataAction):
451 """Base class for Actions that use Jinja templates supplied by the user and
452 are rendered with a context containing data from the triggered action,
453 and the orchestration API."""
455 _object_cache: Dict[str, TemplateContextObject] = PrivateAttr(default_factory=dict)
457 _registered_filters: ClassVar[bool] = False
459 @classmethod
460 def _register_filters_if_needed(cls) -> None:
461 if not cls._registered_filters:
462 # Register our event-related filters
463 from prefect.server.events.jinja_filters import all_filters
465 register_user_template_filters(all_filters)
466 cls._registered_filters = True
468 @classmethod
469 def validate_template(cls, template: str, field_name: str) -> str:
470 cls._register_filters_if_needed()
472 try:
473 validate_user_template(template)
474 except (jinja2.exceptions.TemplateSyntaxError, TemplateSecurityError) as exc:
475 raise ValueError(f"{field_name!r} is not a valid template: {exc}")
477 return template
479 @classmethod
480 def templates_in_dictionary(
481 cls, dict_: dict[Any, Any | dict[Any, Any]]
482 ) -> list[tuple[dict[Any, Any], dict[Any, str]]]:
483 to_traverse: list[dict[Any, Any]] = []
484 templates_at_layer: dict[Any, str] = {}
485 for key, value in dict_.items():
486 if isinstance(value, str) and maybe_template(value):
487 templates_at_layer[key] = value
488 elif isinstance(value, dict):
489 to_traverse.append(value)
491 templates: list[tuple[dict[Any, Any], dict[Any, str]]] = []
493 if templates_at_layer:
494 templates.append((dict_, templates_at_layer))
496 for item in to_traverse:
497 templates += cls.templates_in_dictionary(item)
499 return templates
501 def instantiate_object(
502 self,
503 model: Type[PrefectBaseModel],
504 data: Dict[str, Any],
505 triggered_action: "TriggeredAction",
506 resource: Optional["Resource"] = None,
507 ) -> PrefectBaseModel:
508 object = model.model_validate(data)
510 if isinstance(object, FlowRunResponse) or isinstance(object, TaskRun):
511 # The flow/task run was fetched from the API, but between when its
512 # state changed and now it's possible that the state in the API has
513 # changed again from what's contained in the event. Use the event's
514 # data to rebuild the state object and attach it to the object
515 # received from the API.
516 # https://github.com/PrefectHQ/nebula/issues/3310
517 state_fields = [
518 "prefect.state-message",
519 "prefect.state-name",
520 "prefect.state-timestamp",
521 "prefect.state-type",
522 ]
524 if resource and all(field in resource for field in state_fields):
525 try:
526 timestamp = parse_datetime(resource["prefect.state-timestamp"])
527 if TYPE_CHECKING:
528 assert isinstance(timestamp, DateTime)
529 object.state = State(
530 message=resource["prefect.state-message"],
531 name=resource["prefect.state-name"],
532 timestamp=timestamp,
533 type=StateType(resource["prefect.state-type"]),
534 )
535 except Exception:
536 logger.exception(
537 "Failed to parse state from event resource",
538 extra={
539 **self.logging_context(triggered_action),
540 },
541 )
543 return object
545 async def _get_object_from_prefect_api(
546 self,
547 orchestration_client: "OrchestrationClient",
548 triggered_action: "TriggeredAction",
549 resource: Optional["Resource"],
550 ) -> Optional[PrefectBaseModel]:
551 if not resource:
552 return None
554 kind, obj_id = _kind_and_id_from_resource(resource)
556 if not obj_id:
557 return None
559 kind_to_model_and_methods: Dict[
560 str,
561 Tuple[
562 Type[PrefectBaseModel],
563 List[Callable[..., Coroutine[Any, Any, Response]]],
564 ],
565 ] = {
566 "prefect.deployment": (
567 DeploymentResponse,
568 [orchestration_client.read_deployment_raw],
569 ),
570 "prefect.flow": (Flow, [orchestration_client.read_flow_raw]),
571 "prefect.flow-run": (
572 FlowRunResponse,
573 [orchestration_client.read_flow_run_raw],
574 ),
575 "prefect.task-run": (TaskRun, [orchestration_client.read_task_run_raw]),
576 "prefect.work-pool": (
577 WorkPool,
578 [orchestration_client.read_work_pool_raw],
579 ),
580 "prefect.work-queue": (
581 WorkQueueWithStatus,
582 [
583 orchestration_client.read_work_queue_raw,
584 orchestration_client.read_work_queue_status_raw,
585 ],
586 ),
587 "prefect.concurrency-limit": (
588 ConcurrencyLimitV2,
589 [orchestration_client.read_concurrency_limit_v2_raw],
590 ),
591 }
593 if kind not in kind_to_model_and_methods:
594 return None
596 model, client_methods = kind_to_model_and_methods[kind]
598 responses = await asyncio.gather(
599 *[client_method(obj_id) for client_method in client_methods]
600 )
602 if any(response.status_code >= 300 for response in responses):
603 return None
605 combined_response: dict[Any, Any] = {}
606 for response in responses:
607 data: Any | list[Any] = response.json()
609 # Sometimes we have to call filter endpoints that return a list of 0..1
610 if isinstance(data, list):
611 if len(data) == 0:
612 return None
613 data = data[0]
615 combined_response.update(data)
617 return self.instantiate_object(
618 model, combined_response, triggered_action, resource=resource
619 )
621 async def _relevant_native_objects(
622 self, templates: List[str], triggered_action: "TriggeredAction"
623 ) -> Dict[str, TemplateContextObject]:
624 if not triggered_action.triggering_event: 624 ↛ 625line 624 didn't jump to line 625 because the condition on line 624 was never true
625 return {}
627 orchestration_types = {
628 "deployment",
629 "flow",
630 "flow_run",
631 "task_run",
632 "work_pool",
633 "work_queue",
634 "concurrency_limit",
635 }
636 special_types = {"variables"}
638 types = matching_types_in_templates(
639 templates, types=orchestration_types | special_types
640 )
641 if not types: 641 ↛ 644line 641 didn't jump to line 644 because the condition on line 641 was always true
642 return {}
644 needed_types = list(set(types) - set(self._object_cache.keys()))
646 async with await self.orchestration_client(triggered_action) as orchestration:
647 calls: List[Awaitable[TemplateContextObject]] = []
648 for type_ in needed_types:
649 if type_ in orchestration_types:
650 calls.append(
651 self._get_object_from_prefect_api(
652 orchestration,
653 triggered_action,
654 _first_resource_of_kind(
655 triggered_action.triggering_event,
656 f"prefect.{type_.replace('_', '-')}",
657 ),
658 )
659 )
660 elif type_ == "variables":
661 calls.append(orchestration.read_workspace_variables())
663 objects = await asyncio.gather(*calls)
665 self._object_cache.update(dict(zip(needed_types, objects)))
667 return self._object_cache
669 async def _template_context(
670 self, templates: List[str], triggered_action: "TriggeredAction"
671 ) -> dict[str, Any]:
672 context: dict[str, Any] = {
673 "automation": triggered_action.automation,
674 "event": triggered_action.triggering_event,
675 "labels": LabelDiver(triggered_action.triggering_labels),
676 "firing": triggered_action.firing,
677 "firings": triggered_action.all_firings(),
678 "events": triggered_action.all_events(),
679 }
680 context.update(await self._relevant_native_objects(templates, triggered_action))
681 return context
683 async def _render(
684 self, templates: List[str], triggered_action: "TriggeredAction"
685 ) -> List[str]:
686 self._register_filters_if_needed()
688 context = await self._template_context(templates, triggered_action)
690 try:
691 return await asyncio.gather(
692 *[render_user_template(template, context) for template in templates]
693 )
694 except TemplateRenderError as e:
695 self._result_details["template_error"] = str(e)
696 self._result_details["template_source"] = e.template
697 raise ActionFailed(f"Template rendering failed: {e.error!r}") from e
700class DeploymentAction(Action):
701 """Base class for Actions that operate on Deployments and need to infer them from
702 events"""
704 source: Literal["selected", "inferred"] = Field(
705 "selected",
706 description=(
707 "Whether this Action applies to a specific selected "
708 "deployment (given by `deployment_id`), or to a deployment that is "
709 "inferred from the triggering event. If the source is 'inferred', "
710 "the `deployment_id` may not be set. If the source is 'selected', the "
711 "`deployment_id` must be set."
712 ),
713 )
714 deployment_id: Optional[UUID] = Field(
715 None, description="The identifier of the deployment"
716 )
718 @model_validator(mode="after")
719 def selected_deployment_requires_id(self) -> Self:
720 wants_selected_deployment = self.source == "selected"
721 has_deployment_id = bool(self.deployment_id)
722 if wants_selected_deployment != has_deployment_id:
723 raise ValueError(
724 "deployment_id is "
725 + ("not allowed" if has_deployment_id else "required")
726 )
727 return self
729 async def deployment_id_to_use(self, triggered_action: "TriggeredAction") -> UUID:
730 if self.source == "selected": 730 ↛ 734line 730 didn't jump to line 734 because the condition on line 730 was always true
731 assert self.deployment_id
732 return self.deployment_id
734 event = triggered_action.triggering_event
735 if not event:
736 raise ActionFailed("No event to infer the deployment")
738 assert event
739 if id := _id_of_first_resource_of_kind(event, "prefect.deployment"):
740 return id
742 raise ActionFailed("No deployment could be inferred")
745class DeploymentCommandAction(DeploymentAction, ExternalDataAction):
746 """Executes a command against a matching deployment"""
748 _action_description: ClassVar[str]
750 async def act(self, triggered_action: "TriggeredAction") -> None:
751 deployment_id = await self.deployment_id_to_use(triggered_action)
753 self._resulting_related_resources.append(
754 RelatedResource.model_validate(
755 {
756 "prefect.resource.id": f"prefect.deployment.{deployment_id}",
757 "prefect.resource.role": "target",
758 }
759 )
760 )
762 logger.info(
763 self._action_description,
764 extra={
765 "deployment_id": deployment_id,
766 **self.logging_context(triggered_action),
767 },
768 )
770 async with await self.orchestration_client(triggered_action) as orchestration:
771 response = await self.command(
772 orchestration, deployment_id, triggered_action
773 )
775 self._result_details["status_code"] = response.status_code
776 if response.status_code >= 300: 776 ↛ exitline 776 didn't jump to the function exit
777 raise ActionFailed(self.reason_from_response(response))
779 @abc.abstractmethod
780 async def command(
781 self,
782 orchestration: "OrchestrationClient",
783 deployment_id: UUID,
784 triggered_action: "TriggeredAction",
785 ) -> Response:
786 """Execute the deployment command"""
789class RunDeployment(JinjaTemplateAction, DeploymentCommandAction):
790 """Runs the given deployment with the given parameters"""
792 type: Literal["run-deployment"] = "run-deployment"
794 parameters: Optional[Dict[str, Any]] = Field(
795 None,
796 description=(
797 "The parameters to pass to the deployment, or None to use the "
798 "deployment's default parameters"
799 ),
800 )
801 job_variables: Optional[Dict[str, Any]] = Field(
802 None,
803 description=(
804 "The job variables to pass to the created flow run, or None "
805 "to use the deployment's default job variables"
806 ),
807 )
808 schedule_after: NonNegativeTimeDelta = Field(
809 default_factory=lambda: timedelta(0),
810 description=(
811 "The amount of time to wait before running the deployment. "
812 "Defaults to running the deployment immediately."
813 ),
814 )
816 _action_description: ClassVar[str] = "Running deployment"
818 async def command(
819 self,
820 orchestration: "OrchestrationClient",
821 deployment_id: UUID,
822 triggered_action: "TriggeredAction",
823 ) -> Response:
824 # Calculate when to schedule the deployment
825 scheduled_time = datetime.now(timezone.utc) + self.schedule_after
826 state = Scheduled(scheduled_time=scheduled_time)
828 try:
829 flow_run_create = DeploymentFlowRunCreate( # type: ignore
830 state=StateCreate(
831 type=state.type,
832 name=state.name,
833 message=state.message,
834 state_details=state.state_details,
835 ),
836 parameters=await self.render_parameters(triggered_action),
837 idempotency_key=triggered_action.idempotency_key(),
838 job_variables=self.job_variables,
839 )
840 except Exception as exc:
841 raise ActionFailed(f"Unable to create flow run from deployment: {exc!r}")
843 response = await orchestration.create_flow_run(deployment_id, flow_run_create)
845 if response.status_code < 300: 845 ↛ 846line 845 didn't jump to line 846 because the condition on line 845 was never true
846 flow_run = FlowRunResponse.model_validate(response.json())
848 self._resulting_related_resources.append(
849 RelatedResource.model_validate(
850 {
851 "prefect.resource.id": f"prefect.flow-run.{flow_run.id}",
852 "prefect.resource.role": "flow-run",
853 "prefect.resource.name": flow_run.name,
854 }
855 )
856 )
858 logger.info(
859 "Started flow run",
860 extra={
861 "flow_run": {
862 "id": str(flow_run.id),
863 "name": flow_run.name,
864 },
865 **self.logging_context(triggered_action),
866 },
867 )
869 if response.status_code == 409: 869 ↛ 870line 869 didn't jump to line 870 because the condition on line 869 was never true
870 self._result_details["validation_error"] = response.json().get("detail")
872 return response
874 @field_validator("parameters")
875 def validate_parameters(cls, value: dict[str, Any] | None) -> dict[str, Any] | None:
876 if not value:
877 return value
879 for_testing = copy.deepcopy(value) or {}
880 cls._upgrade_v1_templates(for_testing)
882 problems = cls._collect_errors(
883 hydrate(
884 for_testing,
885 HydrationContext(
886 raise_on_error=False,
887 render_workspace_variables=False,
888 render_jinja=False,
889 ),
890 )
891 )
892 if not problems: 892 ↛ 895line 892 didn't jump to line 895 because the condition on line 892 was always true
893 return value
895 raise ValueError(
896 "Invalid parameters: \n"
897 + "\n ".join(
898 f"{k + ':' if k else ''} {e.message}" for k, e in problems.items()
899 )
900 )
902 @classmethod
903 def _collect_errors(
904 cls,
905 hydrated: Union[dict[str, Any | dict[str, Any] | list[Any]], Placeholder],
906 prefix: str = "",
907 ) -> dict[str, HydrationError]:
908 problems: dict[str, HydrationError] = {}
910 if isinstance(hydrated, HydrationError): 910 ↛ 911line 910 didn't jump to line 911 because the condition on line 910 was never true
911 problems[prefix] = hydrated
913 if isinstance(hydrated, Placeholder): 913 ↛ 914line 913 didn't jump to line 914 because the condition on line 913 was never true
914 return problems
916 for key, value in hydrated.items():
917 if isinstance(value, dict):
918 problems.update(cls._collect_errors(value, f"{prefix}{key}."))
919 elif isinstance(value, list):
920 for item, index in enumerate(value):
921 if isinstance(item, dict): 921 ↛ 922line 921 didn't jump to line 922 because the condition on line 921 was never true
922 problems.update(
923 cls._collect_errors(item, f"{prefix}{key}[{index}].")
924 )
925 elif isinstance(item, HydrationError): 925 ↛ 926line 925 didn't jump to line 926 because the condition on line 925 was never true
926 problems[f"{prefix}{key}[{index}]"] = item
927 elif isinstance(value, HydrationError): 927 ↛ 928line 927 didn't jump to line 928 because the condition on line 927 was never true
928 problems[f"{prefix}{key}"] = value
930 return problems
932 async def render_parameters(
933 self, triggered_action: "TriggeredAction"
934 ) -> Dict[str, Any]:
935 parameters = copy.deepcopy(self.parameters) or {}
937 # pre-process the parameters to upgrade any v1-style template values to v2
938 self._upgrade_v1_templates(parameters)
940 # first-pass, hydrate parameters without rendering in order to collect all of
941 # the embedded Jinja templates, workspace variables, etc
942 placeholders = self._collect_placeholders(
943 hydrate(
944 parameters,
945 HydrationContext(
946 raise_on_error=False,
947 render_jinja=False,
948 render_workspace_variables=False,
949 ),
950 )
951 )
953 # collect all templates so we can build up the context variables they need
954 templates = [p.template for p in placeholders if isinstance(p, ValidJinja)]
955 template_context = await self._template_context(templates, triggered_action)
957 # collect any referenced workspace variables so we can fetch them
958 variable_names = [
959 p.variable_name for p in placeholders if isinstance(p, WorkspaceVariable)
960 ]
961 workspace_variables: Dict[str, StrictVariableValue] = {}
962 if variable_names: 962 ↛ 963line 962 didn't jump to line 963 because the condition on line 962 was never true
963 async with await self.orchestration_client(triggered_action) as client:
964 workspace_variables = await client.read_workspace_variables(
965 variable_names
966 )
968 # second-pass, render the parameters with the full context
969 parameters = hydrate(
970 parameters,
971 HydrationContext(
972 raise_on_error=True,
973 render_jinja=True,
974 jinja_context=template_context,
975 render_workspace_variables=True,
976 workspace_variables=workspace_variables,
977 ),
978 )
980 return parameters
982 @staticmethod
983 def _wrap_v1_template(template: str) -> Dict[str, Any]:
984 """Wraps a v1-style Jinja template string in the appropriate __prefect_kind
985 structure. Single-expression templates (e.g. '{{ value }}') are wrapped
986 in a json + jinja structure with '| tojson' so that the rendered value
987 preserves its original type through JSON round-tripping. Templates that
988 contain literal text outside expressions (e.g. 'Hello {{ name }}') are
989 wrapped in a plain jinja kind since their output is inherently a string."""
990 stripped = template.strip()
991 if (
992 stripped.startswith("{{")
993 and stripped.endswith("}}")
994 and stripped.count("{{") == 1
995 ):
996 # Single expression — wrap in json+jinja with | tojson to preserve
997 # the original value type (int, float, bool, None, list, dict, str).
998 inner_expr = stripped[2:-2].strip()
999 return {
1000 "__prefect_kind": "json",
1001 "value": {
1002 "__prefect_kind": "jinja",
1003 "template": "{{ " + inner_expr + " | tojson }}",
1004 },
1005 }
1006 return {"__prefect_kind": "jinja", "template": template}
1008 @classmethod
1009 def _upgrade_v1_templates(cls, parameters: Parameters):
1010 """
1011 Upgrades all v1-style template values from the parameters dictionary, changing
1012 the values in the given dictionary. v1-style templates are any plain strings
1013 that include Jinja2 template syntax.
1014 """
1015 for key, value in parameters.items():
1016 if isinstance(value, dict):
1017 # if we already have a __prefect_kind, don't upgrade or recurse
1018 if "__prefect_kind" in value: 1018 ↛ 1019line 1018 didn't jump to line 1019 because the condition on line 1018 was never true
1019 continue
1020 cls._upgrade_v1_templates(value)
1021 elif isinstance(value, list):
1022 for i, item in enumerate(value):
1023 if isinstance(item, dict):
1024 cls._upgrade_v1_templates(item)
1025 elif isinstance(item, str) and maybe_template(item): 1025 ↛ 1026line 1025 didn't jump to line 1026 because the condition on line 1025 was never true
1026 value[i] = cls._wrap_v1_template(item)
1027 elif isinstance(value, str) and maybe_template(value): # pyright: ignore[reportUnnecessaryIsInstance] 1027 ↛ 1028line 1027 didn't jump to line 1028 because the condition on line 1027 was never true
1028 parameters[key] = cls._wrap_v1_template(value)
1030 def _collect_placeholders(
1031 self, parameters: Parameters | Placeholder
1032 ) -> list[Placeholder]:
1033 """
1034 Recursively collects all placeholder values embedded within the parameters
1035 dictionary, including templates and workspace variables
1036 """
1037 placeholders: list[Placeholder] = []
1039 if isinstance(parameters, Placeholder): 1039 ↛ 1040line 1039 didn't jump to line 1040 because the condition on line 1039 was never true
1040 return [parameters]
1042 for _, value in parameters.items(): 1042 ↛ 1043line 1042 didn't jump to line 1043 because the loop on line 1042 never started
1043 if isinstance(value, dict):
1044 placeholders += self._collect_placeholders(value)
1045 elif isinstance(value, list):
1046 for item in value:
1047 if isinstance(item, dict):
1048 placeholders += self._collect_placeholders(item)
1049 elif isinstance(item, Placeholder):
1050 placeholders.append(item)
1051 elif isinstance(value, Placeholder):
1052 placeholders.append(value)
1053 return placeholders
1056class PauseDeployment(DeploymentCommandAction):
1057 """Pauses the given Deployment"""
1059 type: Literal["pause-deployment"] = "pause-deployment"
1061 _action_description: ClassVar[str] = "Pausing deployment"
1063 async def command(
1064 self,
1065 orchestration: "OrchestrationClient",
1066 deployment_id: UUID,
1067 triggered_action: "TriggeredAction",
1068 ) -> Response:
1069 return await orchestration.pause_deployment(deployment_id)
1072class ResumeDeployment(DeploymentCommandAction):
1073 """Resumes the given Deployment"""
1075 type: Literal["resume-deployment"] = "resume-deployment"
1077 _action_description: ClassVar[str] = "Resuming deployment"
1079 async def command(
1080 self,
1081 orchestration: "OrchestrationClient",
1082 deployment_id: UUID,
1083 triggered_action: "TriggeredAction",
1084 ) -> Response:
1085 return await orchestration.resume_deployment(deployment_id)
1088class FlowRunAction(ExternalDataAction):
1089 """An action that operates on a flow run"""
1091 async def flow_run(self, triggered_action: "TriggeredAction") -> UUID:
1092 # Proactive triggers won't have an event, but they might be tracking
1093 # buckets per-resource, so check for that first
1094 labels = triggered_action.triggering_labels
1095 if triggering_resource_id := labels.get("prefect.resource.id"): 1095 ↛ 1096line 1095 didn't jump to line 1096 because the condition on line 1095 was never true
1096 if id := _id_from_resource_id(triggering_resource_id, "prefect.flow-run"):
1097 return id
1099 event = triggered_action.triggering_event
1100 if event:
1101 if id := _id_of_first_resource_of_kind(event, "prefect.flow-run"): 1101 ↛ 1102line 1101 didn't jump to line 1102 because the condition on line 1101 was never true
1102 return id
1104 raise ActionFailed("No flow run could be inferred")
1107class FlowRunStateChangeAction(FlowRunAction):
1108 """Changes the state of a flow run associated with the trigger"""
1110 @abc.abstractmethod
1111 async def new_state(self, triggered_action: "TriggeredAction") -> StateCreate:
1112 """Return the new state for the flow run"""
1114 async def act(self, triggered_action: "TriggeredAction") -> None:
1115 flow_run_id = await self.flow_run(triggered_action)
1117 self._resulting_related_resources.append(
1118 RelatedResource.model_validate(
1119 {
1120 "prefect.resource.id": f"prefect.flow-run.{flow_run_id}",
1121 "prefect.resource.role": "target",
1122 }
1123 )
1124 )
1126 logger.info(
1127 "Changing flow run state",
1128 extra={
1129 "flow_run_id": str(flow_run_id),
1130 **self.logging_context(triggered_action),
1131 },
1132 )
1134 async with await self.orchestration_client(triggered_action) as orchestration:
1135 response = await orchestration.set_flow_run_state(
1136 flow_run_id,
1137 await self.new_state(triggered_action=triggered_action),
1138 force=getattr(self, "force", False),
1139 )
1141 self._result_details["status_code"] = response.status_code
1142 if response.status_code >= 300:
1143 raise ActionFailed(self.reason_from_response(response))
1145 result = OrchestrationResult.model_validate(response.json())
1146 if not isinstance(result.details, StateAcceptDetails):
1147 raise ActionFailed(f"Failed to set state: {result.details.reason}")
1150class ChangeFlowRunState(FlowRunStateChangeAction):
1151 """Changes the state of a flow run associated with the trigger"""
1153 type: Literal["change-flow-run-state"] = "change-flow-run-state"
1155 name: str | None = Field(
1156 None,
1157 description="The name of the state to change the flow run to",
1158 )
1159 state: StateType = Field(
1160 ...,
1161 description="The type of the state to change the flow run to",
1162 )
1163 message: str | None = Field(
1164 None,
1165 description="An optional message to associate with the state change",
1166 )
1167 force: bool = Field(
1168 False,
1169 description="Force the state change even if the transition is not allowed",
1170 )
1172 async def new_state(self, triggered_action: "TriggeredAction") -> StateCreate:
1173 message = (
1174 self.message
1175 or f"State changed by Automation {triggered_action.automation.id}"
1176 )
1178 return StateCreate(
1179 name=self.name,
1180 type=self.state,
1181 message=message,
1182 )
1185class CancelFlowRun(FlowRunStateChangeAction):
1186 """Cancels a flow run associated with the trigger"""
1188 type: Literal["cancel-flow-run"] = "cancel-flow-run"
1190 async def new_state(self, triggered_action: "TriggeredAction") -> StateCreate:
1191 return StateCreate(
1192 type=StateType.CANCELLING,
1193 message=f"Cancelled by Automation {triggered_action.automation.id}",
1194 )
1197class DeleteFlowRun(FlowRunAction):
1198 """Deletes a flow run associated with the trigger"""
1200 type: Literal["delete-flow-run"] = "delete-flow-run"
1202 async def act(self, triggered_action: "TriggeredAction") -> None:
1203 flow_run_id = await self.flow_run(triggered_action)
1205 self._resulting_related_resources.append(
1206 RelatedResource.model_validate(
1207 {
1208 "prefect.resource.id": f"prefect.flow-run.{flow_run_id}",
1209 "prefect.resource.role": "target",
1210 }
1211 )
1212 )
1214 logger.info(
1215 "Deleting flow run",
1216 extra={
1217 "flow_run_id": str(flow_run_id),
1218 **self.logging_context(triggered_action),
1219 },
1220 )
1222 async with await self.orchestration_client(triggered_action) as orchestration:
1223 response = await orchestration.delete_flow_run(flow_run_id)
1225 self._result_details["status_code"] = response.status_code
1226 if response.status_code >= 300:
1227 raise ActionFailed(self.reason_from_response(response))
1230class SuspendFlowRun(FlowRunStateChangeAction):
1231 """Suspends a flow run associated with the trigger"""
1233 type: Literal["suspend-flow-run"] = "suspend-flow-run"
1235 async def new_state(self, triggered_action: "TriggeredAction") -> StateCreate:
1236 state = Suspended(
1237 timeout_seconds=3600,
1238 message=f"Suspended by Automation {triggered_action.automation.id}",
1239 )
1241 return StateCreate(
1242 type=state.type,
1243 name=state.name,
1244 message=state.message,
1245 state_details=state.state_details,
1246 )
1249class ResumeFlowRun(FlowRunAction):
1250 """Resumes a paused or suspended flow run associated with the trigger"""
1252 type: Literal["resume-flow-run"] = "resume-flow-run"
1254 async def act(self, triggered_action: "TriggeredAction") -> None:
1255 flow_run_id = await self.flow_run(triggered_action)
1257 self._resulting_related_resources.append(
1258 RelatedResource.model_validate(
1259 {
1260 "prefect.resource.id": f"prefect.flow-run.{flow_run_id}",
1261 "prefect.resource.role": "target",
1262 }
1263 )
1264 )
1266 logger.debug(
1267 "Resuming flow run",
1268 extra={
1269 "flow_run_id": str(flow_run_id),
1270 **self.logging_context(triggered_action),
1271 },
1272 )
1274 async with await self.orchestration_client(triggered_action) as orchestration:
1275 result = await orchestration.resume_flow_run(flow_run_id)
1277 if not isinstance(result.details, StateAcceptDetails):
1278 raise ActionFailed(
1279 f"Failed to resume flow run: {result.details.reason}"
1280 )
1283class CallWebhook(JinjaTemplateAction):
1284 """Call a webhook when an Automation is triggered."""
1286 type: Literal["call-webhook"] = "call-webhook"
1287 block_document_id: UUID = Field(
1288 description="The identifier of the webhook block to use"
1289 )
1290 payload: str = Field(
1291 default="",
1292 description="An optional templatable payload to send when calling the webhook.",
1293 )
1295 @field_validator("payload", mode="before")
1296 @classmethod
1297 def ensure_payload_is_a_string(
1298 cls, value: Union[str, Dict[str, Any], None]
1299 ) -> Optional[str]:
1300 """Temporary measure while we migrate payloads from being a dictionary to
1301 a string template. This covers both reading from the database where values
1302 may currently be a dictionary, as well as the API, where older versions of the
1303 frontend may be sending a JSON object with the single `"message"` key."""
1304 if value is None:
1305 return value
1307 if isinstance(value, str):
1308 return value
1310 return orjson.dumps(value, option=orjson.OPT_INDENT_2).decode()
1312 @field_validator("payload")
1313 @classmethod
1314 def validate_payload_templates(cls, value: Optional[str]) -> Optional[str]:
1315 """
1316 Validate user-provided payload template.
1317 """
1318 if not value:
1319 return value
1321 cls.validate_template(value, "payload")
1323 return value
1325 async def _get_webhook_block(self, triggered_action: "TriggeredAction") -> Webhook:
1326 async with await self.orchestration_client(triggered_action) as orchestration:
1327 response = await orchestration.read_block_document_raw(
1328 self.block_document_id
1329 )
1330 if response.status_code >= 300: 1330 ↛ 1333line 1330 didn't jump to line 1333 because the condition on line 1330 was always true
1331 raise ActionFailed(self.reason_from_response(response))
1333 try:
1334 block_document = BlockDocument.model_validate(response.json())
1335 block = await _load_block_from_block_document(block_document)
1336 except Exception as e:
1337 raise ActionFailed(f"The webhook block was invalid: {e!r}")
1339 if not isinstance(block, Webhook):
1340 raise ActionFailed("The referenced block was not a webhook block")
1342 self._resulting_related_resources += [
1343 RelatedResource.model_validate(
1344 {
1345 "prefect.resource.id": f"prefect.block-document.{self.block_document_id}",
1346 "prefect.resource.role": "block",
1347 "prefect.resource.name": block_document.name,
1348 }
1349 ),
1350 RelatedResource.model_validate(
1351 {
1352 "prefect.resource.id": f"prefect.block-type.{block.get_block_type_slug()}",
1353 "prefect.resource.role": "block-type",
1354 }
1355 ),
1356 ]
1358 return block
1360 async def act(self, triggered_action: "TriggeredAction") -> None:
1361 block = await self._get_webhook_block(triggered_action=triggered_action)
1363 (payload,) = await self._render([self.payload], triggered_action)
1365 try:
1366 response = await block.call(payload=payload)
1368 ok_headers = {
1369 k: v for k, v in response.headers.items() if not should_redact_header(k)
1370 }
1372 self._result_details.update(
1373 {
1374 "status_code": response.status_code,
1375 "response_body": truncated_to(1000, response.text),
1376 "response_headers": {**(ok_headers or {})},
1377 }
1378 )
1379 except Exception as e:
1380 raise ActionFailed(f"Webhook call failed: {e!r}")
1383class SendNotification(JinjaTemplateAction):
1384 """Send a notification when an Automation is triggered"""
1386 type: Literal["send-notification"] = "send-notification"
1387 block_document_id: UUID = Field(
1388 description="The identifier of the notification block to use"
1389 )
1390 subject: str = Field("Prefect automated notification")
1391 body: str = Field(description="The text of the notification to send")
1393 @field_validator("subject", "body")
1394 def is_valid_template(cls, value: str, info: ValidationInfo) -> str:
1395 if TYPE_CHECKING: 1395 ↛ 1396line 1395 didn't jump to line 1396 because the condition on line 1395 was never true
1396 assert isinstance(info.field_name, str)
1397 return cls.validate_template(value, info.field_name)
1399 async def _get_notification_block(
1400 self, triggered_action: "TriggeredAction"
1401 ) -> NotificationBlock:
1402 async with await self.orchestration_client(triggered_action) as orion:
1403 response = await orion.read_block_document_raw(self.block_document_id)
1404 if response.status_code >= 300:
1405 raise ActionFailed(self.reason_from_response(response))
1407 try:
1408 block_document = BlockDocument.model_validate(response.json())
1409 block = await _load_block_from_block_document(block_document)
1410 except Exception as e:
1411 raise ActionFailed(f"The notification block was invalid: {e!r}")
1413 if "notify" not in block.get_block_capabilities():
1414 raise ActionFailed("The referenced block was not a notification block")
1416 self._resulting_related_resources += [
1417 RelatedResource.model_validate(
1418 {
1419 "prefect.resource.id": f"prefect.block-document.{self.block_document_id}",
1420 "prefect.resource.role": "block",
1421 "prefect.resource.name": block_document.name,
1422 }
1423 ),
1424 RelatedResource.model_validate(
1425 {
1426 "prefect.resource.id": f"prefect.block-type.{block.get_block_type_slug()}",
1427 "prefect.resource.role": "block-type",
1428 }
1429 ),
1430 ]
1432 return cast(NotificationBlock, block)
1434 async def act(self, triggered_action: "TriggeredAction") -> None:
1435 block = await self._get_notification_block(triggered_action=triggered_action)
1437 subject, body = await self.render(triggered_action)
1439 with block.raise_on_failure():
1440 try:
1441 await block.notify(subject=subject, body=body)
1442 except NotificationError as e:
1443 self._result_details["notification_log"] = e.log
1444 raise ActionFailed("Notification failed")
1446 async def render(self, triggered_action: "TriggeredAction") -> List[str]:
1447 return await self._render([self.subject, self.body], triggered_action)
1450class WorkPoolAction(Action):
1451 """Base class for Actions that operate on Work Pools and need to infer them from
1452 events"""
1454 source: Literal["selected", "inferred"] = Field(
1455 "selected",
1456 description=(
1457 "Whether this Action applies to a specific selected "
1458 "work pool (given by `work_pool_id`), or to a work pool that is "
1459 "inferred from the triggering event. If the source is 'inferred', "
1460 "the `work_pool_id` may not be set. If the source is 'selected', the "
1461 "`work_pool_id` must be set."
1462 ),
1463 )
1464 work_pool_id: Optional[UUID] = Field(
1465 None,
1466 description="The identifier of the work pool to pause",
1467 )
1469 @model_validator(mode="after")
1470 def selected_work_pool_requires_id(self) -> Self:
1471 wants_selected_work_pool = self.source == "selected"
1472 has_work_pool_id = bool(self.work_pool_id)
1473 if wants_selected_work_pool != has_work_pool_id:
1474 raise ValueError(
1475 "work_pool_id is " + ("not allowed" if has_work_pool_id else "required")
1476 )
1477 return self
1479 async def work_pool_id_to_use(self, triggered_action: "TriggeredAction") -> UUID:
1480 if self.source == "selected":
1481 assert self.work_pool_id
1482 return self.work_pool_id
1484 event = triggered_action.triggering_event
1485 if not event:
1486 raise ActionFailed("No event to infer the work pool")
1488 assert event
1489 if id := _id_of_first_resource_of_kind(event, "prefect.work-pool"):
1490 return id
1492 raise ActionFailed("No work pool could be inferred")
1495class WorkPoolCommandAction(WorkPoolAction, ExternalDataAction):
1496 _action_description: ClassVar[str]
1498 _target_work_pool: Optional[WorkPool] = PrivateAttr(default=None)
1500 async def target_work_pool(self, triggered_action: "TriggeredAction") -> WorkPool:
1501 if not self._target_work_pool:
1502 work_pool_id = await self.work_pool_id_to_use(triggered_action)
1504 async with await self.orchestration_client(
1505 triggered_action
1506 ) as orchestration:
1507 work_pool = await orchestration.read_work_pool(work_pool_id)
1509 if not work_pool:
1510 raise ActionFailed(f"Work pool {work_pool_id} not found")
1511 self._target_work_pool = work_pool
1512 return self._target_work_pool
1514 async def act(self, triggered_action: "TriggeredAction") -> None:
1515 work_pool = await self.target_work_pool(triggered_action)
1517 self._resulting_related_resources += [
1518 RelatedResource.model_validate(
1519 {
1520 "prefect.resource.id": f"prefect.work-pool.{work_pool.id}",
1521 "prefect.resource.name": work_pool.name,
1522 "prefect.resource.role": "target",
1523 }
1524 )
1525 ]
1527 logger.info(
1528 self._action_description,
1529 extra={
1530 "work_pool_id": work_pool.id,
1531 **self.logging_context(triggered_action),
1532 },
1533 )
1535 async with await self.orchestration_client(triggered_action) as orchestration:
1536 response = await self.command(orchestration, work_pool, triggered_action)
1538 self._result_details["status_code"] = response.status_code
1539 if response.status_code >= 300:
1540 raise ActionFailed(self.reason_from_response(response))
1542 @abc.abstractmethod
1543 async def command(
1544 self,
1545 orchestration: "OrchestrationClient",
1546 work_pool: WorkPool,
1547 triggered_action: "TriggeredAction",
1548 ) -> Response:
1549 """Issue the command to the Work Pool"""
1552class PauseWorkPool(WorkPoolCommandAction):
1553 """Pauses a Work Pool"""
1555 type: Literal["pause-work-pool"] = "pause-work-pool"
1557 _action_description: ClassVar[str] = "Pausing work pool"
1559 async def command(
1560 self,
1561 orchestration: "OrchestrationClient",
1562 work_pool: WorkPool,
1563 triggered_action: "TriggeredAction",
1564 ) -> Response:
1565 return await orchestration.pause_work_pool(work_pool.name)
1568class ResumeWorkPool(WorkPoolCommandAction):
1569 """Resumes a Work Pool"""
1571 type: Literal["resume-work-pool"] = "resume-work-pool"
1573 _action_description: ClassVar[str] = "Resuming work pool"
1575 async def command(
1576 self,
1577 orchestration: "OrchestrationClient",
1578 work_pool: WorkPool,
1579 triggered_action: "TriggeredAction",
1580 ) -> Response:
1581 return await orchestration.resume_work_pool(work_pool.name)
1584class WorkQueueAction(Action):
1585 """Base class for Actions that operate on Work Queues and need to infer them from
1586 events"""
1588 source: Literal["selected", "inferred"] = Field(
1589 "selected",
1590 description=(
1591 "Whether this Action applies to a specific selected "
1592 "work queue (given by `work_queue_id`), or to a work queue that is "
1593 "inferred from the triggering event. If the source is 'inferred', "
1594 "the `work_queue_id` may not be set. If the source is 'selected', the "
1595 "`work_queue_id` must be set."
1596 ),
1597 )
1598 work_queue_id: Optional[UUID] = Field(
1599 None, description="The identifier of the work queue to pause"
1600 )
1602 @model_validator(mode="after")
1603 def selected_work_queue_requires_id(self) -> Self:
1604 wants_selected_work_queue = self.source == "selected"
1605 has_work_queue_id = bool(self.work_queue_id)
1606 if wants_selected_work_queue != has_work_queue_id:
1607 raise ValueError(
1608 "work_queue_id is "
1609 + ("not allowed" if has_work_queue_id else "required")
1610 )
1611 return self
1613 async def work_queue_id_to_use(self, triggered_action: "TriggeredAction") -> UUID:
1614 if self.source == "selected": 1614 ↛ 1615line 1614 didn't jump to line 1615 because the condition on line 1614 was never true
1615 assert self.work_queue_id
1616 return self.work_queue_id
1618 event = triggered_action.triggering_event
1619 if not event:
1620 raise ActionFailed("No event to infer the work queue")
1622 assert event
1623 if id := _id_of_first_resource_of_kind(event, "prefect.work-queue"): 1623 ↛ 1624line 1623 didn't jump to line 1624 because the condition on line 1623 was never true
1624 return id
1626 raise ActionFailed("No work queue could be inferred")
1629class WorkQueueCommandAction(WorkQueueAction, ExternalDataAction):
1630 _action_description: ClassVar[str]
1632 async def act(self, triggered_action: "TriggeredAction") -> None:
1633 work_queue_id = await self.work_queue_id_to_use(triggered_action)
1635 self._resulting_related_resources += [
1636 RelatedResource.model_validate(
1637 {
1638 "prefect.resource.id": f"prefect.work-queue.{work_queue_id}",
1639 "prefect.resource.role": "target",
1640 }
1641 )
1642 ]
1644 logger.info(
1645 self._action_description,
1646 extra={
1647 "work_queue_id": work_queue_id,
1648 **self.logging_context(triggered_action),
1649 },
1650 )
1652 async with await self.orchestration_client(triggered_action) as orchestration:
1653 response = await self.command(
1654 orchestration, work_queue_id, triggered_action
1655 )
1657 self._result_details["status_code"] = response.status_code
1658 if response.status_code >= 300:
1659 raise ActionFailed(self.reason_from_response(response))
1661 @abc.abstractmethod
1662 async def command(
1663 self,
1664 orchestration: "OrchestrationClient",
1665 work_queue_id: UUID,
1666 triggered_action: "TriggeredAction",
1667 ) -> Response:
1668 """Issue the command to the Work Queue"""
1671class PauseWorkQueue(WorkQueueCommandAction):
1672 """Pauses a Work Queue"""
1674 type: Literal["pause-work-queue"] = "pause-work-queue"
1676 _action_description: ClassVar[str] = "Pausing work queue"
1678 async def command(
1679 self,
1680 orchestration: "OrchestrationClient",
1681 work_queue_id: UUID,
1682 triggered_action: "TriggeredAction",
1683 ) -> Response:
1684 return await orchestration.pause_work_queue(work_queue_id)
1687class ResumeWorkQueue(WorkQueueCommandAction):
1688 """Resumes a Work Queue"""
1690 type: Literal["resume-work-queue"] = "resume-work-queue"
1692 _action_description: ClassVar[str] = "Resuming work queue"
1694 async def command(
1695 self,
1696 orchestration: "OrchestrationClient",
1697 work_queue_id: UUID,
1698 triggered_action: "TriggeredAction",
1699 ) -> Response:
1700 return await orchestration.resume_work_queue(work_queue_id)
1703class AutomationAction(Action):
1704 """Base class for Actions that operate on Automations and need to infer them from
1705 events"""
1707 source: Literal["selected", "inferred"] = Field(
1708 "selected",
1709 description=(
1710 "Whether this Action applies to a specific selected "
1711 "automation (given by `automation_id`), or to an automation that is "
1712 "inferred from the triggering event. If the source is 'inferred', "
1713 "the `automation_id` may not be set. If the source is 'selected', the "
1714 "`automation_id` must be set."
1715 ),
1716 )
1717 automation_id: Optional[UUID] = Field(
1718 None, description="The identifier of the automation to act on"
1719 )
1721 @model_validator(mode="after")
1722 def selected_automation_requires_id(self) -> Self:
1723 wants_selected_automation = self.source == "selected"
1724 has_automation_id = bool(self.automation_id)
1725 if wants_selected_automation != has_automation_id:
1726 raise ValueError(
1727 "automation_id is "
1728 + ("not allowed" if has_automation_id else "required")
1729 )
1730 return self
1732 async def automation_id_to_use(self, triggered_action: "TriggeredAction") -> UUID:
1733 if self.source == "selected": 1733 ↛ 1734line 1733 didn't jump to line 1734 because the condition on line 1733 was never true
1734 assert self.automation_id
1735 return self.automation_id
1737 event = triggered_action.triggering_event
1738 if not event: 1738 ↛ 1739line 1738 didn't jump to line 1739 because the condition on line 1738 was never true
1739 raise ActionFailed("No event to infer the automation")
1741 assert event
1742 if id := _id_of_first_resource_of_kind(event, "prefect.automation"): 1742 ↛ 1745line 1742 didn't jump to line 1745 because the condition on line 1742 was always true
1743 return id
1745 raise ActionFailed("No automation could be inferred")
1748class AutomationCommandAction(AutomationAction, ExternalDataAction):
1749 _action_description: ClassVar[str]
1751 async def act(self, triggered_action: "TriggeredAction") -> None:
1752 automation_id = await self.automation_id_to_use(triggered_action)
1754 self._resulting_related_resources += [
1755 RelatedResource.model_validate(
1756 {
1757 "prefect.resource.id": f"prefect.automation.{automation_id}",
1758 "prefect.resource.role": "target",
1759 }
1760 )
1761 ]
1763 logger.info(
1764 self._action_description,
1765 extra={
1766 "automation_id": automation_id,
1767 **self.logging_context(triggered_action),
1768 },
1769 )
1771 async with await self.events_api_client(triggered_action) as events:
1772 response = await self.command(events, automation_id, triggered_action)
1774 self._result_details["status_code"] = response.status_code
1775 if response.status_code >= 300: 1775 ↛ 1776line 1775 didn't jump to line 1776 because the condition on line 1775 was never true
1776 raise ActionFailed(self.reason_from_response(response))
1778 @abc.abstractmethod
1779 async def command(
1780 self,
1781 events: PrefectServerEventsAPIClient,
1782 automation_id: UUID,
1783 triggered_action: "TriggeredAction",
1784 ) -> Response:
1785 """Issue the command to the Work Queue"""
1788class PauseAutomation(AutomationCommandAction):
1789 """Pauses a Work Queue"""
1791 type: Literal["pause-automation"] = "pause-automation"
1793 _action_description: ClassVar[str] = "Pausing automation"
1795 async def command(
1796 self,
1797 events: PrefectServerEventsAPIClient,
1798 automation_id: UUID,
1799 triggered_action: "TriggeredAction",
1800 ) -> Response:
1801 return await events.pause_automation(automation_id)
1804class ResumeAutomation(AutomationCommandAction):
1805 """Resumes a Work Queue"""
1807 type: Literal["resume-automation"] = "resume-automation"
1809 _action_description: ClassVar[str] = "Resuming auitomation"
1811 async def command(
1812 self,
1813 events: PrefectServerEventsAPIClient,
1814 automation_id: UUID,
1815 triggered_action: "TriggeredAction",
1816 ) -> Response:
1817 return await events.resume_automation(automation_id)
1820# The actual action types that we support. It's important to update this
1821# Union when adding new subclasses of Action so that they are available for clients
1822# and in the OpenAPI docs
1823ServerActionTypes: TypeAlias = Union[
1824 DoNothing,
1825 RunDeployment,
1826 PauseDeployment,
1827 ResumeDeployment,
1828 CancelFlowRun,
1829 DeleteFlowRun,
1830 ChangeFlowRunState,
1831 PauseWorkQueue,
1832 ResumeWorkQueue,
1833 SendNotification,
1834 CallWebhook,
1835 PauseAutomation,
1836 ResumeAutomation,
1837 SuspendFlowRun,
1838 ResumeFlowRun,
1839 PauseWorkPool,
1840 ResumeWorkPool,
1841]
1844_recent_actions: MutableMapping[UUID, bool] = TTLCache(maxsize=10000, ttl=3600)
1847async def record_action_happening(id: UUID) -> None:
1848 """Record that an action has happened, with an expiration of an hour."""
1849 _recent_actions[id] = True
1852async def action_has_already_happened(id: UUID) -> bool:
1853 """Check if the action has already happened"""
1854 return _recent_actions.get(id, False)
1857@asynccontextmanager
1858async def consumer() -> AsyncGenerator[MessageHandler, None]:
1859 from prefect.server.events.schemas.automations import TriggeredAction
1861 async def message_handler(message: Message):
1862 if not message.data: 1862 ↛ 1863line 1862 didn't jump to line 1863 because the condition on line 1862 was never true
1863 return
1865 triggered_action = TriggeredAction.model_validate_json(message.data)
1866 action = triggered_action.action
1868 if await action_has_already_happened(triggered_action.id): 1868 ↛ 1869line 1868 didn't jump to line 1869 because the condition on line 1868 was never true
1869 logger.info(
1870 "Action %s has already been executed, skipping",
1871 triggered_action.id,
1872 )
1873 return
1875 try:
1876 await action.act(triggered_action)
1877 except ActionFailed as e:
1878 # ActionFailed errors are expected errors and will not be retried
1879 await action.fail(triggered_action, e.reason)
1880 else:
1881 await action.succeed(triggered_action)
1882 await record_action_happening(triggered_action.id)
1884 logger.info("Starting action message handler")
1885 yield message_handler
1888async def _load_block_from_block_document(
1889 block_document: BlockDocument,
1890) -> Block:
1891 if block_document.block_schema is None:
1892 raise ValueError("Unable to determine block schema for provided block document")
1894 block_cls = Block.get_block_class_from_schema(block_document.block_schema)
1896 block = block_cls.model_validate(block_document.data)
1897 block._block_document_id = block_document.id
1898 block.__class__._block_schema_id = block_document.block_schema_id
1899 block.__class__._block_type_id = block_document.block_type_id
1900 block._block_document_name = block_document.name
1901 block._is_anonymous = block_document.is_anonymous
1902 block._define_metadata_on_nested_blocks(block_document.block_document_references)
1904 resources = block._event_method_called_resources()
1905 if resources:
1906 kind = block._event_kind()
1907 resource, related = resources
1908 async with PrefectServerEventsClient() as events_client:
1909 await events_client.emit(
1910 Event(
1911 id=uuid7(),
1912 occurred=now("UTC"),
1913 event=f"{kind}.loaded",
1914 resource=Resource.model_validate(resource),
1915 related=[RelatedResource.model_validate(r) for r in related],
1916 )
1917 )
1919 return block