# Clean IoC
A simple dependency injection library for python that requires nothing of your application code (except that you use typing).
## Basic Registering and resolving
There are 4 basic modes of registering a new set of classes
### Implementation
```python
class UserRepository(abc.ABC):
@abc.abstractmethod
def add(self, user):
pass
class InMemoryUserRepository(UserRepository):
def __init__(self):
self.users = []
def add(self, user):
# This is obviously terrible, but it's for demo purposes
self.users.append(user)
class SqlAlchemyUserRepository(UserRepository):
def __init__(self):
# Do some db stuff here
pass
def add(self, user):
# Do some db stuff here
pass
container = Container()
container.register(UserRepository, InMemoryUserRepository)
repository = container.resolve(UserRepository) # This will return an InMemoryUserRepository
```
### Concrete Class
```python
class ClientDependency:
def get_int(self):
return 10
class Client:
def __init__(self, dep: ClientDependency):
self.dep = dep
def get_number(self):
return self.dep.get_int()
container = Container()
container.register(ClientDependency)
container.register(Client)
client = container.resolve(Client)
client.get_number() # returns 10
```
### Factory
```python
class ClientDependency:
def get_int(self):
return 10
class Client:
def __init__(self, dep: ClientDependency):
self.dep = dep
def get_number(self):
return self.dep.get_int()
def client_factory(dep: ClientDependency):
return Client(dep=dep)
container = Container()
container.register(ClientDependency)
container.register(Client, factory=client_factory)
client = container.resolve(Client)
client.get_number() # returns 10
```
### Instance
```python
class ClientDependency:
def __init__(self, num):
self.num = num
def get_int(self):
return self.num
class Client:
def __init__(self, dep: ClientDependency):
self.dep = dep
def get_number(self):
return self.dep.get_int()
client_dependency = ClientDependency(num=10)
container = Container()
container.register(ClientDependency, instance=client_dependency)
container.register(Client)
client = container.resolve(Client)
client.get_number() # returns 10
```
## List resolving
If you have multiple dependencues you can simply define a dependency as a list[T] and you can return all of the instances.
```python
class ClientDependency:
def __init__(self, numbers: list[int]):
self.numbers = numbers
def get_numbers(self):
return self.numbers
class Client:
def __init__(self, dep: ClientDependency):
self.dep = dep
def get_numbers(self):
return self.dep.get_numbers()
container = Container()
container.register(ClientDependency)
container.register(Client)
container.register(int, instance=1)
container.register(int, instance=2)
container.register(int, instance=3)
client = container.resolve(Client)
client.get_numbers() # returns [3, 2, 1]
```
## Decorators
Follows a object orientated decoration pattern, rather than a decoration annotation.
The main reason for this was to allow decotation of registered instances
```python
class Client:
def __init__(self, number: int):
self.number = number
def get_number(self):
return self.number
class DoubleClientDecorator(Client):
def __init__(self, client: Client):
self.client = client
def get_number(self):
return self.client.get_number() * 2
container = Container()
container.register(Client)
container.register_decorator(Client, DoubleClientDecorator)
container.register(int, instance=10)
client = container.resolve(Client)
client.get_number() # returns 20
```
Decorators are resolved in order of when first registered. So the first registered decorator is the highest in the class tree
```python
class Concrete:
pass
class DecoratorOne(Concrete):
def __init__(self, child: Concrete):
self.child = child
class DecoratorTwo(Concrete):
def __init__(self, child: Concrete):
self.child = child
container = Container()
container.register(Concrete)
container.register_decorator(Concrete, DecoratorOne)
container.register_decorator(Concrete, DecoratorTwo)
root = container.resolve(Concrete)
type(root) # returns DecoratorOne
type(root.child) # returns DecoratorTwo
type(root.child.child) # returns Concrete
```
## Subclasses registration
This feature allows registration of all subclasses of a giveb type
```python
class Client(abc.ABC):
@abc.abstractmethod
def get_number(self):
pass
class TenClient(Client):
def get_number(self):
return 10
class TwentyClient(Client):
def get_number(self):
return 20
container = Container()
container.register_subclasses(Client)
ten_client = container.resolve(TenClient)
ten_client.get_number() # returns 10
twenty_client = container.resolve(TwentyClient)
twenty_client.get_number() # returns 20
# Resolve all subsclasses of Client
client = container.resolve(list[Client]) ## [TwentyClient(), TenClient()]
```
## Lifespans
Lifespans configure how long and resolved object says alive for
There are 4 lifespan types
### transient
Always create a new instance
```python
container.register(Client, lifespan=Lifespan.transient)
```
### once_per_graph (Default behaviour)
Only create one instance throughout the resolve call
```python
container.register(Client, lifespan=Lifespan.once_per_graph)
```
### scoped
Only create a new instance through the lifetime a [scope](#scopes). When not in a scope the behaviour is the same as **once_per_graph**.
```python
container.register(Client, lifespan=Lifespan.scoped)
```
### singleton
Only one instance of the object is created throughout the lifespan of the container
```python
container.register(Client, lifespan=Lifespan.singleton)
```
*Note:*
When registering an instance, then the behaviour is always singleton
```python
container.register(int, instance=10)
```
## Open Generics
Registers all generic subclasses of the service type and allows you to resolve with the generic alias
```python
T = TypeVar("T")
class HelloCommand:
pass
class GoodbyeCommand:
pass
class CommandHandler(Generic[T]):
def handle(self, command: T):
pass
class HelloCommandHandler(CommandHandler[HelloCommand]):
def handle(self, command: HelloCommand):
print('HELLO')
class GoodbyeCommandHandler(CommandHandler[GoodbyeCommand]):
def handle(self, command: GoodbyeCommand):
print('GOODBYE')
container = Container()
container.register_open_generic(CommandHandler)
h1 = container.resolve(CommandHandler[HelloCommand])
h2 = container.resolve(CommandHandler[GoodbyeCommand])
h1.handle(HelloCommand()) # prints 'HELLO'
h2.handle(GoodbyeCommand()) # prints 'GOODBYE'
```
## Open Generic Decorators
Allows you to add decorators to your open generic registrations
```python
T = TypeVar("T")
class HelloCommand:
pass
class GoodbyeCommand:
pass
class CommandHandler(Generic[T]):
def handle(self, command: T):
pass
class HelloCommandHandler(CommandHandler[HelloCommand]):
def handle(self, command: HelloCommand):
print('HELLO')
class GoodbyeCommandHandler(CommandHandler[GoodbyeCommand]):
def handle(self, command: GoodbyeCommand):
print('GOODBYE')
class AVeryBigCommandHandlerDecorator(Generic[T]):
def __init__(self, handler: CommandHandler[T]):
self.handler = handler
def handle(self, command: T):
print('A VERY BIG')
self.handler.handle(command=command)
container = Container()
container.register_open_generic(CommandHandler)
container.register_open_generic_decorator(CommandHandler, AVeryBigCommandHandlerDecorator)
h1 = container.resolve(CommandHandler[HelloCommand])
h2 = container.resolve(CommandHandler[GoodbyeCommand])
h1.handle(HelloCommand()) # prints 'A VERY BIG\nHELLO'
h2.handle(GoodbyeCommand()) # prints 'A VERY BIG\nGOODBYE'
```
## Scopes
Scopes are a machanism where you guarantee that dependency can be temporarily a singleton within the scope. You can also register dependencies that that are only available withon the scope.
Some good use cases for scope lifetimes are:
- http request in a web server
- message/event if working on a message based system
For instance you could keep an single database connection open for the entire lifetime of the http request
```python
class DbConnection:
def run_sql(self, statement):
# Done some sql Stuff
pass
container.register(DbConnection, lifespan=Lifespan.scoped)
with container.get_scope() as scope:
db_conn = scope.resolve(DbConnection)
db_conn.run_sql("UPDATE table SET column = 1")
```
Scopes can also be use with asyncio
```python
class AsyncDbConnection:
async def run_sql(self, statement):
# Done some sql Stuff
pass
container.register(AsyncDbConnection, lifespan=Lifespan.scoped)
async with container.get_scope() as scope:
db_conn = scope.resolve(AsyncDbConnection)
await db_conn.run_sql("UPDATE table SET column = 1")
```
### Scoped Teardowns
When you are finished with some dependenies within a scope you might want to perform some teardown action before you exit the scope. For example if we want to close our db connection.
```python
class AsyncDbConnection:
async def run_sql(self, statement):
# Done some sql Stuff
pass
async def close(self):
# Close the connection
pass
async def close_connection(conn: AsyncDbConnection):
await conn.close()
container.register(DbConnection, lifespan=Lifespan.scoped, scoped_teardown=close_connection)
async with container.get_scope() as scope:
db_conn = scope.resolve(AsyncDbConnection)
await db_conn.run_sql("UPDATE table SET column = 1")
# close connection is run when we exit the scope
```
***Note***: *When using the scope as an async context manager you need both sync and async teardowns are run, when a scope is used as a normal sync context manager async teardowns are ignored*
## Named registrations & Registration filters
By default the last unnamed registration is what the container will return when resolve is called as below.
```python
container = Container()
container.register(int, instance=1)
container.register(int, instance=2)
container.register(int, instance=3)
number = container.resolve(int) # returns 3
```
To be more selective of what we return we can add a name to the registration and apply a registration filter when we resolve.
A registration filter is simply function that receives a **Registration** and returns a **bool**
For example if we wanted to get the int named **"One"** we do the following
```python
container = Container()
container.register(int, instance=1, name="One")
container.register(int, instance=2, name="Two")
container.register(int, instance=3, name="Three")
number = container.resolve(int, filter=lambda r: r.name == "One") # returns 1
```
Clean IOC comes with a set of in built registration filters that can be found [here](./clean_ioc/registration_filters.py)
We can get the desired behaviour as above
```python
from clean_ioc.registration_filters import with_name
container = Container()
container.register(int, instance=1, name="One")
container.register(int, instance=2, name="Two")
container.register(int, instance=3, name="Three")
number = container.resolve(int, filter=with_name("One")) # returns 1
```
## Dependency Settings
Dependency settings are defined at registration and allow you to define the selection or setting dependencies
```python
class Client:
def __init__(self, number=10):
self.number = number
def get_number(self):
return self.number
container = Container()
container.register(int, instance=1, name="One")
container.register(int, instance=2)
container.register(
Client,
name="SetsValue",
dependency_config={"number": DependencySettings(value_factory=set_value(50))}
)
container.register(
Client,
name="UsesDefaultValue"
)
container.register(
Client,
name="IgnoresDefaultParameterValue",
dependency_config={"number": DependencySettings(value_factory=dont_use_default_parameter)}
)
container.register(
Client,
name="UsesRegistrationFilter",
dependency_config={"number": DependencySettings(value_factory=dont_use_default_parameter, filter=with_name("One"))}
)
client1 = container.resolve(Client, filter=with_name("SetsValue"))
client2 = container.resolve(Client, filter=with_name("UsesDefaultValue"))
client3 = container.resolve(Client, filter=with_name("IgnoresDefaultParameterValue"))
client4 = container.resolve(Client, filter=with_name("UsesRegistrationFilter"))
client1.get_number() # returns 50
client2.get_number() # returns 10
client3.get_number() # returns 2
client4.get_number() # returns 1
```
The order of a dependant value is as follows
1. Setting the dependency value_factory to an explicit value
```python
DependencySettings(value_factory=set_value(50))
```
If the falue is a default parameter then the default value factory will use that default parameter value
```python
class Client:
def __init__(self, number=10):
self.number = number
```
If you don't want to use the default parameter value you can change the value_factory to pybass it
```python
DependencySettings(value_factory=dont_use_default_parameter)
```
2. Going to the container registry to find a registration using the registration filter if, if there is a default value on the dependant paramater you must explicity set.
## Tags
Tags can be added to registrations in order to support filtering. This can be useful as a means to filter registrations when resolving lists of a particular type
```python
class A:
pass
a1 = A()
a2 = A()
a3 = A()
container = Container()
container.register(A, instance=a1, tags=[Tag("a", "a1")])
container.register(A, instance=a2, tags=[Tag("a")])
container.register(A, instance=a3)
ar1 = container.resolve(A, filter=has_tag("a", "a1")) # returns a1
al1 = container.resolve(list[A], filter=has_tag("a")) # returns [a2, a1]
al2 = container.resolve(list[A], filter=has_tag("a", "a1")) # returns [a1]
al3 = container.resolve(list[A], filter=~has_tag("a", "a1")) # returns [a3, a2]
al4 = container.resolve(list[A], filter=~has_tag("a")) # returns [a3]
al5 = container.resolve(list[A]) # returns [a3, a2, a1]
```
## Parent Node Filters
Registrations can also specify that should only apply to certain parents objects by setting the parent_node_filter
```python
class A:
pass
class B(A):
pass
class C(A):
pass
class D:
def __init__(self, a: A):
self.a = a
class E:
def __init__(self, a: A):
self.a = a
container = Container()
container.register(A, B, parent_node_filter=implementation_type_is(E))
container.register(A, C, parent_node_filter=implementation_type_is(D))
container.register(D)
container.register(E)
e = container.resolve(E)
d = container.resolve(D)
type(e.a) # returns B
type(d.a) # returns C
```
## Accessing the Container, Scope and Resolver within dependencies
Accessing container directly
```python
class Client:
def __init__(self, container: Container):
self.container = container
def get_number(self):
return self.container.resolve(int)
container.register(int, instance=2)
container.register(Client)
client = container.resolve(Client)
client.get_number() # returns 2
```
Accessing Resolver also returns the container
```python
class Client:
def __init__(self, resolver: Resolver):
self.resolver = resolver
def get_number(self):
return self.resolver.resolve(int)
container.register(int, instance=2)
container.register(Client)
client = container.resolve(Client)
client.get_number() # returns 2
```
When within a scope, Resolver returns the current scope rather than the container
```python
class Client:
def __init__(self, resolver: Resolver):
self.resolver = resolver
def get_number(self):
return self.resolver.resolve(int)
container.register(int, instance=2)
container.register(Client)
client = container.resolve(Client)
client.get_number() # returns 2
with container.get_scope() as scope:
scope.register(int, instance=10)
scoped_client = scope.resolve(Client)
scoped_client.get_number() # returns 10
```
Scopes can also be used as an async context manager
```python
class Client:
async def get_number(self):
return 10
container.register(Client)
async with container.get_scope() as scope:
scoped_client = scope.resolve(Client)
await scoped_client.get_number() # returns 10
```
## Bundles
A bundle is a just a function that accepts a container, it can be used to set up related registrations on the container
```python
class ClientDependency:
def get_int(self):
return 10
class Client:
def __init__(self, dep: ClientDependency):
self.dep = dep
def get_number(self):
return self.dep.get_int()
def client_bundle(c: Container):
c.register(ClientDependency)
c.register(Client)
container.apply_bundle(client_bundle)
client = container.resolve(Client)
client.get_number() # returns 10
```
### Helper for bundles
There is now a ```BaseBundle``` class that gives you a bit more safety around running a module twice etc. Also you might want to pass in instances into the module.
You can find the ```BaseBundle``` in ```clean_ioc.bundles``` module
```python
@dataclass
class ClientConfig:
url: str
class Client:
def __init__(self, config: ClientConfig):
self.base_url = config.url
def get_thing(self):
# Do some requests stuff here
pass
class ClientBundle(BaseBundle):
def __init__(self, config: ClientConfig):
self.config = config
def apply(self, c: Container):
c.register(ClientConfig, instance=self.config)
c.register(Client)
client_config = ClientConfig(
url = "https://example.com"
)
container.apply_bundle(ClientBundle(config=client_config))
client = container.resolve(Client)
client.get_thing()
```
## Dependency Context (BETA feature)
You can inject a special type into your dependants that allows you to inspect the current dependency tree. For instances you can check the parent of the current class you are constructing
One example of where this becomes useful is if injecting a logger, you can get information about the loggers parent to add extra context
```python
class Client:
def __init__(self, logger: logging.Logger):
self.logger = logger
def do_a_thing(self):
self.logger.info('Doing a thing')
def logger_fac(context: DependencyContext):
module = context.parent.implementation.__module__
return logging.getLogger(module)
container = Container()
container.register(Client)
container.register(logging.Logger, factory=logger_fac, lifespan=Lifespan.transient)
client = container.resolve(Client)
```
***Note*** *If using dependency context on your dependency it's recommended that you use a lifespan of **transient**, because any other lifespan will create only use the parent of the first resolved instance*
## Pre-configurations
Pre configurations run a side-effect for a type before the type gets resolved.
This is useful if some python modules have some sort of module level functions that need to be called before the object get created
```python
import logging
class Client:
def __init__(self, logger: logging.Logger):
self.logger = logger
def do_a_thing(self):
self.logger.info('Doing a thing')
def logger_fac(context: DependencyContext):
module = context.parent.implementation.__module__
return logging.getLogger(module)
def configure_logging():
logging.basicConfig()
container = Container()
container.register(Client)
container.register(logging.Logger, factory=logger_fac, lifespan=Lifespan.transient)
container.pre_configure(logging.Logger, configure_logging)
client = container.resolve(Client)
```
Raw data
{
"_id": null,
"home_page": "https://github.com/peter-daly/clean_ioc",
"name": "clean-ioc",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.10",
"maintainer_email": null,
"keywords": null,
"author": "Peter Daly",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/2e/e3/dd629921e8e3f1c603e7b5936f2d6739d8e7b714e54cc6568fa42b158827/clean_ioc-0.18.1.tar.gz",
"platform": null,
"description": "# Clean IoC\nA simple dependency injection library for python that requires nothing of your application code (except that you use typing).\n\n\n## Basic Registering and resolving\n\nThere are 4 basic modes of registering a new set of classes\n\n### Implementation\n\n```python\n\nclass UserRepository(abc.ABC):\n @abc.abstractmethod\n def add(self, user):\n pass\n\nclass InMemoryUserRepository(UserRepository):\n\n def __init__(self):\n self.users = []\n\n def add(self, user):\n # This is obviously terrible, but it's for demo purposes\n self.users.append(user)\n\nclass SqlAlchemyUserRepository(UserRepository):\n\n def __init__(self):\n # Do some db stuff here\n pass\n\n def add(self, user):\n # Do some db stuff here\n pass\n\ncontainer = Container()\ncontainer.register(UserRepository, InMemoryUserRepository)\n\n\nrepository = container.resolve(UserRepository) # This will return an InMemoryUserRepository\n\n```\n\n### Concrete Class\n\n```python\n\nclass ClientDependency:\n def get_int(self):\n return 10\n\nclass Client:\n def __init__(self, dep: ClientDependency):\n self.dep = dep\n\n def get_number(self):\n return self.dep.get_int()\n\n\ncontainer = Container()\ncontainer.register(ClientDependency)\ncontainer.register(Client)\n\nclient = container.resolve(Client)\n\nclient.get_number() # returns 10\n\n```\n\n### Factory\n\n```python\n\nclass ClientDependency:\n def get_int(self):\n return 10\n\nclass Client:\n def __init__(self, dep: ClientDependency):\n self.dep = dep\n\n def get_number(self):\n return self.dep.get_int()\n\ndef client_factory(dep: ClientDependency):\n return Client(dep=dep)\n\n\ncontainer = Container()\ncontainer.register(ClientDependency)\ncontainer.register(Client, factory=client_factory)\n\nclient = container.resolve(Client)\n\nclient.get_number() # returns 10\n\n```\n\n\n### Instance\n\n```python\n\nclass ClientDependency:\n def __init__(self, num):\n self.num = num\n\n def get_int(self):\n return self.num\n\nclass Client:\n def __init__(self, dep: ClientDependency):\n self.dep = dep\n\n def get_number(self):\n return self.dep.get_int()\n\nclient_dependency = ClientDependency(num=10)\n\ncontainer = Container()\ncontainer.register(ClientDependency, instance=client_dependency)\ncontainer.register(Client)\n\nclient = container.resolve(Client)\n\nclient.get_number() # returns 10\n\n```\n\n## List resolving\n\nIf you have multiple dependencues you can simply define a dependency as a list[T] and you can return all of the instances.\n\n```python\n\nclass ClientDependency:\n def __init__(self, numbers: list[int]):\n self.numbers = numbers\n\n def get_numbers(self):\n return self.numbers\n\nclass Client:\n def __init__(self, dep: ClientDependency):\n self.dep = dep\n\n def get_numbers(self):\n return self.dep.get_numbers()\n\ncontainer = Container()\ncontainer.register(ClientDependency)\ncontainer.register(Client)\ncontainer.register(int, instance=1)\ncontainer.register(int, instance=2)\ncontainer.register(int, instance=3)\n\nclient = container.resolve(Client)\n\nclient.get_numbers() # returns [3, 2, 1]\n```\n\n\n## Decorators\n\nFollows a object orientated decoration pattern, rather than a decoration annotation.\nThe main reason for this was to allow decotation of registered instances\n\n```python\nclass Client:\n def __init__(self, number: int):\n self.number = number\n\n def get_number(self):\n return self.number\n\n\nclass DoubleClientDecorator(Client):\n def __init__(self, client: Client):\n self.client = client\n def get_number(self):\n return self.client.get_number() * 2\n\ncontainer = Container()\n\ncontainer.register(Client)\ncontainer.register_decorator(Client, DoubleClientDecorator)\ncontainer.register(int, instance=10)\n\nclient = container.resolve(Client)\n\nclient.get_number() # returns 20\n```\n\n\nDecorators are resolved in order of when first registered. So the first registered decorator is the highest in the class tree\n\n\n```python\n class Concrete:\n pass\n\n class DecoratorOne(Concrete):\n def __init__(self, child: Concrete):\n self.child = child\n\n class DecoratorTwo(Concrete):\n def __init__(self, child: Concrete):\n self.child = child\n\n container = Container()\n\n container.register(Concrete)\n container.register_decorator(Concrete, DecoratorOne)\n container.register_decorator(Concrete, DecoratorTwo)\n\n root = container.resolve(Concrete)\n\n type(root) # returns DecoratorOne\n type(root.child) # returns DecoratorTwo\n type(root.child.child) # returns Concrete\n```\n\n\n## Subclasses registration\n\nThis feature allows registration of all subclasses of a giveb type\n\n```python\nclass Client(abc.ABC):\n @abc.abstractmethod\n def get_number(self):\n pass\n\n\nclass TenClient(Client):\n def get_number(self):\n return 10\n\nclass TwentyClient(Client):\n def get_number(self):\n return 20\n\ncontainer = Container()\n\ncontainer.register_subclasses(Client)\n\nten_client = container.resolve(TenClient)\nten_client.get_number() # returns 10\n\ntwenty_client = container.resolve(TwentyClient)\ntwenty_client.get_number() # returns 20\n\n# Resolve all subsclasses of Client\nclient = container.resolve(list[Client]) ## [TwentyClient(), TenClient()]\n```\n\n\n## Lifespans\nLifespans configure how long and resolved object says alive for\nThere are 4 lifespan types\n\n### transient\nAlways create a new instance\n\n```python\ncontainer.register(Client, lifespan=Lifespan.transient)\n```\n\n\n### once_per_graph (Default behaviour)\nOnly create one instance throughout the resolve call\n\n```python\ncontainer.register(Client, lifespan=Lifespan.once_per_graph)\n```\n\n### scoped\nOnly create a new instance through the lifetime a [scope](#scopes). When not in a scope the behaviour is the same as **once_per_graph**.\n\n```python\ncontainer.register(Client, lifespan=Lifespan.scoped)\n```\n\n### singleton\nOnly one instance of the object is created throughout the lifespan of the container\n\n```python\ncontainer.register(Client, lifespan=Lifespan.singleton)\n```\n\n*Note:*\nWhen registering an instance, then the behaviour is always singleton\n\n```python\ncontainer.register(int, instance=10)\n```\n\n## Open Generics\n\nRegisters all generic subclasses of the service type and allows you to resolve with the generic alias\n\n```python\nT = TypeVar(\"T\")\n\nclass HelloCommand:\n pass\n\nclass GoodbyeCommand:\n pass\n\nclass CommandHandler(Generic[T]):\n def handle(self, command: T):\n pass\n\nclass HelloCommandHandler(CommandHandler[HelloCommand]):\n def handle(self, command: HelloCommand):\n print('HELLO')\n\nclass GoodbyeCommandHandler(CommandHandler[GoodbyeCommand]):\n def handle(self, command: GoodbyeCommand):\n print('GOODBYE')\n\ncontainer = Container()\ncontainer.register_open_generic(CommandHandler)\n\nh1 = container.resolve(CommandHandler[HelloCommand])\nh2 = container.resolve(CommandHandler[GoodbyeCommand])\n\nh1.handle(HelloCommand()) # prints 'HELLO'\nh2.handle(GoodbyeCommand()) # prints 'GOODBYE'\n\n```\n\n## Open Generic Decorators\n\n\nAllows you to add decorators to your open generic registrations\n\n```python\nT = TypeVar(\"T\")\n\nclass HelloCommand:\n pass\n\nclass GoodbyeCommand:\n pass\n\nclass CommandHandler(Generic[T]):\n def handle(self, command: T):\n pass\n\nclass HelloCommandHandler(CommandHandler[HelloCommand]):\n def handle(self, command: HelloCommand):\n print('HELLO')\n\nclass GoodbyeCommandHandler(CommandHandler[GoodbyeCommand]):\n def handle(self, command: GoodbyeCommand):\n print('GOODBYE')\n\nclass AVeryBigCommandHandlerDecorator(Generic[T]):\n def __init__(self, handler: CommandHandler[T]):\n self.handler = handler\n\n def handle(self, command: T):\n print('A VERY BIG')\n self.handler.handle(command=command)\n\ncontainer = Container()\ncontainer.register_open_generic(CommandHandler)\ncontainer.register_open_generic_decorator(CommandHandler, AVeryBigCommandHandlerDecorator)\nh1 = container.resolve(CommandHandler[HelloCommand])\nh2 = container.resolve(CommandHandler[GoodbyeCommand])\n\nh1.handle(HelloCommand()) # prints 'A VERY BIG\\nHELLO'\nh2.handle(GoodbyeCommand()) # prints 'A VERY BIG\\nGOODBYE'\n\n```\n\n## Scopes\nScopes are a machanism where you guarantee that dependency can be temporarily a singleton within the scope. You can also register dependencies that that are only available withon the scope.\nSome good use cases for scope lifetimes are:\n - http request in a web server\n - message/event if working on a message based system\nFor instance you could keep an single database connection open for the entire lifetime of the http request\n\n```python\nclass DbConnection:\n def run_sql(self, statement):\n # Done some sql Stuff\n pass\n\ncontainer.register(DbConnection, lifespan=Lifespan.scoped)\n\nwith container.get_scope() as scope:\n db_conn = scope.resolve(DbConnection)\n db_conn.run_sql(\"UPDATE table SET column = 1\")\n```\n\nScopes can also be use with asyncio\n\n```python\nclass AsyncDbConnection:\n async def run_sql(self, statement):\n # Done some sql Stuff\n pass\n\ncontainer.register(AsyncDbConnection, lifespan=Lifespan.scoped)\n\nasync with container.get_scope() as scope:\n db_conn = scope.resolve(AsyncDbConnection)\n await db_conn.run_sql(\"UPDATE table SET column = 1\")\n```\n\n\n### Scoped Teardowns\nWhen you are finished with some dependenies within a scope you might want to perform some teardown action before you exit the scope. For example if we want to close our db connection.\n\n```python\nclass AsyncDbConnection:\n async def run_sql(self, statement):\n # Done some sql Stuff\n pass\n async def close(self):\n # Close the connection\n pass\n\nasync def close_connection(conn: AsyncDbConnection):\n await conn.close()\n\ncontainer.register(DbConnection, lifespan=Lifespan.scoped, scoped_teardown=close_connection)\n\nasync with container.get_scope() as scope:\n db_conn = scope.resolve(AsyncDbConnection)\n await db_conn.run_sql(\"UPDATE table SET column = 1\")\n\n# close connection is run when we exit the scope\n```\n\n***Note***: *When using the scope as an async context manager you need both sync and async teardowns are run, when a scope is used as a normal sync context manager async teardowns are ignored*\n\n\n## Named registrations & Registration filters\n\nBy default the last unnamed registration is what the container will return when resolve is called as below.\n\n```python\n\ncontainer = Container()\ncontainer.register(int, instance=1)\ncontainer.register(int, instance=2)\ncontainer.register(int, instance=3)\n\nnumber = container.resolve(int) # returns 3\n\n```\nTo be more selective of what we return we can add a name to the registration and apply a registration filter when we resolve.\n\nA registration filter is simply function that receives a **Registration** and returns a **bool**\n\nFor example if we wanted to get the int named **\"One\"** we do the following\n\n```python\ncontainer = Container()\ncontainer.register(int, instance=1, name=\"One\")\ncontainer.register(int, instance=2, name=\"Two\")\ncontainer.register(int, instance=3, name=\"Three\")\n\nnumber = container.resolve(int, filter=lambda r: r.name == \"One\") # returns 1\n```\n\nClean IOC comes with a set of in built registration filters that can be found [here](./clean_ioc/registration_filters.py)\n\nWe can get the desired behaviour as above\n```python\nfrom clean_ioc.registration_filters import with_name\n\ncontainer = Container()\ncontainer.register(int, instance=1, name=\"One\")\ncontainer.register(int, instance=2, name=\"Two\")\ncontainer.register(int, instance=3, name=\"Three\")\n\nnumber = container.resolve(int, filter=with_name(\"One\")) # returns 1\n```\n\n## Dependency Settings\n\nDependency settings are defined at registration and allow you to define the selection or setting dependencies\n\n\n```python\nclass Client:\n def __init__(self, number=10):\n self.number = number\n\n def get_number(self):\n return self.number\n\ncontainer = Container()\n\ncontainer.register(int, instance=1, name=\"One\")\ncontainer.register(int, instance=2)\n\ncontainer.register(\n Client,\n name=\"SetsValue\",\n dependency_config={\"number\": DependencySettings(value_factory=set_value(50))}\n)\ncontainer.register(\n Client,\n name=\"UsesDefaultValue\"\n)\ncontainer.register(\n Client,\n name=\"IgnoresDefaultParameterValue\",\n dependency_config={\"number\": DependencySettings(value_factory=dont_use_default_parameter)}\n)\ncontainer.register(\n Client,\n name=\"UsesRegistrationFilter\",\n dependency_config={\"number\": DependencySettings(value_factory=dont_use_default_parameter, filter=with_name(\"One\"))}\n)\n\nclient1 = container.resolve(Client, filter=with_name(\"SetsValue\"))\nclient2 = container.resolve(Client, filter=with_name(\"UsesDefaultValue\"))\nclient3 = container.resolve(Client, filter=with_name(\"IgnoresDefaultParameterValue\"))\nclient4 = container.resolve(Client, filter=with_name(\"UsesRegistrationFilter\"))\n\n\nclient1.get_number() # returns 50\nclient2.get_number() # returns 10\nclient3.get_number() # returns 2\nclient4.get_number() # returns 1\n```\n\nThe order of a dependant value is as follows\n1. Setting the dependency value_factory to an explicit value\n ```python\n DependencySettings(value_factory=set_value(50))\n ```\n If the falue is a default parameter then the default value factory will use that default parameter value\n ```python\n class Client:\n def __init__(self, number=10):\n self.number = number\n ```\n If you don't want to use the default parameter value you can change the value_factory to pybass it\n ```python\n DependencySettings(value_factory=dont_use_default_parameter)\n ```\n2. Going to the container registry to find a registration using the registration filter if, if there is a default value on the dependant paramater you must explicity set.\n\n\n## Tags\n\nTags can be added to registrations in order to support filtering. This can be useful as a means to filter registrations when resolving lists of a particular type\n\n```python\nclass A:\n pass\n\na1 = A()\na2 = A()\na3 = A()\n\ncontainer = Container()\n\ncontainer.register(A, instance=a1, tags=[Tag(\"a\", \"a1\")])\ncontainer.register(A, instance=a2, tags=[Tag(\"a\")])\ncontainer.register(A, instance=a3)\n\nar1 = container.resolve(A, filter=has_tag(\"a\", \"a1\")) # returns a1\nal1 = container.resolve(list[A], filter=has_tag(\"a\")) # returns [a2, a1]\nal2 = container.resolve(list[A], filter=has_tag(\"a\", \"a1\")) # returns [a1]\nal3 = container.resolve(list[A], filter=~has_tag(\"a\", \"a1\")) # returns [a3, a2]\nal4 = container.resolve(list[A], filter=~has_tag(\"a\")) # returns [a3]\nal5 = container.resolve(list[A]) # returns [a3, a2, a1]\n```\n\n\n\n## Parent Node Filters\n\nRegistrations can also specify that should only apply to certain parents objects by setting the parent_node_filter\n\n```python\nclass A:\n pass\n\nclass B(A):\n pass\n\nclass C(A):\n pass\n\nclass D:\n def __init__(self, a: A):\n self.a = a\n\nclass E:\n def __init__(self, a: A):\n self.a = a\n\ncontainer = Container()\n\ncontainer.register(A, B, parent_node_filter=implementation_type_is(E))\ncontainer.register(A, C, parent_node_filter=implementation_type_is(D))\ncontainer.register(D)\ncontainer.register(E)\n\ne = container.resolve(E)\nd = container.resolve(D)\n\ntype(e.a) # returns B\ntype(d.a) # returns C\n\n```\n\n\n\n## Accessing the Container, Scope and Resolver within dependencies\n\nAccessing container directly\n\n```python\nclass Client:\n def __init__(self, container: Container):\n self.container = container\n\n def get_number(self):\n return self.container.resolve(int)\n\ncontainer.register(int, instance=2)\n\ncontainer.register(Client)\n\nclient = container.resolve(Client)\nclient.get_number() # returns 2\n```\n\nAccessing Resolver also returns the container\n\n```python\n\nclass Client:\n def __init__(self, resolver: Resolver):\n self.resolver = resolver\n\n def get_number(self):\n return self.resolver.resolve(int)\n\ncontainer.register(int, instance=2)\n\ncontainer.register(Client)\n\nclient = container.resolve(Client)\nclient.get_number() # returns 2\n```\n\nWhen within a scope, Resolver returns the current scope rather than the container\n\n```python\nclass Client:\n def __init__(self, resolver: Resolver):\n self.resolver = resolver\n\n def get_number(self):\n return self.resolver.resolve(int)\n\ncontainer.register(int, instance=2)\n\ncontainer.register(Client)\n\nclient = container.resolve(Client)\nclient.get_number() # returns 2\n\nwith container.get_scope() as scope:\n scope.register(int, instance=10)\n scoped_client = scope.resolve(Client)\n scoped_client.get_number() # returns 10\n```\n\nScopes can also be used as an async context manager\n\n```python\nclass Client:\n async def get_number(self):\n return 10\n\ncontainer.register(Client)\n\nasync with container.get_scope() as scope:\n scoped_client = scope.resolve(Client)\n await scoped_client.get_number() # returns 10\n```\n\n## Bundles\n\n\nA bundle is a just a function that accepts a container, it can be used to set up related registrations on the container\n\n```python\nclass ClientDependency:\n def get_int(self):\n return 10\n\nclass Client:\n def __init__(self, dep: ClientDependency):\n self.dep = dep\n\n def get_number(self):\n return self.dep.get_int()\n\ndef client_bundle(c: Container):\n c.register(ClientDependency)\n c.register(Client)\n\ncontainer.apply_bundle(client_bundle)\n\nclient = container.resolve(Client)\n\nclient.get_number() # returns 10\n```\n\n### Helper for bundles\n\nThere is now a ```BaseBundle``` class that gives you a bit more safety around running a module twice etc. Also you might want to pass in instances into the module.\nYou can find the ```BaseBundle``` in ```clean_ioc.bundles``` module\n\n\n\n```python\n@dataclass\nclass ClientConfig:\n url: str\n\nclass Client:\n def __init__(self, config: ClientConfig):\n self.base_url = config.url\n\n def get_thing(self):\n # Do some requests stuff here\n pass\n\n\n\nclass ClientBundle(BaseBundle):\n\n def __init__(self, config: ClientConfig):\n self.config = config\n\n def apply(self, c: Container):\n c.register(ClientConfig, instance=self.config)\n c.register(Client)\n\n\n\nclient_config = ClientConfig(\n url = \"https://example.com\"\n)\n\ncontainer.apply_bundle(ClientBundle(config=client_config))\n\nclient = container.resolve(Client)\n\nclient.get_thing()\n```\n\n\n## Dependency Context (BETA feature)\n\nYou can inject a special type into your dependants that allows you to inspect the current dependency tree. For instances you can check the parent of the current class you are constructing\nOne example of where this becomes useful is if injecting a logger, you can get information about the loggers parent to add extra context\n\n```python\nclass Client:\n def __init__(self, logger: logging.Logger):\n self.logger = logger\n\n def do_a_thing(self):\n self.logger.info('Doing a thing')\n\ndef logger_fac(context: DependencyContext):\n module = context.parent.implementation.__module__\n return logging.getLogger(module)\n\n\ncontainer = Container()\ncontainer.register(Client)\ncontainer.register(logging.Logger, factory=logger_fac, lifespan=Lifespan.transient)\nclient = container.resolve(Client)\n```\n\n***Note*** *If using dependency context on your dependency it's recommended that you use a lifespan of **transient**, because any other lifespan will create only use the parent of the first resolved instance*\n## Pre-configurations\n\nPre configurations run a side-effect for a type before the type gets resolved.\nThis is useful if some python modules have some sort of module level functions that need to be called before the object get created\n\n```python\nimport logging\n\nclass Client:\n def __init__(self, logger: logging.Logger):\n self.logger = logger\n\n def do_a_thing(self):\n self.logger.info('Doing a thing')\n\ndef logger_fac(context: DependencyContext):\n module = context.parent.implementation.__module__\n return logging.getLogger(module)\n\ndef configure_logging():\n logging.basicConfig()\n\n\n\n\ncontainer = Container()\ncontainer.register(Client)\ncontainer.register(logging.Logger, factory=logger_fac, lifespan=Lifespan.transient)\ncontainer.pre_configure(logging.Logger, configure_logging)\n\nclient = container.resolve(Client)\n\n\n```",
"bugtrack_url": null,
"license": "MIT",
"summary": "An IOC Container for Python 3.10+",
"version": "0.18.1",
"project_urls": {
"Documentation": "https://github.com/peter-daly/clean_ioc",
"Homepage": "https://github.com/peter-daly/clean_ioc",
"Repository": "https://github.com/peter-daly/clean_ioc"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "abe8aecacd221f6c0ffcdeceb3b23f8a2459de266d78cc9a9b44a27075bf4667",
"md5": "22ce93728382de9e4897e8cfdece32ad",
"sha256": "11545f6ea070cad9455d521b5be9c8ebfd6a40a8d266ea1b8fc672c6b2f30876"
},
"downloads": -1,
"filename": "clean_ioc-0.18.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "22ce93728382de9e4897e8cfdece32ad",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.10",
"size": 26205,
"upload_time": "2024-05-08T11:17:52",
"upload_time_iso_8601": "2024-05-08T11:17:52.429834Z",
"url": "https://files.pythonhosted.org/packages/ab/e8/aecacd221f6c0ffcdeceb3b23f8a2459de266d78cc9a9b44a27075bf4667/clean_ioc-0.18.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "2ee3dd629921e8e3f1c603e7b5936f2d6739d8e7b714e54cc6568fa42b158827",
"md5": "155955a3252d6d32953b14368ad855b2",
"sha256": "fcaa3121c7180c85c7c78bbd01bf7e7e19944ab7197c8ca2a4bdc5e8b674c3d4"
},
"downloads": -1,
"filename": "clean_ioc-0.18.1.tar.gz",
"has_sig": false,
"md5_digest": "155955a3252d6d32953b14368ad855b2",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.10",
"size": 26925,
"upload_time": "2024-05-08T11:17:53",
"upload_time_iso_8601": "2024-05-08T11:17:53.917310Z",
"url": "https://files.pythonhosted.org/packages/2e/e3/dd629921e8e3f1c603e7b5936f2d6739d8e7b714e54cc6568fa42b158827/clean_ioc-0.18.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-05-08 11:17:53",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "peter-daly",
"github_project": "clean_ioc",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "clean-ioc"
}
JSON
