update container
This commit is contained in:
1 parent
675b9c8b7e
commit
e73c849a7e
5 files changed
+150
No files matched your search
@@ -0,0 +1,14 @@
|
|||||||
|
[build-system]
|
||||||
|
requires = ["setuptools>=61"]
|
||||||
|
build-backend = "setuptools.build_meta"
|
||||||
|
|
||||||
|
[project]
|
||||||
|
name = "container"
|
||||||
|
version = "0.1.0"
|
||||||
|
description = "Dependency injection container for the BookSync applications."
|
||||||
|
requires-python = ">=3.12"
|
||||||
|
|
||||||
|
[tool.setuptools.packages.find]
|
||||||
|
where = ["src"]
|
||||||
|
include = ["container*"]
|
||||||
|
namespaces = true
|
||||||
@@ -0,0 +1,6 @@
|
|||||||
|
"""Public API for the dependency injection container."""
|
||||||
|
|
||||||
|
from .container import Container
|
||||||
|
from .container_interface import IContainer
|
||||||
|
|
||||||
|
__all__ = ["Container", "IContainer"]
|
||||||
@@ -0,0 +1,95 @@
|
|||||||
|
# __ _____ _ __ ____________________ __ _____________________ _____ _______________
|
||||||
|
# / ___/ / /_ ____ ____ / /_ ___ ____ / __ \ ___ _ __/
|
||||||
|
# \__ \ / __ \ / __ \ / __ \ / __/ / _ \ / __/ ____ / / / / / _ \ | | / /
|
||||||
|
# ___/ / / / / / / /_/ / / /_/ / / /_ / __/ / / /___/ / /_/ / / __/ | |/ /
|
||||||
|
# /____/ /_/ /_/ \____/ \____/ \__/ \___/ /_/ _______ /______/ \___/ |___/
|
||||||
|
import importlib
|
||||||
|
import inspect
|
||||||
|
from typing import Any, TypeVar, cast
|
||||||
|
|
||||||
|
from .container_interface import IContainer
|
||||||
|
|
||||||
|
T = TypeVar("T")
|
||||||
|
|
||||||
|
|
||||||
|
class Container(IContainer):
|
||||||
|
"""Conteneur léger de services avec résolution par type."""
|
||||||
|
|
||||||
|
def __init__(self) -> None:
|
||||||
|
"""Initialise des registres indépendants pour ce conteneur."""
|
||||||
|
self.__instances: dict[str, Any] = {
|
||||||
|
"container": self,
|
||||||
|
self.get_type(Container): self,
|
||||||
|
}
|
||||||
|
self.__alias: dict[str, Any] = {}
|
||||||
|
self.__parameters: dict[str, Any] = {}
|
||||||
|
self.alias(IContainer, Container)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def get_type(name: type[Any] | str) -> str:
|
||||||
|
"""Retourner l'identifiant stable d'un type ou d'une chaîne."""
|
||||||
|
if isinstance(name, str):
|
||||||
|
return name
|
||||||
|
module = inspect.getmodule(name)
|
||||||
|
if module is None:
|
||||||
|
raise TypeError(f"Impossible de déterminer le module de {name!r}.")
|
||||||
|
return f"{module.__name__}.{name.__name__}"
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def get_instance(type_: str, args: list[Any]) -> object:
|
||||||
|
"""Construire dynamiquement une instance à partir de son identifiant."""
|
||||||
|
module = importlib.import_module(type_[0 : type_.rfind(".")])
|
||||||
|
return getattr(module, type_[type_.rfind(".") + 1 :])(*args)
|
||||||
|
|
||||||
|
def set(self, name: type[Any], obj: object) -> None:
|
||||||
|
"""Enregistrer une instance pour un type."""
|
||||||
|
self.__instances[Container.get_type(name)] = obj
|
||||||
|
|
||||||
|
def __getitem__(self, name: type[T]) -> T:
|
||||||
|
"""Résoudre un service avec la syntaxe d'indexation historique du sandbox."""
|
||||||
|
return cast("T", self.get(name))
|
||||||
|
|
||||||
|
def __setitem__(self, name: type[T], obj: T) -> None:
|
||||||
|
"""Enregistrer un service avec la syntaxe d'indexation historique du sandbox."""
|
||||||
|
self.set(name, obj)
|
||||||
|
|
||||||
|
def __contains__(self, name: type[Any] | str) -> bool:
|
||||||
|
"""Indiquer si un service a déjà été enregistré dans le conteneur."""
|
||||||
|
return self.get_type(name) in self.__instances
|
||||||
|
|
||||||
|
def get(self, name: type[Any] | str) -> Any:
|
||||||
|
"""Résoudre ou construire un service par son type ou son identifiant."""
|
||||||
|
if name in self.__parameters:
|
||||||
|
return self.__parameters[name]
|
||||||
|
|
||||||
|
service_id: str = self.get_type(name)
|
||||||
|
|
||||||
|
if service_id in self.__alias:
|
||||||
|
return self.get(self.__alias[Container.get_type(name)])
|
||||||
|
|
||||||
|
if service_id not in self.__instances:
|
||||||
|
if isinstance(name, str):
|
||||||
|
raise KeyError(f"Service introuvable : {name}.")
|
||||||
|
args: list[Any] = []
|
||||||
|
if len(inspect.signature(name.__init__).parameters) > 1:
|
||||||
|
for arg, parameter in inspect.signature(name.__init__).parameters.items():
|
||||||
|
if arg in ["self", "args", "kwargs"]:
|
||||||
|
continue
|
||||||
|
if parameter.annotation != inspect.Parameter.empty:
|
||||||
|
try:
|
||||||
|
args.append(self.get(parameter.annotation))
|
||||||
|
except (AttributeError, KeyError, TypeError):
|
||||||
|
args.append(self.get(arg))
|
||||||
|
else:
|
||||||
|
args.append(self.get(arg))
|
||||||
|
self.__instances[service_id] = Container.get_instance(service_id, args)
|
||||||
|
|
||||||
|
return self.__instances[service_id]
|
||||||
|
|
||||||
|
def alias(self, alias: type[Any] | str, name: type[Any] | str) -> None:
|
||||||
|
"""Associer un identifiant à un autre service."""
|
||||||
|
self.__alias[Container.get_type(alias)] = name
|
||||||
|
|
||||||
|
def set_parameter(self, name: str, value: Any) -> None:
|
||||||
|
"""Enregistrer un paramètre nommé."""
|
||||||
|
self.__parameters[name] = value
|
||||||
@@ -0,0 +1,26 @@
|
|||||||
|
# __ _____ _ __ ____________________ __ _____________________ _____ _______________
|
||||||
|
# / ___/ / /_ ____ ____ / /_ ___ ____ / __ \ ___ _ __/
|
||||||
|
# \__ \ / __ \ / __ \ / __ \ / __/ / _ \ / __/ ____ / / / / / _ \ | | / /
|
||||||
|
# ___/ / / / / / / /_/ / / /_/ / / /_ / __/ / / /___/ / /_/ / / __/ | |/ /
|
||||||
|
# /____/ /_/ /_/ \____/ \____/ \__/ \___/ /_/ _______ /______/ \___/ |___/
|
||||||
|
from typing import Any, Protocol
|
||||||
|
|
||||||
|
|
||||||
|
class IContainer(Protocol):
|
||||||
|
"""Définir le port partagé de résolution des dépendances."""
|
||||||
|
|
||||||
|
def get(self, name: type[Any] | str) -> Any:
|
||||||
|
"""Résoudre un service par son type ou son identifiant."""
|
||||||
|
...
|
||||||
|
|
||||||
|
def set(self, name: type[Any], obj: object) -> None:
|
||||||
|
"""Enregistrer une instance pour un type."""
|
||||||
|
...
|
||||||
|
|
||||||
|
def alias(self, alias: type[Any] | str, name: type[Any] | str) -> None:
|
||||||
|
"""Associer un identifiant à un autre service."""
|
||||||
|
...
|
||||||
|
|
||||||
|
def set_parameter(self, name: str, value: Any) -> None:
|
||||||
|
"""Enregistrer un paramètre nommé."""
|
||||||
|
...
|
||||||
@@ -0,0 +1,9 @@
|
|||||||
|
import unittest
|
||||||
|
|
||||||
|
|
||||||
|
class ContainerPublicApiTests(unittest.TestCase):
|
||||||
|
def test_public_types_are_importable(self) -> None:
|
||||||
|
from container import Container, IContainer
|
||||||
|
|
||||||
|
self.assertEqual(Container.__name__, "Container")
|
||||||
|
self.assertEqual(IContainer.__name__, "IContainer")
|
||||||
Reference in new issue
Block a user