From 23483c0a454e95684e04633a0a886528665a2c8c Mon Sep 17 00:00:00 2001 From: Stevie Gayet Date: Fri, 12 May 2023 18:36:59 +0200 Subject: [PATCH] chore(typing): annotate utils/compat.py --- kombu/utils/compat.py | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/kombu/utils/compat.py b/kombu/utils/compat.py index e1b22f665..79813bf12 100644 --- a/kombu/utils/compat.py +++ b/kombu/utils/compat.py @@ -6,6 +6,7 @@ import sys from contextlib import contextmanager from functools import wraps +from typing import TYPE_CHECKING try: from importlib import metadata as importlib_metadata @@ -17,6 +18,11 @@ from kombu.exceptions import reraise +if TYPE_CHECKING: + from importlib.metadata import EntryPoint + from io import StringIO + from typing import Any, Callable, Generator, Optional, Tuple, Union + FILENO_ERRORS = (AttributeError, ValueError, UnsupportedOperation) try: @@ -31,7 +37,7 @@ _environment = None -def coro(gen): +def coro(gen: Generator) -> Callable: """Decorator to mark generator as co-routine.""" @wraps(gen) def wind_up(*args, **kwargs): @@ -41,7 +47,7 @@ def wind_up(*args, **kwargs): return wind_up -def _detect_environment(): +def _detect_environment() -> str: # ## -eventlet- if 'eventlet' in sys.modules: try: @@ -69,7 +75,7 @@ def _detect_environment(): return 'default' -def detect_environment(): +def detect_environment() -> str: """Detect the current environment: default, eventlet, or gevent.""" global _environment if _environment is None: @@ -77,7 +83,7 @@ def detect_environment(): return _environment -def entrypoints(namespace): +def entrypoints(namespace: str) -> Tuple[Tuple[EntryPoint, Any]]: """Return setuptools entrypoints for namespace.""" if sys.version_info >= (3,10): entry_points = importlib_metadata.entry_points(group=namespace) @@ -94,19 +100,19 @@ def entrypoints(namespace): ) -def fileno(f): +def fileno(f: Union[numbers.Integral, StringIO]) -> numbers.Integral | int: """Get fileno from file-like object.""" if isinstance(f, numbers.Integral): return f return f.fileno() -def maybe_fileno(f): +def maybe_fileno(f) -> Optional[numbers.Integral | int]: """Get object fileno, or :const:`None` if not defined.""" try: return fileno(f) except FILENO_ERRORS: - pass + return None @contextmanager