ets-labs/python-dependency-injector
1
1
Fork 0
mirror of https://github.com/ets-labs/python-dependency-injector.git synced 2025-07-01 10:53:14 +03:00
Code Issues Packages Projects Releases Wiki Activity

feat: Add resource_type argument on init_resources and shutdown_resources methods to provide a more granular way to initialize the desired resources and add the possibility to scope that ones.

Browse Source
This commit is contained in:
Aran Moncusi Ramirez 2025-02-21 20:51:21 +01:00 committed by ZipFile
parent b411807572
commit 23fbb71fde
2 changed files with 7 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/dependency_injector/containers.pyi
View File

@ -22,7 +22,7 @@ try:
except ImportError:
from typing_extensions import Self as _Self
from .providers import Provider, ProviderParent, Self
from .providers import Provider, Resource, Self, ProviderParent
C_Base = TypeVar("C_Base", bound="Container")
C = TypeVar("C", bound="DeclarativeContainer")
@ -74,8 +74,8 @@ class Container:
from_package: Optional[str] = None,
) -> None: ...
def unwire(self) -> None: ...
def init_resources(self) -> Optional[Awaitable[None]]: ...
def shutdown_resources(self) -> Optional[Awaitable[None]]: ...
def init_resources(self, resource_type: Type[Resource]=None) -> Optional[Awaitable]: ...
def shutdown_resources(self, resource_type: Type[Resource]=None) -> Optional[Awaitable]: ...
def load_config(self) -> None: ...
def apply_container_providers_overridings(self) -> None: ...
def reset_singletons(self) -> SingletonResetContext[C_Base]: ...

8
src/dependency_injector/containers.pyx
View File

@ -315,11 +315,11 @@ class DynamicContainer(Container):
self.wired_to_modules.clear()
self.wired_to_packages.clear()
def init_resources(self):
def init_resources(self, resource_type=providers.Resource):
"""Initialize all container resources."""
futures = []
for provider in self.traverse(types=[providers.Resource]):
for provider in self.traverse(types=[resource_type]):
resource = provider.init()
if __is_future_or_coroutine(resource):
@ -328,7 +328,7 @@ class DynamicContainer(Container):
if futures:
return asyncio.gather(*futures)
def shutdown_resources(self):
def shutdown_resources(self, resource_type=providers.Resource):
"""Shutdown all container resources."""
def _independent_resources(resources):
for resource in resources:
@ -360,7 +360,7 @@ class DynamicContainer(Container):
for resource in resources_to_shutdown:
resource.shutdown()
resources = list(self.traverse(types=[providers.Resource]))
resources = list(self.traverse(types=[resource_type]))
if any(resource.is_async_mode_enabled() for resource in resources):
return _async_ordered_shutdown(resources)
else: