Lightweight, type-safe dependency injection for Python. One decorator, zero dependencies, full type inference
Getting Started · Features · Lifecycle · Errors · Testing · License
pip install injektafrom typing import Annotated
from injekta import Needs, inject
def get_db() -> Database:
return PostgresDB(os.environ["DATABASE_URL"])
@inject
def create_user(db: Annotated[Database, Needs(get_db)], name: str):
db.execute(f"INSERT INTO users (name) VALUES ('{name}')")
return {"created": name}
create_user(name="John") # db is resolved automaticallyThat's it. No configuration, no boilerplate, no framework required.
If you've used FastAPI's Depends, injekta is that same idea extracted into a standalone library. Most DI libraries in Python are either too complex for what they do, or too magical to reason about. injekta takes a different approach:
- A single decorator (
@inject) handles everything - Dependencies are declared in the function signature, not in external config
- Full type inference, your editor knows the types, mypy validates them
- Zero runtime dependencies, just the standard library
- Works with both sync and async functions
injekta supports four styles. Pick the one that fits your use case, or mix them freely.
The simplest form. Pass a callable to Needs and use it as a default value.
@inject
def handler(db: Database = Needs(get_db)):
...Injected parameters must come after regular parameters in this style.
Using Annotated places the dependency in the type hint, so parameter order is unrestricted.
from typing import Annotated
@inject
def handler(db: Annotated[Database, Needs(get_db)], name: str):
...For larger applications, register implementations in a Container and reference them by type.
from injekta import Container
container = Container()
container.register(Database, PostgresDB())
container.register(Logger, ConsoleLogger())
@inject
def handler(db: Database = container.Needs(Database)):
...Instances are singletons. Pass a class, lambda, or function to get a new value on each resolution:
container.register(Database, PostgresDB) # class factory
container.register(Database, lambda: PostgresDB("localhost")) # lambda factoryThe recommended style for production code. Combines type safety with free parameter ordering.
@inject
def handler(
db: Annotated[Database, container.Needs(Database)],
logger: Annotated[Logger, container.Needs(Logger)],
name: str,
):
logger.info(f"Creating {name}")
db.execute("INSERT INTO users ...")injekta works naturally with Python's Protocol for structural typing. No base classes required.
from typing import Protocol
class Database(Protocol):
def execute(self, query: str) -> None: ...
def fetch(self, query: str) -> list[dict]: ...
class Logger(Protocol):
def info(self, msg: str) -> None: ...
# Concrete implementations don't inherit from the protocol
class PostgresDB:
def execute(self, query: str) -> None: ...
def fetch(self, query: str) -> list[dict]: ...
container = Container()
container.register(Database, PostgresDB())@inject works on __init__ methods, making it straightforward to build service classes.
class UserService:
@inject
def __init__(
self,
db: Annotated[Database, container.Needs(Database)],
logger: Annotated[Logger, container.Needs(Logger)],
):
self.db = db
self.logger = logger
def create(self, name: str) -> None:
self.logger.info(f"Creating {name}")
self.db.execute(f"INSERT INTO users (name) VALUES ('{name}')")
service = UserService() # dependencies injected automaticallyBoth sync and async dependencies work transparently.
async def get_db() -> Database:
db = PostgresDB()
await db.connect()
return db
@inject
async def handler(db: Annotated[Database, Needs(get_db)]):
await db.fetch("SELECT * FROM users")Async factories work with Container too. Register an async def and use it from an async function:
async def make_db() -> Database:
db = PostgresDB(os.environ["DATABASE_URL"])
await db.connect()
return db
container.register(Database, make_db)
@inject
async def handler(db: Annotated[Database, container.Needs(Database)]):
await db.fetch("SELECT * FROM users")You can also resolve async factories directly with resolve_async():
db = await container.resolve_async(Database)Dependencies that need cleanup (database connections, HTTP sessions, file handles) can use yield instead of return. Code after yield runs automatically when the function returns:
def get_db() -> Generator[Database]:
db = PostgresDB(os.environ["DATABASE_URL"])
db.connect()
yield db
db.close() # runs after handler returns
@inject
def handler(db: Database = Needs(get_db)):
db.execute("INSERT INTO users ...")Async generators work the same way:
async def get_session() -> AsyncGenerator[ClientSession]:
session = ClientSession()
yield session
await session.close()
@inject
async def handler(session: Annotated[ClientSession, Needs(get_session)]):
await session.get("https://api.example.com")Cleanup runs even if the function raises an exception.
Dependencies can depend on other dependencies. injekta resolves the full tree.
def get_config() -> Config:
return Config.from_env()
def get_db(config: Config = Needs(get_config)) -> Database:
return PostgresDB(config.database_url)
def get_user_repo(db: Database = Needs(get_db)) -> UserRepository:
return UserRepository(db)
@inject
def handler(repo: UserRepository = Needs(get_user_repo)):
return repo.list_all()
# Resolves: get_config -> get_db -> get_user_repo -> handlerUnderstanding how injekta manages instance lifetime avoids surprises.
@inject analyzes the function signature once at decoration time and caches the dependency tree. Subsequent calls skip all introspection and go straight to resolution. This means there is zero reflection overhead per call.
Every call to the injected function re-executes the factory from scratch. There is no implicit caching between calls:
def get_db() -> Database:
print("connecting...")
return PostgresDB()
@inject
def handler(db: Database = Needs(get_db)):
...
handler() # prints "connecting..."
handler() # prints "connecting..." againWithin a single call, if the same factory appears in multiple branches of the dependency tree (diamond dependency), it is executed only once:
def get_config() -> Config:
return Config.from_env() # called once, not twice
def get_db(config: Config = Needs(get_config)) -> Database: ...
def get_cache(config: Config = Needs(get_config)) -> Cache: ...
@inject
def handler(
db: Database = Needs(get_db),
cache: Cache = Needs(get_cache),
):
...
handler() # get_config runs once, result shared by get_db and get_cachegraph TD
handler --> get_db
handler --> get_cache
get_db --> get_config
get_cache --> get_config
get_configis resolved once. The same instance is injected into bothget_dbandget_cache.
Container.register auto-detects the strategy based on what you pass:
| You register | Behavior | Example |
|---|---|---|
| An instance | Singleton. Same object returned every time. | container.register(Database, PostgresDB()) |
| A class | Factory. New instance created on each resolution. | container.register(Database, PostgresDB) |
| A function/lambda | Factory. Called on each resolution. | container.register(Database, lambda: PostgresDB("url")) |
| An async function | Async factory. Resolved via resolve_async() or @inject on async functions. |
container.register(Database, make_db) |
container = Container()
# Singleton: same connection reused everywhere
db = PostgresDB(os.environ["DATABASE_URL"])
container.register(Database, db)
# Factory (class): fresh instance per resolution, no constructor args
container.register(Logger, ConsoleLogger)
# Factory (lambda): fresh instance with custom arguments
container.register(Database, lambda: PostgresDB("localhost", 5432))
# Factory (function): same as lambda, useful for complex setup
def make_cache() -> RedisCache:
cache = RedisCache(os.environ["REDIS_URL"])
cache.ping()
return cache
container.register(Cache, make_cache)
# Async factory: for dependencies that need async setup
async def make_db() -> Database:
db = PostgresDB(os.environ["DATABASE_URL"])
await db.connect()
return db
container.register(Database, make_db)Factories can also be decorated with @inject to receive their own dependencies:
@inject
def make_service(db: Annotated[Database, container.Needs(Database)]) -> UserService:
return UserService(db)
container.register(UserService, make_service)If you need a singleton, instantiate it yourself and register the instance. If you need a fresh object every time, register a class, lambda, or function.
All four styles work together in the same function signature.
@inject
def handler(
db: Annotated[Database, container.Needs(Database)], # container, Annotated
config: Annotated[Config, Needs(get_config)], # factory, Annotated
cache: Cache = container.Needs(Cache), # container, default
metrics: Metrics = Needs(get_metrics), # factory, default
name: str = "", # regular parameter
):
...injekta raises clear, specific exceptions when something goes wrong:
| Exception | When | Phase |
|---|---|---|
ResolutionError |
Circular dependency, invalid signature | Decoration time (@inject) |
InjectionError |
Unregistered type, async dep in sync context, async factory resolved synchronously | Call time (function execution) |
InjektaError |
Base class for all injekta errors | Catch-all |
from injekta.exceptions import InjectionError, ResolutionErrorUnregistered type:
container.resolve(Database)
# InjectionError: No registration found for 'Database'Circular dependency:
def get_a(b=Needs(get_b)): ...
def get_b(a=Needs(get_a)): ...
@inject
def handler(a=Needs(get_a)): ...
# ResolutionError: Circular dependency detected for 'get_a'Async dependency in sync context:
async def get_db(): ...
@inject
def handler(db=Needs(get_db)): ...
handler()
# InjectionError: Cannot use async dependency 'get_db' in sync context.Async factory resolved synchronously:
async def make_db(): ...
container.register(Database, make_db)
container.resolve(Database)
# InjectionError: Cannot resolve async factory for 'Database' synchronously.
# Use resolve_async() or an async context with @inject.All exceptions inherit from InjektaError, so you can catch them all with a single except InjektaError.
Use container.override to swap dependencies in tests. The original registration is restored automatically when the context exits:
def test_create_user():
fake_db = FakeDB()
with container.override(Database, fake_db):
result = handler(name="John")
assert result == "John"
assert fake_db.last_query == "INSERT John"Overrides are safe against exceptions and support nesting:
def test_with_nested_overrides():
with container.override(Database, FakeDB()):
with container.override(Logger, FakeLogger()):
result = handler(name="John")Or bypass injection entirely by passing dependencies directly:
def test_handler_directly():
result = handler(db=FakeDB(), name="John")The Container is fully thread-safe. All operations (register, resolve, override) are protected by an internal lock, so it's safe to use a single container across multiple threads in any server model (Gunicorn, Uvicorn, threaded servers).
MIT. See LICENSE for details.