|
1 | | -from functools import lru_cache |
2 | | -from importlib import import_module |
3 | 1 | from typing import TYPE_CHECKING |
4 | 2 |
|
5 | | -from sentry_sdk.integrations.boto3._services.base import _ServiceExtension |
6 | | -from sentry_sdk.utils import capture_internal_exceptions |
7 | | - |
8 | 3 | if TYPE_CHECKING: |
9 | | - from typing import Dict, Optional, Tuple |
| 4 | + from typing import Dict, Optional |
| 5 | + |
| 6 | + from sentry_sdk.integrations.boto3._services.base import _ServiceExtension |
10 | 7 |
|
11 | 8 |
|
12 | | -# service modules are imported lazily. |
13 | | -# e.g. `s3` -> (`sentry_sdk.integrations.boto3._services.s3`, `_S3Extension) |
14 | | -_SERVICE_EXTENSIONS: "Dict[str, Tuple[str, str]]" = {} |
| 9 | +# when py 3.15 drops, we might want to take a look at using |
| 10 | +# a lazy-loading approach using the new `lazy` keyword. |
| 11 | +# e.g. {"s3": _S3Extension} |
| 12 | +_SERVICE_EXTENSIONS: "Dict[str, _ServiceExtension]" = {} |
15 | 13 |
|
16 | 14 |
|
17 | | -@lru_cache(maxsize=None) |
18 | 15 | def _resolve_service( |
19 | | - service: "str", |
| 16 | + service_name: "str", |
20 | 17 | ) -> "Optional[_ServiceExtension]": |
21 | | - target = _SERVICE_EXTENSIONS.get(service) |
22 | | - if target is None: |
23 | | - return None |
24 | | - |
25 | | - with capture_internal_exceptions(): |
26 | | - module_name, class_name = target |
27 | | - extension_class = getattr(import_module(module_name), class_name) |
28 | | - candidate = extension_class() |
29 | | - if isinstance(candidate, _ServiceExtension): |
30 | | - return candidate |
31 | | - |
| 18 | + if service_name in _SERVICE_EXTENSIONS: |
| 19 | + return _SERVICE_EXTENSIONS[service_name] |
32 | 20 | # preserve generic instrumentation when lookup fails. |
33 | 21 | return None |
0 commit comments