Recipes
This is a collection of how to use certain features of Antidote or simply examples of what can be done.
Interface & Implementations
An interface defines a contract which should be respected by the implementation. It can be declared
with interface()
and implementations with implements
:
from antidote import interface, implements
@interface
class Command:
def run(self) -> int:
...
@implements(Command)
class CommandImpl(Command):
def run(self) -> int:
return 0
Command
can also be a Protocol
. If it’s runtime_checkable()
,
it’ll be enforced at runtime. The implementation can then be retrieved as if Command
was a
regular service:
>>> from antidote import inject, world
>>> @inject
... def cmd(command: Command = inject.me()) -> Command:
... return command
>>> cmd()
<CommandImpl object at ...>
>>> world.get(Command)
<CommandImpl object at ...>
Qualifiers
When working with multiple implementations for an interface qualifiers offer an easy way to distinguish them:
from enum import auto, Enum
from typing import Protocol
from antidote import implements, interface
class Event(Enum):
START = auto()
INITIALIZATION = auto()
RELOAD = auto()
SHUTDOWN = auto()
@interface
class Hook(Protocol):
def run(self, event: Event) -> None:
...
@implements(Hook).when(qualified_by=Event.START)
class StartUpHook:
def run(self, event: Event) -> None:
pass
@implements(Hook).when(qualified_by=[Event.INITIALIZATION,
Event.RELOAD])
class OnAnyUpdateHook:
def run(self, event: Event) -> None:
pass
@implements(Hook).when(qualified_by=list(Event))
class LogAnyEventHook:
def run(self, event: Event) -> None:
pass
Note
For Python <3.9 you can use the following trick or create your own implements_when()
wrapper.
from typing import TypeVar
T = TypeVar('T')
def _(x: T) -> T:
return x
@_(implements(Hook).when(qualified_by=Event.START))
class StartUpHook:
def run(self, event: Event) -> None:
pass
Now Antidote will raise an error if one tries to use LifeCycleHook
like a service:
>>> from antidote import world
>>> world.get(Hook)
Traceback (most recent call last):
File "<stdin>", line 1, in ?
DependencyInstantiationError: ...
To retrieve a single implementation you can use:
>>> from antidote import inject
>>> world.get[Hook].single(qualified_by=Event.SHUTDOWN)
<LogAnyEventHook object at ...>
>>> @inject
... def single_hook(hook: Hook = inject.me(qualified_by=Event.SHUTDOWN)
... ) -> Hook:
... return hook
>>> single_hook()
<LogAnyEventHook object at ...>
>>> @inject
... def single_hook2(hook = inject.get[Hook].single(qualified_by=Event.SHUTDOWN)
... ) -> Hook:
... return hook
>>> single_hook2()
<LogAnyEventHook object at ...>
And to retrieve multiple of them:
>>> world.get[Hook].all(qualified_by=Event.START)
[<LogAnyEventHook object at ...>, <StartUpHook object at ...>]
>>> @inject
... def all_hooks(hook: list[Hook] = inject.me(qualified_by=Event.START)
... ) -> list[Hook]:
... return hook
>>> all_hooks()
[<LogAnyEventHook object at ...>, <StartUpHook object at ...>]
>>> @inject
... def all_hooks2(hook = inject.get[Hook].all(qualified_by=Event.START)
... ) -> list[Hook]:
... return hook
>>> all_hooks2()
[<LogAnyEventHook object at ...>, <StartUpHook object at ...>]
It’s also possible to define more complex constraints, see single()
for example
and QualifiedBy
.
Lazily call a function
Calling lazily a function can be done with lazy
or
LazyMethodCall
for methods. Both will pass any arguments passed on
and can either be singletons or not.
Function
import requests
from antidote import LazyCall, inject
def fetch_remote_conf(name):
return requests.get(f"https://example.com/conf/{name}")
CONF_A = LazyCall(fetch_remote_conf)("conf_a")
@inject(dependencies=(CONF_A,))
def f(conf):
return conf
Using CONF_A
as a representation of the result allows one to easily identify
where this dependency is needed. Moreover neither f
nor its caller needs to
be aware on how to call fetch_remote_conf
.
Method
Lazily calling a method requires the class to be Service
.
from antidote import LazyMethodCall, injectable
@injectable
class ExampleCom:
def get(url):
return requests.get(f"https://example.com{url}")
STATUS = LazyMethodCall(get, singleton=False)("/status")
Note
If you intend to define lazy constants, consider using
Constants
instead.
Create a stateful factory
Antidote supports stateful factories simply by using defining a class as a factory:
from antidote import factory
class ID:
def __init__(self, id: str):
self.id = id
def __repr__(self):
return "ID(id='{}')".format(self.id)
@factory(singleton=False)
class IDFactory:
def __init__(self, id_prefix: str = "example"):
self._prefix = id_prefix
self._next = 1
def __call__(self) -> ID:
id = ID("{}_{}".format(self._prefix, self._next))
self._next += 1
return id
>>> from antidote import world
>>> world.get(ID, source=IDFactory)
ID(id='example_1')
>>> world.get(ID, source=IDFactory)
ID(id='example_2')
In this example we choose to inject id_prefix
in the __init__()
, but we
also could have done it in the __call__()
. Both are injected by default, but they
have different use cases. The factory itself is always a singleton, so static dependencies
should be injected through __init__()
. If you need dependencies that changes, get
them through __call__()
. Obviously you can change that behavior through the
Factory.Conf
: defined in __antidote__
.
Note
Stateful factories can also be used to provide dependencies that have a more complex
scope than Antidote provides (singleton or not). Although, if you need to handle some
scope for multiples dependencies it might be worth just extending Antidote through a
Provider
.
Configuration
Here are some examples on how to use Constants
to handle configuration coming
from different sources.
From the environment
import os
from typing import Optional
from antidote import Constants, const
class Env(Constants):
SECRET = const[str]()
def provide_const(self, name: str, arg: Optional[object]):
return os.environ[name]
>>> from antidote import world, inject
>>> os.environ['SECRET'] = 'my_secret'
>>> world.get[str](Env.SECRET)
'my_secret'
>>> @inject
... def f(secret: str = Env.SECRET) -> str:
... return secret
>>> f()
'my_secret'
From a dictionary
Configuration can be stored in a lot of different formats, or even be retrieved on a remote endpoint at start-up. Most of the time you would be able to easily convert it to a dictionary and use the following:
import os
from typing import Optional
from antidote import Constants, const
class Conf(Constants):
HOST = const[str]('host')
AWS_API_KEY = const[str]('aws.api_key')
def __init__(self):
# Load your configuration into a dictionary
self._raw_conf = {
"host": "localhost",
"aws": {
"api_key": "my key"
}
}
def provide_const(self, name: str, arg: Optional[str]):
from functools import reduce
assert arg is not None and isinstance(arg, str) # sanity check
return reduce(dict.get, arg.split('.'), self._raw_conf) # type: ignore
>>> from antidote import world, inject
>>> world.get[str](Conf.HOST)
'localhost'
>>> world.get(Conf.AWS_API_KEY)
'my key'
>>> @inject
... def f(key: str = Conf.AWS_API_KEY) -> str:
... return key
>>> f()
'my key'
Specifying a type / Using Enums
You can specify a type when using const()
. It’s main purpose is to provide
a type for Mypy when the constants are directly accessed from an instance. However
Constants
will also automatically force the cast if the type is one
of str
, float
or int
. You can control this behavior with
the auto_cast
argument of Conf
. A typical use case
would be to support enums as presented here:
from enum import Enum
from typing import Optional
from antidote import Constants, const
class Env(Enum):
PROD = 'prod'
PREPRDO = 'preprod'
class Conf(Constants):
__antidote__ = Constants.Conf(auto_cast=[int, Env])
DB_PORT = const[int]()
ENV = const[Env]()
def provide_const(self, name: str, arg: Optional[object]):
return {'db_port': '5432', 'env': 'prod'}[name.lower()]
>>> from antidote import world, inject
>>> Conf().DB_PORT
5432
>>> Conf().ENV
<Env.PROD: 'prod'>
>>> world.get[int](Conf.DB_PORT)
5432
>>> world.get[Env](Conf.ENV)
<Env.PROD: 'prod'>
>>> @inject
... def f(env: Env = Conf.ENV) -> Env:
... return env
>>> f()
<Env.PROD: 'prod'>
The goal of this is to simplify common operations when manipulating the environment
or configuration files. If you need complex behavior, consider using a service for this
or define your Configuration class as public=True
in Conf
and use it as a one.
Default values
Default values can be specified in const()
:
import os
from antidote import Constants, const
class Env(Constants):
HOST = const[str]('HOST', default='localhost')
def get(self, value):
return os.environ[value]
It will be use if get
raises a py:exec:KeyError. For more complex behavior,
using a collections.ChainMap
which loads your defaults and the user is a good
alternative:
from collections import ChainMap
from antidote import Constants, const
class Configuration(Constants):
def __init__(self):
user_conf = dict() # load conf from a file, etc..
default_conf = dict()
# User conf will override default_conf
self._raw_conf = ChainMap(user_conf, default_conf)
An alternative to this would be using a configuration format that supports overrides, such as HOCON.
Scopes
A dependency may be associated with a scope. If so it’ll cached for as along as the scope is
valid. The most common scope being the singleton scope where dependencies are cached forever.
When the scope is set to None
, the dependency value will be retrieved each time.
Scopes can be create through world.scopes.new()
. The name is only used to
have a friendly identifier when debugging.
>>> from antidote import world
>>> REQUEST_SCOPE = world.scopes.new(name='request')
To use the newly created scope, use scope
parameters:
>>> from antidote import injectable
>>> @injectable(scope=REQUEST_SCOPE)
... class Dummy:
... pass
As Dummy
has been defined with a custom scope, the dependency value will
be kep as long as REQUEST_SCOPE
stays valid. That is to say, until you reset
it with world.scopes.reset()
:
>>> current = world.get(Dummy)
>>> current is world.get(Dummy)
True
>>> world.scopes.reset(REQUEST_SCOPE)
>>> current is world.get(Dummy)
False
In a Flask app for example you would then just reset the scope after each request:
from flask import Flask, Request
from antidote import factory
app = Flask(__name__)
@app.after_request
def reset_request_scope():
world.scopes.reset(REQUEST_SCOPE)
@factory(scope=REQUEST_SCOPE)
def current_request() -> Request:
from flask import request
return request