Coverage for /usr/local/lib/python3.12/site-packages/prefect/cli/deploy/_models.py: 77%

79 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 11:21 +0000

1from __future__ import annotations 1a

2 

3from typing import Any, Dict, List, Optional, Union 1a

4 

5from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator 1a

6 

7from prefect._experimental.sla.objects import SlaTypes 1a

8from prefect.client.schemas.actions import DeploymentScheduleCreate 1a

9from prefect.client.schemas.schedules import SCHEDULE_TYPES 1a

10from prefect.events import DeploymentTriggerTypes 1a

11 

12 

13class WorkPoolConfig(BaseModel): 1a

14 model_config = ConfigDict(extra="ignore") 1a

15 

16 name: Optional[str] = None 1a

17 work_queue_name: Optional[str] = None 1a

18 job_variables: Dict[str, Any] = Field(default_factory=dict) 1a

19 

20 

21class DeploymentConfig(BaseModel): 1a

22 model_config = ConfigDict(extra="ignore") 1a

23 

24 # base metadata 

25 name: Optional[str] = None 1a

26 version: Optional[str] = None 1a

27 version_type: Optional[str] = None 1a

28 tags: Optional[Union[str, list[Any]]] = ( 1a

29 None # allow raw templated string or list; templating will normalize 

30 ) 

31 description: Optional[str] = None 1a

32 

33 # schedule metadata 

34 schedule: Optional["ScheduleItem"] = None 1a

35 schedules: Optional[List["ScheduleItem"]] = None 1a

36 paused: Optional[bool] = None 1a

37 concurrency_limit: Optional[Union[int, "ConcurrencyLimitSpec"]] = None 1a

38 

39 # flow-specific 

40 flow_name: Optional[str] = None 1a

41 entrypoint: Optional[str] = None 1a

42 parameters: Dict[str, Any] = Field(default_factory=dict) 1a

43 enforce_parameter_schema: Optional[bool] = None 1a

44 

45 # per-deployment actions (optional overrides) 

46 # Accept list, mapping (empty or step), or null for flexibility 

47 build: Optional[Union[List[Dict[str, Any]], Dict[str, Any]]] = None 1a

48 push: Optional[Union[List[Dict[str, Any]], Dict[str, Any]]] = None 1a

49 pull: Optional[Union[List[Dict[str, Any]], Dict[str, Any]]] = None 1a

50 

51 # infra-specific 

52 work_pool: Optional[WorkPoolConfig] = None 1a

53 

54 # automations metadata 

55 triggers: Optional[List[DeploymentTriggerTypes]] = None 1a

56 sla: Optional[List[SlaTypes]] = None 1a

57 

58 

59class PrefectYamlModel(BaseModel): 1a

60 model_config = ConfigDict(populate_by_name=True, extra="ignore") 1a

61 

62 # generic metadata (currently unused by CLI but allowed) 

63 prefect_version: Optional[str] = Field(default=None, alias="prefect-version") 1a

64 name: Optional[str] = None 1a

65 

66 # global actions 

67 build: Optional[Union[List[Dict[str, Any]], Dict[str, Any]]] = None 1a

68 push: Optional[Union[List[Dict[str, Any]], Dict[str, Any]]] = None 1a

69 pull: Optional[Union[List[Dict[str, Any]], Dict[str, Any]]] = None 1a

70 

71 # deployments 

72 deployments: List[DeploymentConfig] = Field(default_factory=list) 1a

73 

74 @staticmethod 1a

75 def _validate_action_steps(steps: Optional[List[Dict[str, Any]]]) -> None: 1a

76 # Light validation: allow any mapping; prefer single-key style but do not enforce 

77 if not steps: 

78 return 

79 for step in steps: 

80 if not isinstance(step, dict): 

81 raise TypeError("Each action step must be a mapping") 

82 # empty or multi-key steps will be passed through unchanged 

83 

84 @field_validator("build", "push", "pull") 1a

85 @classmethod 1a

86 def _validate_actions(cls, v: Optional[List[Dict[str, Any]]]): 1a

87 cls._validate_action_steps(v) 

88 return v 

89 

90 @field_validator("deployments") 1a

91 @classmethod 1a

92 def _validate_deployments(cls, v: List[DeploymentConfig]): 1a

93 # Ensure deployments is a list 

94 return v or [] 

95 

96 

97class ConcurrencyLimitSpec(BaseModel): 1a

98 model_config = ConfigDict(extra="ignore") 1a

99 

100 limit: Optional[int] = None 1a

101 collision_strategy: Optional[str] = None 1a

102 

103 

104class RawScheduleConfig(BaseModel): 1a

105 """ 

106 Strongly-typed schedule config that mirrors the CLI's accepted YAML shape. 

107 Exactly one of cron, interval, or rrule must be provided. 

108 """ 

109 

110 model_config = ConfigDict(extra="forbid") 1a

111 

112 # One-of schedule selectors 

113 cron: Optional[str] = None 1a

114 interval: Optional[int] = None 1a

115 rrule: Optional[str] = None 1a

116 

117 # Common extras 

118 timezone: Optional[str] = None 1a

119 anchor_date: Optional[str] = None 1a

120 active: Optional[Union[bool, str]] = None # Allow string for template values 1a

121 parameters: Dict[str, Any] = Field(default_factory=dict) 1a

122 slug: Optional[str] = None 1a

123 

124 # Cron-specific 

125 day_or: Optional[Union[bool, str]] = None # Allow string for template values 1a

126 

127 @model_validator(mode="after") 1a

128 def _one_of_schedule(self): 1a

129 provided = [v is not None for v in (self.cron, self.interval, self.rrule)] 

130 if sum(provided) != 1: 

131 raise ValueError( 

132 "Exactly one of 'cron', 'interval', or 'rrule' must be provided" 

133 ) 

134 return self 

135 

136 

137ScheduleItem = Union[ 1a

138 RawScheduleConfig, DeploymentScheduleCreate, SCHEDULE_TYPES, Dict[None, None] 

139]