mypy

Author: sanderegg

packages/service-library/src/servicelib/aiohttp/application_setup.py

@@ -17,19 +17,17 @@
 
 log = logging.getLogger(__name__)
 
-APP_SETUP_COMPLETED_KEY = f"{__name__ }.setup"
+APP_SETUP_COMPLETED_KEY = f"{__name__}.setup"
 
 
 class _SetupFunc(Protocol):
     __name__: str
 
-    def __call__(self, app: web.Application, *args: Any, **kwds: Any) -> bool:
-        ...
+    def __call__(self, app: web.Application, *args: Any, **kwds: Any) -> bool: ...
 
 
 class _ApplicationSettings(Protocol):
-    def is_enabled(self, field_name: str) -> bool:
-        ...
+    def is_enabled(self, field_name: str) -> bool: ...
 
 
 class ModuleCategory(Enum):
@@ -46,12 +44,10 @@ def __init__(self, *, reason) -> None:
         super().__init__(reason)
 
 
-class ApplicationSetupError(Exception):
-    ...
+class ApplicationSetupError(Exception): ...
 
 
-class DependencyError(ApplicationSetupError):
-    ...
+class DependencyError(ApplicationSetupError): ...
 
 
 class SetupMetadataDict(TypedDict):
@@ -91,9 +87,9 @@ def _is_addon_enabled_from_config(
     cfg: dict[str, Any], dotted_section: str, section
 ) -> bool:
     try:
-        parts: list[str] = dotted_section.split(".")
+        parts = dotted_section.split(".")
         # navigates app_config (cfg) searching for section
-        searched_config = deepcopy(cfg)
+        searched_config: Any = deepcopy(cfg)
         for part in parts:
             if section and part == "enabled":
                 # if section exists, no need to explicitly enable it
@@ -278,7 +274,7 @@ def _wrapper(app: web.Application, *args, **kargs) -> bool:
 
                 # post-setup
                 if completed is None:
-                    completed = True
+                    completed = True  # type: ignore[unreachable]
 
                 if completed:  # registers completed setup
                     app[APP_SETUP_COMPLETED_KEY].append(module_name)

packages/service-library/src/servicelib/aiohttp/docker_utils.py

@@ -97,4 +97,4 @@ async def retrieve_image_layer_information(
                 return TypeAdapter(DockerImageManifestsV2).validate_python(
                     json_response
                 )
-    return None
+    return None  # type: ignore[unreachable]

packages/service-library/src/servicelib/deferred_tasks/_deferred_manager.py

@@ -122,7 +122,6 @@ def __init__(
         max_workers: NonNegativeInt = _DEFAULT_DEFERRED_MANAGER_WORKER_SLOTS,
         delay_when_requeuing_message: timedelta = _DEFAULT_DELAY_BEFORE_NACK,
     ) -> None:
-
         self._task_tracker: BaseTaskTracker = RedisTaskTracker(scheduler_redis_sdk)
 
         self._worker_tracker = WorkerTracker(max_workers)
@@ -216,7 +215,7 @@ def un_patch_base_deferred_handlers(cls) -> None:
             )
 
             if isinstance(subclass.start, _PatchStartDeferred):
