Common: Mitigate mypy errors from DIContainer
This commit is contained in:
parent
9a880123da
commit
783cc06c0d
|
@ -1,6 +1,6 @@
|
||||||
import inspect
|
import inspect
|
||||||
from contextlib import suppress
|
from contextlib import suppress
|
||||||
from typing import Any, Sequence, Type, TypeVar
|
from typing import Any, Sequence, Type, TypeVar, no_type_check
|
||||||
|
|
||||||
from common.utils.code_utils import del_key
|
from common.utils.code_utils import del_key
|
||||||
|
|
||||||
|
@ -15,6 +15,9 @@ class UnregisteredConventionError(ValueError):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
# Mypy doesn't handle cases where abstract class is passed as Type[...]
|
||||||
|
# https://github.com/python/mypy/issues/4717
|
||||||
|
# We are using typing.no_type_check to mitigate these errors
|
||||||
class DIContainer:
|
class DIContainer:
|
||||||
"""
|
"""
|
||||||
A dependency injection (DI) container that uses type annotations to resolve and inject
|
A dependency injection (DI) container that uses type annotations to resolve and inject
|
||||||
|
@ -26,6 +29,7 @@ class DIContainer:
|
||||||
self._instance_registry = {}
|
self._instance_registry = {}
|
||||||
self._convention_registry = {}
|
self._convention_registry = {}
|
||||||
|
|
||||||
|
@no_type_check
|
||||||
def register(self, interface: Type[T], concrete_type: Type[T]):
|
def register(self, interface: Type[T], concrete_type: Type[T]):
|
||||||
"""
|
"""
|
||||||
Register a concrete `type` that satisfies a given interface.
|
Register a concrete `type` that satisfies a given interface.
|
||||||
|
@ -55,6 +59,7 @@ class DIContainer:
|
||||||
self._type_registry[interface] = concrete_type
|
self._type_registry[interface] = concrete_type
|
||||||
del_key(self._instance_registry, interface)
|
del_key(self._instance_registry, interface)
|
||||||
|
|
||||||
|
@no_type_check
|
||||||
def register_instance(self, interface: Type[T], instance: T):
|
def register_instance(self, interface: Type[T], instance: T):
|
||||||
"""
|
"""
|
||||||
Register a concrete instance that satisfies a given interface.
|
Register a concrete instance that satisfies a given interface.
|
||||||
|
@ -73,6 +78,7 @@ class DIContainer:
|
||||||
self._instance_registry[interface] = instance
|
self._instance_registry[interface] = instance
|
||||||
del_key(self._type_registry, interface)
|
del_key(self._type_registry, interface)
|
||||||
|
|
||||||
|
@no_type_check
|
||||||
def register_convention(self, type_: Type[T], name: str, instance: T):
|
def register_convention(self, type_: Type[T], name: str, instance: T):
|
||||||
"""
|
"""
|
||||||
Register an instance as a convention
|
Register an instance as a convention
|
||||||
|
@ -101,6 +107,7 @@ class DIContainer:
|
||||||
"""
|
"""
|
||||||
self._convention_registry[(type_, name)] = instance
|
self._convention_registry[(type_, name)] = instance
|
||||||
|
|
||||||
|
@no_type_check
|
||||||
def resolve(self, type_: Type[T]) -> T:
|
def resolve(self, type_: Type[T]) -> T:
|
||||||
"""
|
"""
|
||||||
Resolves all dependencies and returns a new instance of `type_` using constructor dependency
|
Resolves all dependencies and returns a new instance of `type_` using constructor dependency
|
||||||
|
|
|
@ -33,12 +33,10 @@ def _subscribe_and_store_to_event_repository(container: DIContainer):
|
||||||
|
|
||||||
|
|
||||||
def _subscribe_ping_scan_event(container: DIContainer):
|
def _subscribe_ping_scan_event(container: DIContainer):
|
||||||
# Mypy don't handle cases where abstract class is passed as Type[...]
|
agent_event_queue = container.resolve(IAgentEventQueue)
|
||||||
# https://github.com/python/mypy/issues/4717
|
agent_repository = container.resolve(IAgentRepository)
|
||||||
agent_event_queue = container.resolve(IAgentEventQueue) # type: ignore
|
machine_repository = container.resolve(IMachineRepository)
|
||||||
agent_repository = container.resolve(IAgentRepository) # type: ignore
|
node_repository = container.resolve(INodeRepository)
|
||||||
machine_repository = container.resolve(IMachineRepository) # type: ignore
|
|
||||||
node_repository = container.resolve(INodeRepository) # type: ignore
|
|
||||||
|
|
||||||
handler = handle_ping_scan_event(agent_repository, machine_repository, node_repository)
|
handler = handle_ping_scan_event(agent_repository, machine_repository, node_repository)
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue