Coverage for /usr/local/lib/python3.12/site-packages/prefect/client/orchestration/_blocks_types/client.py: 16%
115 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +0000
1from __future__ import annotations 1a
3from typing import TYPE_CHECKING 1a
5from httpx import HTTPStatusError 1a
7from prefect.client.orchestration.base import BaseAsyncClient, BaseClient 1a
8from prefect.exceptions import ObjectAlreadyExists, ObjectNotFound, ProtectedBlockError 1a
10if TYPE_CHECKING: 10 ↛ 11line 10 didn't jump to line 11 because the condition on line 10 was never true1a
11 from uuid import UUID
13 from prefect.client.schemas.actions import (
14 BlockTypeCreate,
15 BlockTypeUpdate,
16 )
17 from prefect.client.schemas.objects import (
18 BlockDocument,
19 BlockType,
20 )
23class BlocksTypeClient(BaseClient): 1a
24 def read_block_documents_by_type( 1a
25 self,
26 block_type_slug: str,
27 offset: int | None = None,
28 limit: int | None = None,
29 include_secrets: bool = True,
30 ) -> "list[BlockDocument]":
31 """Retrieve block documents by block type slug.
33 Args:
34 block_type_slug: The block type slug.
35 offset: an offset
36 limit: the number of blocks to return
37 include_secrets: whether to include secret values
39 Returns:
40 A list of block documents
41 """
42 response = self.request(
43 "GET",
44 "/block_types/slug/{slug}/block_documents",
45 path_params={"slug": block_type_slug},
46 params=dict(
47 offset=offset,
48 limit=limit,
49 include_secrets=include_secrets,
50 ),
51 )
53 from prefect.client.schemas.objects import BlockDocument
55 return BlockDocument.model_validate_list(response.json())
57 def create_block_type(self, block_type: "BlockTypeCreate") -> "BlockType": 1a
58 """
59 Create a block type in the Prefect API.
60 """
61 try:
62 response = self.request(
63 "POST",
64 "/block_types/",
65 json=block_type.model_dump(
66 mode="json", exclude_unset=True, exclude={"id"}
67 ),
68 )
69 except HTTPStatusError as e:
70 if e.response.status_code == 409:
71 raise ObjectAlreadyExists(http_exc=e) from e
72 else:
73 raise
74 from prefect.client.schemas.objects import BlockType
76 return BlockType.model_validate(response.json())
78 def read_block_type_by_slug(self, slug: str) -> "BlockType": 1a
79 """
80 Read a block type by its slug.
81 """
82 try:
83 response = self.request(
84 "GET",
85 "/block_types/slug/{slug}",
86 path_params={"slug": slug},
87 )
88 except HTTPStatusError as e:
89 if e.response.status_code == 404:
90 raise ObjectNotFound(http_exc=e) from e
91 else:
92 raise
93 from prefect.client.schemas.objects import BlockType
95 return BlockType.model_validate(response.json())
97 def update_block_type( 1a
98 self, block_type_id: "UUID", block_type: "BlockTypeUpdate"
99 ) -> None:
100 """
101 Update a block document in the Prefect API.
102 """
103 from prefect.client.schemas.actions import BlockTypeUpdate
105 try:
106 self.request(
107 "PATCH",
108 "/block_types/{id}",
109 path_params={"id": block_type_id},
110 json=block_type.model_dump(
111 mode="json",
112 exclude_unset=True,
113 include=BlockTypeUpdate.updatable_fields(),
114 ),
115 )
116 except HTTPStatusError as e:
117 if e.response.status_code == 404:
118 raise ObjectNotFound(http_exc=e) from e
119 else:
120 raise
122 def delete_block_type(self, block_type_id: "UUID") -> None: 1a
123 """
124 Delete a block type.
125 """
126 try:
127 self.request(
128 "DELETE",
129 "/block_types/{id}",
130 path_params={"id": block_type_id},
131 )
132 except HTTPStatusError as e:
133 if e.response.status_code == 404:
134 raise ObjectNotFound(http_exc=e) from e
135 elif (
136 e.response.status_code == 403
137 and e.response.json()["detail"]
138 == "protected block types cannot be deleted."
139 ):
140 raise ProtectedBlockError(
141 "Protected block types cannot be deleted."
142 ) from e
143 else:
144 raise
146 def read_block_types(self) -> "list[BlockType]": 1a
147 """
148 Read all block types
149 Raises:
150 httpx.RequestError: if the block types were not found
152 Returns:
153 List of BlockTypes.
154 """
155 response = self.request("POST", "/block_types/filter", json={})
156 from prefect.client.schemas.objects import BlockType
158 return BlockType.model_validate_list(response.json())
160 def read_block_document_by_name( 1a
161 self,
162 name: str,
163 block_type_slug: str,
164 include_secrets: bool = True,
165 ) -> "BlockDocument":
166 """
167 Read the block document with the specified name that corresponds to a
168 specific block type name.
170 Args:
171 name: The block document name.
172 block_type_slug: The block type slug.
173 include_secrets (bool): whether to include secret values
174 on the Block, corresponding to Pydantic's `SecretStr` and
175 `SecretBytes` fields. These fields are automatically obfuscated
176 by Pydantic, but users can additionally choose not to receive
177 their values from the API. Note that any business logic on the
178 Block may not work if this is `False`.
180 Raises:
181 httpx.RequestError: if the block document was not found for any reason
183 Returns:
184 A block document or None.
185 """
186 try:
187 response = self.request(
188 "GET",
189 "/block_types/slug/{slug}/block_documents/name/{block_document_name}",
190 path_params={"slug": block_type_slug, "block_document_name": name},
191 params=dict(include_secrets=include_secrets),
192 )
193 except HTTPStatusError as e:
194 if e.response.status_code == 404:
195 raise ObjectNotFound(http_exc=e) from e
196 else:
197 raise
198 from prefect.client.schemas.objects import BlockDocument
200 return BlockDocument.model_validate(response.json())
203class BlocksTypeAsyncClient(BaseAsyncClient): 1a
204 async def read_block_documents_by_type( 1a
205 self,
206 block_type_slug: str,
207 offset: int | None = None,
208 limit: int | None = None,
209 include_secrets: bool = True,
210 ) -> "list[BlockDocument]":
211 """Retrieve block documents by block type slug.
213 Args:
214 block_type_slug: The block type slug.
215 offset: an offset
216 limit: the number of blocks to return
217 include_secrets: whether to include secret values
219 Returns:
220 A list of block documents
221 """
222 response = await self.request(
223 "GET",
224 "/block_types/slug/{slug}/block_documents",
225 path_params={"slug": block_type_slug},
226 params=dict(
227 offset=offset,
228 limit=limit,
229 include_secrets=include_secrets,
230 ),
231 )
233 from prefect.client.schemas.objects import BlockDocument
235 return BlockDocument.model_validate_list(response.json())
237 async def create_block_type(self, block_type: "BlockTypeCreate") -> "BlockType": 1a
238 """
239 Create a block type in the Prefect API.
240 """
241 try:
242 response = await self.request(
243 "POST",
244 "/block_types/",
245 json=block_type.model_dump(
246 mode="json", exclude_unset=True, exclude={"id"}
247 ),
248 )
249 except HTTPStatusError as e:
250 if e.response.status_code == 409:
251 raise ObjectAlreadyExists(http_exc=e) from e
252 else:
253 raise
254 from prefect.client.schemas.objects import BlockType
256 return BlockType.model_validate(response.json())
258 async def read_block_type_by_slug(self, slug: str) -> "BlockType": 1a
259 """
260 Read a block type by its slug.
261 """
262 try:
263 response = await self.request(
264 "GET",
265 "/block_types/slug/{slug}",
266 path_params={"slug": slug},
267 )
268 except HTTPStatusError as e:
269 if e.response.status_code == 404:
270 raise ObjectNotFound(http_exc=e) from e
271 else:
272 raise
273 from prefect.client.schemas.objects import BlockType
275 return BlockType.model_validate(response.json())
277 async def update_block_type( 1a
278 self, block_type_id: "UUID", block_type: "BlockTypeUpdate"
279 ) -> None:
280 """
281 Update a block document in the Prefect API.
282 """
283 from prefect.client.schemas.actions import BlockTypeUpdate
285 try:
286 await self.request(
287 "PATCH",
288 "/block_types/{id}",
289 path_params={"id": block_type_id},
290 json=block_type.model_dump(
291 mode="json",
292 exclude_unset=True,
293 include=BlockTypeUpdate.updatable_fields(),
294 ),
295 )
296 except HTTPStatusError as e:
297 if e.response.status_code == 404:
298 raise ObjectNotFound(http_exc=e) from e
299 else:
300 raise
302 async def delete_block_type(self, block_type_id: "UUID") -> None: 1a
303 """
304 Delete a block type.
305 """
306 try:
307 await self.request(
308 "DELETE",
309 "/block_types/{id}",
310 path_params={"id": block_type_id},
311 )
312 except HTTPStatusError as e:
313 if e.response.status_code == 404:
314 raise ObjectNotFound(http_exc=e) from e
315 elif (
316 e.response.status_code == 403
317 and e.response.json()["detail"]
318 == "protected block types cannot be deleted."
319 ):
320 raise ProtectedBlockError(
321 "Protected block types cannot be deleted."
322 ) from e
323 else:
324 raise
326 async def read_block_types(self) -> "list[BlockType]": 1a
327 """
328 Read all block types
329 Raises:
330 httpx.RequestError: if the block types were not found
332 Returns:
333 List of BlockTypes.
334 """
335 response = await self.request("POST", "/block_types/filter", json={})
336 from prefect.client.schemas.objects import BlockType
338 return BlockType.model_validate_list(response.json())
340 async def read_block_document_by_name( 1a
341 self,
342 name: str,
343 block_type_slug: str,
344 include_secrets: bool = True,
345 ) -> "BlockDocument":
346 """
347 Read the block document with the specified name that corresponds to a
348 specific block type name.
350 Args:
351 name: The block document name.
352 block_type_slug: The block type slug.
353 include_secrets (bool): whether to include secret values
354 on the Block, corresponding to Pydantic's `SecretStr` and
355 `SecretBytes` fields. These fields are automatically obfuscated
356 by Pydantic, but users can additionally choose not to receive
357 their values from the API. Note that any business logic on the
358 Block may not work if this is `False`.
360 Raises:
361 httpx.RequestError: if the block document was not found for any reason
363 Returns:
364 A block document or None.
365 """
366 try:
367 response = await self.request(
368 "GET",
369 "/block_types/slug/{slug}/block_documents/name/{block_document_name}",
370 path_params={"slug": block_type_slug, "block_document_name": name},
371 params=dict(include_secrets=include_secrets),
372 )
373 except HTTPStatusError as e:
374 if e.response.status_code == 404:
375 raise ObjectNotFound(http_exc=e) from e
376 else:
377 raise
378 from prefect.client.schemas.objects import BlockDocument
380 return BlockDocument.model_validate(response.json())