-                with log_context(
+                with log_context(  # type: ignore[unreachable]
                     _logger,
                     logging.DEBUG,
                     f"Remove `start` patch for {class_unique_reference}",
@@ -345,7 +344,6 @@ async def __get_task_schedule(
     async def _fs_handle_scheduled(  # pylint:disable=method-hidden
         self, task_uid: TaskUID
     ) -> None:
-
         _log_state(TaskState.SCHEDULED, task_uid)
 
         task_schedule = await self.__get_task_schedule(
@@ -425,7 +423,7 @@ async def _fs_handle_worker(  # pylint:disable=method-hidden
             await self.__publish_to_queue(task_uid, _FastStreamRabbitQueue.ERROR_RESULT)
             return
 
-        msg = (
+        msg = (  # type: ignore[unreachable]
             f"Unexpected state, result type={type(task_schedule.result)} should be an instance "
             f"of {TaskResultSuccess.__name__}, {TaskResultError.__name__} or {TaskResultCancelledError.__name__}"
         )

packages/service-library/src/servicelib/fastapi/docker_utils.py

@@ -104,7 +104,7 @@ async def retrieve_image_layer_information(
                 return TypeAdapter(DockerImageManifestsV2).validate_python(
                     json_response
                 )
-    return None
+    return None  # type: ignore[unreachable]
 
 
 async def pull_images(
@@ -130,7 +130,6 @@ async def pull_images(
         progress_unit="Byte",
         description=f"pulling {len(images)} images",
     ) as pbar:
-
         await asyncio.gather(
             *[
                 pull_image(

packages/service-library/src/servicelib/fastapi/lifespan_utils.py

@@ -37,7 +37,8 @@ def is_lifespan_called(state: State, lifespan_name: str) -> bool:
     # Valid signatures include: `()`, `(app)`, `(app, state)`, or even `(_, state)`.
     # It's easy to accidentally swap or misplace these arguments.
     assert not isinstance(  # nosec
-        state, FastAPI
+        state,  # type: ignore[unreachable]
+        FastAPI,
     ), "Did you swap arguments? `lifespan(app, state)` expects (app: FastAPI, state: State)"
 
     called_lifespans = state.get(_CALLED_LIFESPANS_KEY, set())

packages/service-library/src/servicelib/pools.py

@@ -1,11 +1,11 @@
+from collections.abc import Iterator
 from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
 from contextlib import contextmanager
-from typing import Iterator
 
 # only gets created on use and is guaranteed to be the s
 # ame for the entire lifetime of the application
-__shared_process_pool_executor = {}
-__shared_thread_pool_executor = {}
+__shared_process_pool_executor: dict[str, ProcessPoolExecutor] = {}
+__shared_thread_pool_executor: dict[str, ThreadPoolExecutor] = {}
 
 
 def _get_shared_process_pool_executor(**kwargs) -> ProcessPoolExecutor:

packages/service-library/src/servicelib/rabbitmq/_client_base.py

@@ -67,7 +67,7 @@ async def ping(self) -> bool:
             async with await aio_pika.connect(self.settings.dsn, timeout=1):
                 ...
             return True
-        return False
+        return False  # type: ignore[unreachable]
 
     @abstractmethod
     async def close(self) -> None: ...

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/resource_usage_tracker/pricing_plans.py

@@ -114,7 +114,7 @@ async def list_connected_services_to_pricing_plan_by_pricing_plan(
     product_name: ProductName,
     pricing_plan_id: PricingPlanId,
 ) -> list[PricingPlanToServiceGet]:
-    result: RutPricingPlanGet = await rabbitmq_rpc_client.request(
+    result = await rabbitmq_rpc_client.request(
         RESOURCE_USAGE_TRACKER_RPC_NAMESPACE,
         _RPC_METHOD_NAME_ADAPTER.validate_python(
             "list_connected_services_to_pricing_plan_by_pricing_plan"

packages/service-library/src/servicelib/redis/_client.py

@@ -99,7 +99,7 @@ async def ping(self) -> bool:
             # NOTE: retry_* input parameters from aioredis.from_url do not apply for the ping call
             await self._client.ping()
             return True
-        return False
+        return False  # type: ignore[unreachable]
 
     @property
     def is_healthy(self) -> bool:

packages/service-library/src/servicelib/rest_responses.py

@@ -25,7 +25,7 @@ def is_enveloped(payload: Mapping | str) -> bool:
         return is_enveloped_from_map(payload)
     if isinstance(payload, str):
         return is_enveloped_from_text(text=payload)
-    return False
+    return False  # type: ignore[unreachable]
 
 
 def unwrap_envelope(payload: Mapping[str, Any]) -> tuple: