Coverage for /usr/local/lib/python3.12/site-packages/prefect/client/orchestration/_variables/client.py: 21%

1from __future__ import annotations 1 ctx1a

2 

3from typing import TYPE_CHECKING 1 ctx1a

4 

5import httpx 1 ctx1a

6 

7from prefect.client.orchestration.base import BaseAsyncClient, BaseClient 1 ctx1a

8from prefect.exceptions import ObjectAlreadyExists, ObjectNotFound 1 ctx1a

9 

10if TYPE_CHECKING: 10 ↛ 11line 10 didn't jump to line 11 because the condition on line 10 was never true1 ctx1a

11 from prefect.client.schemas.actions import ( 

12 VariableCreate, 

13 VariableUpdate, 

14 ) 

15 from prefect.client.schemas.objects import ( 

16 Variable, 

17 ) 

18 

19 

20class VariableClient(BaseClient): 1 ctx1a

21 def create_variable(self, variable: "VariableCreate") -> "Variable": 1 ctx1a

22 """ 

23 Creates an variable with the provided configuration. 

24 

25 Args: 

26 variable: Desired configuration for the new variable. 

27 Returns: 

28 Information about the newly created variable. 

29 """ 

30 try: 

31 response = self.request( 

32 "POST", 

33 "/variables/", 

34 json=variable.model_dump(mode="json", exclude_unset=True), 

35 ) 

36 except httpx.HTTPStatusError as e: 

37 if e.response.status_code == 409: 

38 raise ObjectAlreadyExists(http_exc=e) from e 

39 else: 

40 raise 

41 

42 from prefect.client.schemas.objects import Variable 

43 

44 return Variable.model_validate(response.json()) 

45 

46 def read_variable_by_name(self, name: str) -> "Variable | None": 1 ctx1a

47 """Reads a variable by name. Returns None if no variable is found.""" 

48 try: 

49 response = self.request( 

50 "GET", "/variables/name/{name}", path_params={"name": name} 

51 ) 

52 from prefect.client.schemas.objects import Variable 

53 

54 return Variable(**response.json()) 

55 except httpx.HTTPStatusError as e: 

56 if e.response.status_code == 404: 

57 return None 

58 else: 

59 raise 

60 

61 def read_variables(self, limit: int | None = None) -> list["Variable"]: 1 ctx1a

62 """Reads all variables.""" 

63 response = self.request("POST", "/variables/filter", json={"limit": limit}) 

64 from prefect.client.schemas.objects import Variable 

65 

66 return Variable.model_validate_list(response.json()) 

67 

68 def update_variable(self, variable: "VariableUpdate") -> None: 1 ctx1a

69 """ 

70 Updates a variable with the provided configuration. 

71 

72 Args: 

73 variable: Desired configuration for the updated variable. 

74 Returns: 

75 Information about the updated variable. 

76 """ 

77 self._client.patch( 

78 f"/variables/name/{variable.name}", 

79 json=variable.model_dump(mode="json", exclude_unset=True), 

80 ) 

81 return None 

82 

83 def delete_variable_by_name(self, name: str) -> None: 1 ctx1a

84 """Deletes a variable by name.""" 

85 try: 

86 self.request( 

87 "DELETE", 

88 "/variables/name/{name}", 

89 path_params={"name": name}, 

90 ) 

91 return None 

92 except httpx.HTTPStatusError as e: 

93 if e.response.status_code == 404: 

94 raise ObjectNotFound(http_exc=e) from e 

95 else: 

96 raise 

97 

98 

99class VariableAsyncClient(BaseAsyncClient): 1 ctx1a

100 async def create_variable(self, variable: "VariableCreate") -> "Variable": 1 ctx1a

101 """Creates a variable with the provided configuration.""" 

102 try: 

103 response = await self.request( 

104 "POST", 

105 "/variables/", 

106 json=variable.model_dump(mode="json", exclude_unset=True), 

107 ) 

108 except httpx.HTTPStatusError as e: 

109 if e.response.status_code == 409: 

110 raise ObjectAlreadyExists(http_exc=e) from e 

111 else: 

112 raise 

113 

114 from prefect.client.schemas.objects import Variable 

115 

116 return Variable.model_validate(response.json()) 

117 

118 async def read_variable_by_name(self, name: str) -> "Variable | None": 1 ctx1a

119 """Reads a variable by name. Returns None if no variable is found.""" 

120 try: 

121 response = await self.request( 

122 "GET", 

123 "/variables/name/{name}", 

124 path_params={"name": name}, 

125 ) 

126 from prefect.client.schemas.objects import Variable 

127 

128 return Variable.model_validate(response.json()) 

129 except httpx.HTTPStatusError as e: 

130 if e.response.status_code == 404: 

131 return None 

132 else: 

133 raise 

134 

135 async def read_variables(self, limit: int | None = None) -> list["Variable"]: 1 ctx1a

136 """Reads all variables.""" 

137 response = await self.request( 

138 "POST", "/variables/filter", json={"limit": limit} 

139 ) 

140 from prefect.client.schemas.objects import Variable 

141 

142 return Variable.model_validate_list(response.json()) 

143 

144 async def update_variable(self, variable: "VariableUpdate") -> None: 1 ctx1a

145 """ 

146 Updates a variable with the provided configuration. 

147 

148 Args: 

149 variable: Desired configuration for the updated variable. 

150 Returns: 

151 Information about the updated variable. 

152 """ 

153 await self.request( 

154 "PATCH", 

155 "/variables/name/{name}", 

156 path_params={"name": variable.name}, 

157 json=variable.model_dump(mode="json", exclude_unset=True), 

158 ) 

159 return None 

160 

161 async def delete_variable_by_name(self, name: str) -> None: 1 ctx1a

162 """Deletes a variable by name.""" 

163 try: 

164 await self.request( 

165 "DELETE", 

166 "/variables/name/{name}", 

167 path_params={"name": name}, 

168 ) 

169 except httpx.HTTPStatusError as e: 

170 if e.response.status_code == 404: 

171 raise ObjectNotFound(http_exc=e) from e 

172 else: 

173 raise