80 lines
1.8 KiB
Python
80 lines
1.8 KiB
Python
"""`AbstractFactory` providers example."""
|
|
|
|
import abc
|
|
import dataclasses
|
|
import random
|
|
from typing import List
|
|
|
|
from dependency_injector import containers, providers
|
|
|
|
|
|
class AbstractCacheClient(metaclass=abc.ABCMeta):
|
|
...
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class RedisCacheClient(AbstractCacheClient):
|
|
host: str
|
|
port: int
|
|
db: int
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class MemcachedCacheClient(AbstractCacheClient):
|
|
hosts: List[str]
|
|
port: int
|
|
prefix: str
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class Service:
|
|
cache: AbstractCacheClient
|
|
|
|
|
|
class Container(containers.DeclarativeContainer):
|
|
|
|
cache_client_factory = providers.AbstractFactory(AbstractCacheClient)
|
|
|
|
service_factory = providers.Factory(
|
|
Service,
|
|
cache=cache_client_factory,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
container = Container()
|
|
|
|
cache_type = random.choice(["redis", "memcached"])
|
|
if cache_type == "redis":
|
|
container.cache_client_factory.override(
|
|
providers.Factory(
|
|
RedisCacheClient,
|
|
host="localhost",
|
|
port=6379,
|
|
db=0,
|
|
),
|
|
)
|
|
elif cache_type == "memcached":
|
|
container.cache_client_factory.override(
|
|
providers.Factory(
|
|
MemcachedCacheClient,
|
|
hosts=["10.0.1.1"],
|
|
port=11211,
|
|
prefix="my_app",
|
|
),
|
|
)
|
|
|
|
service = container.service_factory()
|
|
print(service.cache)
|
|
# The output depends on cache_type variable value.
|
|
#
|
|
# If the value is "redis":
|
|
# RedisCacheClient(host="localhost", port=6379, db=0)
|
|
#
|
|
# If the value is "memcached":
|
|
# MemcachedCacheClient(hosts=["10.0.1.1"], port=11211, prefix="my_app")
|
|
#
|
|
# If the value is None:
|
|
# Error: AbstractFactory(<class "__main__.AbstractCacheClient">) must be
|
|
# overridden before calling
|