Coverage for /usr/local/lib/python3.12/site-packages/prefect/client/orchestration/_blocks_documents/client.py: 18%
85 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 10:48 +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 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 BlockDocumentCreate,
15 BlockDocumentUpdate,
16 )
17 from prefect.client.schemas.objects import (
18 BlockDocument,
19 )
22class BlocksDocumentClient(BaseClient): 1a
23 def create_block_document( 1a
24 self,
25 block_document: "BlockDocument | BlockDocumentCreate",
26 include_secrets: bool = True,
27 ) -> "BlockDocument":
28 """
29 Create a block document in the Prefect API. This data is used to configure a
30 corresponding Block.
32 Args:
33 include_secrets (bool): whether to include secret values
34 on the stored Block, corresponding to Pydantic's `SecretStr` and
35 `SecretBytes` fields. Note Blocks may not work as expected if
36 this is set to `False`.
37 """
38 block_document_data = block_document.model_dump(
39 mode="json",
40 exclude_unset=True,
41 exclude={"id", "block_schema", "block_type"},
42 context={"include_secrets": include_secrets},
43 serialize_as_any=True,
44 )
45 try:
46 response = self.request(
47 "POST",
48 "/block_documents/",
49 json=block_document_data,
50 )
51 except HTTPStatusError as e:
52 if e.response.status_code == 409:
53 raise ObjectAlreadyExists(http_exc=e) from e
54 else:
55 raise
56 from prefect.client.schemas.objects import BlockDocument
58 return BlockDocument.model_validate(response.json())
60 def update_block_document( 1a
61 self,
62 block_document_id: "UUID",
63 block_document: "BlockDocumentUpdate",
64 ) -> None:
65 """
66 Update a block document in the Prefect API.
67 """
68 try:
69 self.request(
70 "PATCH",
71 "/block_documents/{id}",
72 path_params={"id": block_document_id},
73 json=block_document.model_dump(
74 mode="json",
75 exclude_unset=True,
76 include={"data", "merge_existing_data", "block_schema_id"},
77 ),
78 )
79 except HTTPStatusError as e:
80 if e.response.status_code == 404:
81 raise ObjectNotFound(http_exc=e) from e
82 else:
83 raise
85 def delete_block_document(self, block_document_id: "UUID") -> None: 1a
86 """
87 Delete a block document.
88 """
89 try:
90 self.request(
91 "DELETE",
92 "/block_documents/{id}",
93 path_params={"id": block_document_id},
94 )
95 except HTTPStatusError as e:
96 if e.response.status_code == 404:
97 raise ObjectNotFound(http_exc=e) from e
98 else:
99 raise
101 def read_block_document( 1a
102 self,
103 block_document_id: "UUID",
104 include_secrets: bool = True,
105 ) -> "BlockDocument":
106 """
107 Read the block document with the specified ID.
109 Args:
110 block_document_id: the block document id
111 include_secrets (bool): whether to include secret values
112 on the Block, corresponding to Pydantic's `SecretStr` and
113 `SecretBytes` fields. These fields are automatically obfuscated
114 by Pydantic, but users can additionally choose not to receive
115 their values from the API. Note that any business logic on the
116 Block may not work if this is `False`.
118 Raises:
119 httpx.RequestError: if the block document was not found for any reason
121 Returns:
122 A block document or None.
123 """
125 try:
126 response = self.request(
127 "GET",
128 "/block_documents/{id}",
129 path_params={"id": block_document_id},
130 params=dict(include_secrets=include_secrets),
131 )
132 except HTTPStatusError as e:
133 if e.response.status_code == 404:
134 raise ObjectNotFound(http_exc=e) from e
135 else:
136 raise
137 from prefect.client.schemas.objects import BlockDocument
139 return BlockDocument.model_validate(response.json())
141 def read_block_documents( 1a
142 self,
143 block_schema_type: str | None = None,
144 offset: int | None = None,
145 limit: int | None = None,
146 include_secrets: bool = True,
147 ) -> "list[BlockDocument]":
148 """
149 Read block documents
151 Args:
152 block_schema_type: an optional block schema type
153 offset: an offset
154 limit: the number of blocks to return
155 include_secrets (bool): whether to include secret values
156 on the Block, corresponding to Pydantic's `SecretStr` and
157 `SecretBytes` fields. These fields are automatically obfuscated
158 by Pydantic, but users can additionally choose not to receive
159 their values from the API. Note that any business logic on the
160 Block may not work if this is `False`.
162 Returns:
163 A list of block documents
164 """
165 response = self.request(
166 "POST",
167 "/block_documents/filter",
168 json=dict(
169 block_schema_type=block_schema_type,
170 offset=offset,
171 limit=limit,
172 include_secrets=include_secrets,
173 ),
174 )
175 from prefect.client.schemas.objects import BlockDocument
177 return BlockDocument.model_validate_list(response.json())
180class BlocksDocumentAsyncClient(BaseAsyncClient): 1a
181 async def create_block_document( 1a
182 self,
183 block_document: "BlockDocument | BlockDocumentCreate",
184 include_secrets: bool = True,
185 ) -> "BlockDocument":
186 """
187 Create a block document in the Prefect API. This data is used to configure a
188 corresponding Block.
190 Args:
191 include_secrets (bool): whether to include secret values
192 on the stored Block, corresponding to Pydantic's `SecretStr` and
193 `SecretBytes` fields. Note Blocks may not work as expected if
194 this is set to `False`.
195 """
196 block_document_data = block_document.model_dump(
197 mode="json",
198 exclude_unset=True,
199 exclude={"id", "block_schema", "block_type"},
200 context={"include_secrets": include_secrets},
201 serialize_as_any=True,
202 )
203 try:
204 response = await self.request(
205 "POST",
206 "/block_documents/",
207 json=block_document_data,
208 )
209 except HTTPStatusError as e:
210 if e.response.status_code == 409:
211 raise ObjectAlreadyExists(http_exc=e) from e
212 else:
213 raise
214 from prefect.client.schemas.objects import BlockDocument
216 return BlockDocument.model_validate(response.json())
218 async def update_block_document( 1a
219 self,
220 block_document_id: "UUID",
221 block_document: "BlockDocumentUpdate",
222 ) -> None:
223 """
224 Update a block document in the Prefect API.
225 """
226 try:
227 await self.request(
228 "PATCH",
229 "/block_documents/{id}",
230 path_params={"id": block_document_id},
231 json=block_document.model_dump(
232 mode="json",
233 exclude_unset=True,
234 include={"data", "merge_existing_data", "block_schema_id"},
235 ),
236 )
237 except HTTPStatusError as e:
238 if e.response.status_code == 404:
239 raise ObjectNotFound(http_exc=e) from e
240 else:
241 raise
243 async def delete_block_document(self, block_document_id: "UUID") -> None: 1a
244 """
245 Delete a block document.
246 """
247 try:
248 await self.request(
249 "DELETE",
250 "/block_documents/{id}",
251 path_params={"id": block_document_id},
252 )
253 except HTTPStatusError as e:
254 if e.response.status_code == 404:
255 raise ObjectNotFound(http_exc=e) from e
256 else:
257 raise
259 async def read_block_document( 1a
260 self,
261 block_document_id: "UUID",
262 include_secrets: bool = True,
263 ) -> "BlockDocument":
264 """
265 Read the block document with the specified ID.
267 Args:
268 block_document_id: the block document id
269 include_secrets (bool): whether to include secret values
270 on the Block, corresponding to Pydantic's `SecretStr` and
271 `SecretBytes` fields. These fields are automatically obfuscated
272 by Pydantic, but users can additionally choose not to receive
273 their values from the API. Note that any business logic on the
274 Block may not work if this is `False`.
276 Raises:
277 httpx.RequestError: if the block document was not found for any reason
279 Returns:
280 A block document or None.
281 """
282 try:
283 response = await self.request(
284 "GET",
285 "/block_documents/{id}",
286 path_params={"id": block_document_id},
287 params=dict(include_secrets=include_secrets),
288 )
289 except HTTPStatusError as e:
290 if e.response.status_code == 404:
291 raise ObjectNotFound(http_exc=e) from e
292 else:
293 raise
294 from prefect.client.schemas.objects import BlockDocument
296 return BlockDocument.model_validate(response.json())
298 async def read_block_documents( 1a
299 self,
300 block_schema_type: str | None = None,
301 offset: int | None = None,
302 limit: int | None = None,
303 include_secrets: bool = True,
304 ) -> "list[BlockDocument]":
305 """
306 Read block documents
308 Args:
309 block_schema_type: an optional block schema type
310 offset: an offset
311 limit: the number of blocks to return
312 include_secrets (bool): whether to include secret values
313 on the Block, corresponding to Pydantic's `SecretStr` and
314 `SecretBytes` fields. These fields are automatically obfuscated
315 by Pydantic, but users can additionally choose not to receive
316 their values from the API. Note that any business logic on the
317 Block may not work if this is `False`.
319 Returns:
320 A list of block documents
321 """
322 response = await self.request(
323 "POST",
324 "/block_documents/filter",
325 json=dict(
326 block_schema_type=block_schema_type,
327 offset=offset,
328 limit=limit,
329 include_secrets=include_secrets,
330 ),
331 )
332 from prefect.client.schemas.objects import BlockDocument
334 return BlockDocument.model_validate_list(response.json())