|
1 | 1 | """Utility functions for PydanticAI span instrumentation.""" |
2 | 2 |
|
3 | 3 | import sentry_sdk |
4 | | -from sentry_sdk._types import BLOB_DATA_SUBSTITUTE |
5 | | -from sentry_sdk.ai.utils import get_modality_from_mime_type |
6 | 4 | from sentry_sdk.consts import SPANDATA |
7 | 5 |
|
8 | | -from sentry_sdk.ai.consts import DATA_URL_BASE64_REGEX |
9 | | - |
10 | 6 | from typing import TYPE_CHECKING |
11 | 7 |
|
12 | 8 | if TYPE_CHECKING: |
13 | | - from typing import Union, Dict, Any |
| 9 | + from typing import Union |
14 | 10 | from pydantic_ai.usage import RequestUsage, RunUsage # type: ignore |
15 | 11 |
|
16 | 12 |
|
17 | | -def _serialize_image_url_item(item: "Any") -> "Dict[str, Any]": |
18 | | - """Serialize an ImageUrl content item for span data. |
19 | | -
|
20 | | - For data URLs containing base64-encoded images, the content is redacted. |
21 | | - For regular HTTP URLs, the URL string is preserved. |
22 | | - """ |
23 | | - url = str(item.url) |
24 | | - data_url_match = DATA_URL_BASE64_REGEX.match(url) |
25 | | - |
26 | | - if data_url_match: |
27 | | - return { |
28 | | - "type": "image", |
29 | | - "content": BLOB_DATA_SUBSTITUTE, |
30 | | - } |
31 | | - |
32 | | - return { |
33 | | - "type": "image", |
34 | | - "content": url, |
35 | | - } |
36 | | - |
37 | | - |
38 | | -def _serialize_binary_content_item(item: "Any") -> "Dict[str, Any]": |
39 | | - """Serialize a BinaryContent item for span data, redacting the blob data.""" |
40 | | - return { |
41 | | - "type": "blob", |
42 | | - "modality": get_modality_from_mime_type(item.media_type), |
43 | | - "mime_type": item.media_type, |
44 | | - "content": BLOB_DATA_SUBSTITUTE, |
45 | | - } |
46 | | - |
47 | | - |
48 | 13 | def _set_usage_data( |
49 | 14 | span: "sentry_sdk.tracing.Span", usage: "Union[RequestUsage, RunUsage]" |
50 | 15 | ) -> None: |
|
0 commit comments