Async dependency injection library
Project description
Async-first dependency injection library based on python type hints
Framework integrations:
Installation
Install using pip pip install aioinject
Example
import aioinject
class Database:
def __init__(self) -> None:
self._storage = {1: "Username"}
def get(self, id: int) -> str | None:
return self._storage.get(id)
class UserService:
def __init__(self, database: Database) -> None:
self._database = database
def get(self, id: int) -> str:
user = self._database.get(id)
if user is None:
raise ValueError
return user
container = aioinject.Container()
container.register(aioinject.Singleton(Database))
container.register(aioinject.Singleton(UserService))
def main() -> None:
with container.sync_context() as ctx:
service = ctx.resolve(UserService)
user = service.get(1)
assert user == "Username"
print(user)
if __name__ == "__main__":
main()
Injecting dependencies into a function
You can inject dependencies into a function using @inject
decorator,
but that's usually only necessary if you're working with a framework:
import contextlib
from collections.abc import AsyncIterator
from contextlib import aclosing
from typing import Annotated
import uvicorn
from fastapi import FastAPI
from aioinject import Container, Inject, Singleton
from aioinject.ext.fastapi import AioInjectMiddleware, inject
class Service:
pass
container = Container()
container.register(Singleton(Service))
@contextlib.asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
async with aclosing(container):
yield
app = FastAPI(lifespan=lifespan)
app.add_middleware(AioInjectMiddleware, container=container)
@app.get("/")
@inject
async def root(service: Annotated[Service, Inject]) -> str:
return str(service)
if __name__ == "__main__":
uvicorn.run("main:app")
Using multiple dependencies with same type
If you have multiple implementations for the same dependency
you can use typing.NewType
to differentiate between them:
import dataclasses
from typing import Annotated, NewType
from aioinject import Container, providers, inject, Inject
@dataclasses.dataclass
class Client:
name: str
GitHubClient = NewType("GitHubClient", Client)
GitLabClient = NewType("GitLabClient", Client)
def get_github_client() -> GitHubClient:
return GitHubClient(Client(name="GitHub Client"))
def get_gitlab_client() -> GitLabClient:
return GitLabClient(Client(name="GitLab Client"))
container = Container()
container.register(providers.Scoped(get_github_client))
container.register(providers.Scoped(get_gitlab_client))
with container.sync_context() as ctx:
github_client = ctx.resolve(GitHubClient)
gitlab_client = ctx.resolve(GitLabClient)
print(github_client, gitlab_client)
Working with Resources
Often you need to initialize and close a resource (file, database connection, etc...),
you can do that by using a contextlib.(async)contextmanager
that would return your resource.
Aioinject would automatically close them when context
is closed,
or when you call container.aclose()
if your dependency is a Singleton
.
import contextlib
from aioinject import Container, providers
class Session:
pass
@contextlib.contextmanager
def get_session() -> Session:
print("Startup")
yield Session()
print("Shutdown")
container = Container()
container.register(providers.Scoped(get_session))
with container.sync_context() as ctx:
session = ctx.resolve(Session) # Startup
session = ctx.resolve(Session) # Nothing is printed, Session is cached
# Shutdown
Async Dependencies
You can register async resolvers the same way as you do with other dependencies,
all you need to change is to use Container.context
instead of Container.sync_context
:
import asyncio
from aioinject import Container, providers
class Service:
pass
async def get_service() -> Service:
await asyncio.sleep(1)
return Service()
async def main() -> None:
container = Container()
container.register(providers.Scoped(get_service))
async with container.context() as ctx:
service = await ctx.resolve(Service)
print(service)
if __name__ == "__main__":
asyncio.run(main())
Providers
Scoped
Objects provided with Scoped
provider would be cached within context.
Transient
Transient
provider would provide different instances even if used within same context
Singleton
Singleton
works as you expect - there would be only one instance of a singleton
object, bound to a specific container
Object
Object
provider just returns an object provided to it, e.g. Object(42)
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for aioinject-0.20.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | c1ecdfee3c155e996aa5170cf5394a9c963d6691f671d5e3443ea7317c94a313 |
|
MD5 | cfdb6ba726290f199ce85db10ae71c9e |
|
BLAKE2b-256 | 742f6ab72468d766c836ab1690cd1f60b78b4e77924266f5b192c1cadb975de0 |