-
Notifications
You must be signed in to change notification settings - Fork 5
Expand file tree
/
Copy pathdi_container.py
More file actions
68 lines (55 loc) · 2.01 KB
/
di_container.py
File metadata and controls
68 lines (55 loc) · 2.01 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
from dependency_injector.containers import DeclarativeContainer, WiringConfiguration
from dependency_injector.providers import Dependency, Factory, Singleton
from sqlalchemy_bind_manager import SQLAlchemyBindManager
from sqlalchemy_bind_manager.repository import SQLAlchemyAsyncRepository
from common.config import AppConfig
from domains.books._gateway_interfaces import (
BookEventGatewayInterface,
BookRepositoryInterface,
)
from domains.books._models import BookModel
from gateways.event import NullEventGateway
class Container(DeclarativeContainer):
"""
Dependency injection container.
Docs: https://python-dependency-injector.ets-labs.org/
"""
wiring_config = WiringConfiguration(
packages=[
"common",
"domains",
]
)
"""
We could use the config provider but it would transform our nice typed
configuration in a dictionary, therefore we return it as a raw object.
"""
config = Dependency(instance_of=AppConfig)
"""
Class mappings
These are classes we want the container to manage the life cycle for
(e.g. Singletons), we map them using their class name directly.
"""
SQLAlchemyBindManager = Singleton(
SQLAlchemyBindManager,
config=config.provided.SQLALCHEMY_CONFIG,
)
"""
Interface => Class mappings
We use the interface class name as key so that we can trigger the injection
using `class.__name__` and avoid using any hardcoded string or constant.
e.g.
Mapping
MyInterface = providers.Factory("http_app.storage.repositories.ConcreteClass")
Usage
@inject
def function(
service: MyInterface = Provide[MyInterface.__name__],
)
"""
BookRepositoryInterface: Factory[BookRepositoryInterface] = Factory(
SQLAlchemyAsyncRepository,
bind=SQLAlchemyBindManager.provided.get_bind.call(),
model_class=BookModel,
)
BookEventGatewayInterface: Factory[BookEventGatewayInterface] = Factory(NullEventGateway)