Coverage for /usr/local/lib/python3.12/site-packages/prefect/_experimental/sla/objects.py: 80%
33 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +0000
1from __future__ import annotations 1a
3import abc 1a
4from datetime import timedelta 1a
5from typing import Literal, Optional, Union 1a
6from uuid import UUID 1a
8from pydantic import Field, PrivateAttr, computed_field 1a
9from typing_extensions import Self, TypeAlias 1a
11from prefect._internal.schemas.bases import PrefectBaseModel 1a
14class ServiceLevelAgreement(PrefectBaseModel, abc.ABC): 1a
15 """An ORM representation of a Service Level Agreement."""
17 _deployment_id: Optional[UUID] = PrivateAttr(default=None) 1a
19 name: str = Field( 1a
20 default=...,
21 description="The name of the SLA. Names must be unique on a per-deployment basis.",
22 )
23 severity: Literal["minor", "low", "moderate", "high", "critical"] = Field( 1a
24 default="moderate",
25 description="The severity of the SLA.",
26 )
27 enabled: Optional[bool] = Field( 1a
28 default=True,
29 description="Whether the SLA is enabled.",
30 )
32 def set_deployment_id(self, deployment_id: UUID) -> Self: 1a
33 self._deployment_id = deployment_id
34 return self
36 @computed_field 1a
37 @property 1a
38 def owner_resource(self) -> Union[str, None]: 1a
39 if self._deployment_id:
40 return f"prefect.deployment.{self._deployment_id}"
41 return None
44class TimeToCompletionSla(ServiceLevelAgreement): 1a
45 """An SLA that triggers when a flow run takes longer than the specified duration."""
47 duration: int = Field( 1a
48 default=...,
49 description="The maximum flow run duration allowed before the SLA is violated, expressed in seconds.",
50 )
53class FrequencySla(ServiceLevelAgreement): 1a
54 """An SLA that triggers when a completed flow run is not detected in the specified time.
56 For example, if stale_after is 1 hour, if a flow run does not complete
57 within an hour of the previous flow run, the SLA will trigger.
58 """
60 stale_after: timedelta = Field( 1a
61 default=...,
62 description="The amount of time after which a flow run is considered in violation.",
63 )
66class LatenessSla(ServiceLevelAgreement): 1a
67 """An SLA that triggers when a flow run does not start within the specified window.
69 For example, if you schedule the deployment to run every day at 2:00pm and you pass
70 within=timedelta(minutes=10) to this SLA, if a run hasn't started by 2:10pm the SLA
71 violation will be recorded.
72 """
74 within: timedelta = Field( 1a
75 default=...,
76 description="The amount of time before a flow run is considered in violation.",
77 )
80class SlaMergeResponse(PrefectBaseModel): 1a
81 """A response object for the apply_slas_for_deployment method. Contains the names of the created, updated, and deleted SLAs."""
83 created: list[str] 1a
84 updated: list[str] 1a
85 deleted: list[str] 1a
88# Concrete SLA types
89SlaTypes: TypeAlias = Union[TimeToCompletionSla, LatenessSla, FrequencySla] 1a