Add type hints to BaseInstrumentor (#3084)

This commit is contained in:
Marcelo Trylesinski 2024-12-11 10:37:51 +01:00 committed by GitHub
parent ecf5529f99
commit 7804e0a4e8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 18 additions and 14 deletions

View File

@ -12,8 +12,10 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
from __future__ import annotations
from logging import getLogger from logging import getLogger
from typing import Collection, Optional, Union from typing import Collection
from packaging.requirements import InvalidRequirement, Requirement from packaging.requirements import InvalidRequirement, Requirement
@ -27,10 +29,10 @@ logger = getLogger(__name__)
class DependencyConflict: class DependencyConflict:
required: str = None required: str | None = None
found: Optional[str] = None found: str | None = None
def __init__(self, required, found=None): def __init__(self, required: str | None, found: str | None = None):
self.required = required self.required = required
self.found = found self.found = found
@ -40,7 +42,7 @@ class DependencyConflict:
def get_dist_dependency_conflicts( def get_dist_dependency_conflicts(
dist: Distribution, dist: Distribution,
) -> Optional[DependencyConflict]: ) -> DependencyConflict | None:
instrumentation_deps = [] instrumentation_deps = []
extra = "extra" extra = "extra"
instruments = "instruments" instruments = "instruments"
@ -57,8 +59,8 @@ def get_dist_dependency_conflicts(
def get_dependency_conflicts( def get_dependency_conflicts(
deps: Collection[Union[str, Requirement]], deps: Collection[str | Requirement],
) -> Optional[DependencyConflict]: ) -> DependencyConflict | None:
for dep in deps: for dep in deps:
if isinstance(dep, Requirement): if isinstance(dep, Requirement):
req = dep req = dep

View File

@ -17,9 +17,11 @@
OpenTelemetry Base Instrumentor OpenTelemetry Base Instrumentor
""" """
from __future__ import annotations
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from logging import getLogger from logging import getLogger
from typing import Collection, Optional from typing import Any, Collection
from opentelemetry.instrumentation._semconv import ( from opentelemetry.instrumentation._semconv import (
_OpenTelemetrySemanticConventionStability, _OpenTelemetrySemanticConventionStability,
@ -33,7 +35,7 @@ _LOG = getLogger(__name__)
class BaseInstrumentor(ABC): class BaseInstrumentor(ABC):
"""An ABC for instrumentors """An ABC for instrumentors.
Child classes of this ABC should instrument specific third Child classes of this ABC should instrument specific third
party libraries or frameworks either by using the party libraries or frameworks either by using the
@ -74,18 +76,18 @@ class BaseInstrumentor(ABC):
is present in the environment. is present in the environment.
""" """
def _instrument(self, **kwargs): def _instrument(self, **kwargs: Any):
"""Instrument the library""" """Instrument the library"""
@abstractmethod @abstractmethod
def _uninstrument(self, **kwargs): def _uninstrument(self, **kwargs: Any):
"""Uninstrument the library""" """Uninstrument the library"""
def _check_dependency_conflicts(self) -> Optional[DependencyConflict]: def _check_dependency_conflicts(self) -> DependencyConflict | None:
dependencies = self.instrumentation_dependencies() dependencies = self.instrumentation_dependencies()
return get_dependency_conflicts(dependencies) return get_dependency_conflicts(dependencies)
def instrument(self, **kwargs): def instrument(self, **kwargs: Any):
"""Instrument the library """Instrument the library
This method will be called without any optional arguments by the This method will be called without any optional arguments by the
@ -117,7 +119,7 @@ class BaseInstrumentor(ABC):
self._is_instrumented_by_opentelemetry = True self._is_instrumented_by_opentelemetry = True
return result return result
def uninstrument(self, **kwargs): def uninstrument(self, **kwargs: Any):
"""Uninstrument the library """Uninstrument the library
See ``BaseInstrumentor.instrument`` for more information regarding the See ``BaseInstrumentor.instrument`` for more information regarding the