Dependency injection container made for python
pip install kink
If you don't know poetry, I highly recommend visiting their webpage
poetry add kink
Because python is a multi paradigm language and this should encourage you to use best OOP practices improving your workflow and your code and have more time for your hobbies and families instead monkey-patching entire world.
Dependency happens when one component (component might be a class, or a function) A uses other component
B. We say than that A depends on B.
Instead hardcoding dependency inside your components and making your code tightly coupled
you are loosing it by providing(injecting) required behaviour either by subclassing or
plugging additional code. This is called Inversion of Control which keeps your code
oriented around behaviour rather than control. There are many benefits coming out of it:
This is where dependency injection comes in place. Dependency injection is a specific
style of inversion of control, which generally says instead hardcoding dependency pass
dependant object as a parameter to a method rather than having method creating it itself.
( who would thought it is so easy :)? ). It can go even further than that; when you pass
a dependency don't rely on a particular implementation rely on an abstraction (Dependency Inversion Principle).
So you might ask why do I need it? Here is couple reasons:
Coding is hard enough ( business requirements are changing all the time, deadlines are shortening, clients wants more, there are so many unknowns you have to figure out), relying on unpredictable state makes it even harder:
True, that is why you should pick up Dependency Injection Container to do all this work
for you. Kink gives you one decorator and simple dict-like object to bootstrap and manipulate
your container.
No need for manual work and manual dependency management. Give it a try and you will love it!
To fully utilise the potential of kink it is recommended to bootstrap your initial dependencies
(config values, or instances of classes that are standalone, requires no other dependencies than themselves).
Some people prefer to keep it in __init__.py in the top module of your application, other
create separate bootstrap.py file for this purpose. Once all is setup the only step left
is to decorate your classes/functions with @inject decorator.
Dependency container is a dict-like object, adding new service to dependency container is as simple as the following example:
from kink import di
from os import getenv
di["db_name"] = getenv("DB_NAME")
di["db_password"] = getenv("DB_PASSWORD")
Kink also supports on-demand service creation. In order to define such a service, lambda function should be used:
from kink import di
from sqlite3 import connect
di["db_connection"] = lambda di: connect(di["db_name"])
In this scenario connection to database will not be established until service is requested.
Factorised services are services that are instantiated every time they are requested.
from kink import di
from sqlite3 import connect
di.factories["db_connection"] = lambda di: connect(di["db_name"])
connection_1 = di["db_connection"]
connection_2 = di["db_connection"]
connection_1 != connection_2
In the above example we defined factorised service db_connection, and below by accessing the service from di we created
two separate connection to database.
To access given service just reference it inside di like you would do this with
a normal dictionary, full example below:
from kink import di
from sqlite3 import connect
# Bootstrapping
di["db_name"] = "test_db.db"
di["db_connection"] = lambda di: connect(di["db_name"])
# Getting a service
connection = di["db_connection"] # will return instance of sqlite3.Connection
assert connection == di["db_connection"] # True
Autowiring is the ability of the container to automatically create and inject dependencies. It detects dependencies of the component tries to search for references in the container and if all references are present an instance of requested service is returned.
Autowiring system in kink works in two ways:
Autowiring mechanism priorities dependencies automatically, so when multiple matches are found for the service this is how it works; Firstly passed arguments are prioritied - if you pass arguments manually to the service they will take precendence over anything else. Next argument's names are taken into consideration and last but not least argument's type annotations.
If you don't like type annotations or would like to take advantage of autowiring's precedence mechanism use this style.
This is a very simple mechanism we have already seen in previous examples. Autowiring system checks function argument's names and tries to search for services with the same names inside the container.
If you are like me and like type annotations and use static analysis tools this is a preferred way working with DI container.
In this scenario names are ignored instead argument's type annotations are inspected and looked up inside di container. This requires aliases when bootstrapping your services in DI container or simply adding them to container in the way that its type is the key by which service is accessed. Please consider the following example:
from kink import di, inject
from sqlite3 import connect, Connection
di["db_name"] = "test_db.db"
di[Connection] = lambda di: connect(di["db_name"]) # sqlite connection can be accessed by its type
@inject # Constructor injection will happen here
class UserRepository:
def __init__(self, db: Connection): # `db` argument will be resolved because `Connection` instance is present in the container.
self.db = db
repo = di[UserRepository]
assert repo.db == di[Connection] # True
from kink import inject, di
import MySQLdb
# Set dependencies
di["db_host"] = "localhost"
di["db_name"] = "test"
di["db_user"] = "user"
di["db_password"] = "password"
di["db_connection"] = lambda di: MySQLdb.connect(host=di["db_host"], user=di["db_user"], passwd=di["db_password"], db=di["db_name"])
@inject
class AbstractRepository:
def __init__(self, db_connection):
self.connection = db_connection
class UserRepository(AbstractRepository):
...
repository = di[UserRepository] # will retrieve instance of UserRepository from di container
repository.connection # mysql db connection is resolved and available to use.
When class is annotated by inject annotation it will be automatically added to the container for future use (eg autowiring).
When you register a service with @inject decorator you can attach your own alias name, please consider the following example:
from kink import inject
from typing import Protocol
class IUserRepository(Protocol):
...
@inject(alias=IUserRepository)
class UserRepository:
...
assert di[IUserRepository] == di[UserRepository] # returns true
For more examples check tests directory
Aliases in kink do not have to be unique, but by default when autowiring mechnism is called the service that
was registered first within given alias will be returned. If for some reason you would like to retrieve all
services that alias to the same name (eg implementing strategy pattern), kink provides a useful functionality
for doing so. Please consider the following example:
from kink import inject
from typing import Protocol, List
class IUserRepository(Protocol):
...
@inject(alias=IUserRepository)
class MongoUserRepository:
...
@inject(alias=IUserRepository)
class MySQLUserRepository:
...
@inject()
class UserRepository:
def __init__(self, repos: List[IUserRepository]) -> None: # all services that alias to IUserRepository will be passed here
self._repos = repos
def store_to_mysql(self, user: ...):
self._repos[1].store(user)
def store_to_mongo(self, user: ...):
self._repos[0].store(user)
Sometimes it might come handy to clear cached services in di container. Simple way of
doing this is calling di.clear_cache() method like in the following example.
from kink import inject, di
... # set and accesss your services
di.clear_cache() # this will clear cache of all services inside di container that are not factorised services
from fastapi import APIRouter, Depends, status
from fastapi.responses import JSONResponse, Response
from kink import di
router = APIRouter()
# register service in the DI container
di[ClientService] = ClientService()
@router.post(
"/clients",
response_model=ClientDTO,
responses={400: {"model": APIErrorMessage}, 500: {"model": APIErrorMessage}},
tags=["clients"],
)
async def create_client(
request: CreateClientDTO, service: ClientService = Depends(lambda: di[ClientService])
) -> JSONResponse:
result = service.create(request)
return JSONResponse(content=result.dict(), status_code=status.HTTP_201_CREATED)
A complete example, together with tests you can find it here.
FAQs
Dependency injection for python.
We found that kink demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.
Security News
The Ultralytics' PyPI Package was compromised four times in one weekend through GitHub Actions cache poisoning and failure to rotate previously compromised API tokens.