
Security News
npm ‘is’ Package Hijacked in Expanding Supply Chain Attack
The ongoing npm phishing campaign escalates as attackers hijack the popular 'is' package, embedding malware in multiple versions.
ididi is 100% test covered and strictly typed.
📚 Docs: : https://raceychan.github.io/ididi
ididi requires python >= 3.9
pip install ididi
To view viusal dependency graph, install graphviz
pip install ididi[graphviz]
typing
module.ididi has strong support to typing
module, includes:
...and more.
Check out tests/features/test_typing_support.py
for examples.
from typing import AsyncGenerator
from ididi import use, entry
async def conn_factory(engine: AsyncEngine) -> AsyncGenerator[AsyncConnection, None]:
async with engine.begin() as conn:
yield conn
class UnitOfWork:
def __init__(self, conn: AsyncConnection=use(conn_factory)):
self._conn = conn
@entry
async def main(command: CreateUser, uow: UnitOfWork):
await uow.execute(build_query(command))
# note uow is automatically injected here
await main(CreateUser(name='user'))
To resolve AsyncConnection
outside of entry function
from ididi import Graph
dg = Graph()
async with dg.ascope():
conn = await scope.resolve(conn_factory)
ididi provides two mark
s, use
and Ignore
, they are convenient shortcut for Graph.node
.
Technically, they are just metadata carried by typing.Annotated
, and should work fine with other Annotated metadata.
use
You can use Graph.node
to register a dependent with its factory,
here we register dependent Database
with its factory db_factory
.
This means whenver we call dg.resolve(Database)
, db_factory
will be call.
def db_factory() -> Database:
return Database()
dg = Graph()
dg.node(db_factory)
Alternatively, you can annotate it inside __init__
, this allow you to instantiate
Graph
in a lazy manner.
from ididi import use
class Repository:
def __init__(self, db: Annotated[Database, use(db_factory)]):
...
Ignore
ididi takes a "resolve by default" approach, for dependencies you would like ididi to ignore, you can config ididi to ignore them.
from datetime import datetime
from pathlib import Path
dg = Graph(ignore=(datetime, Path))
dg = Graph()
class Clock:
def __init__(self, dt: datetime): ...
dg.node(Clock, ignore=datetime)
Alternatively, you can mark a dependency using ididi.Ignore
,
from ididi import Ignore
class Clock:
def __init__(self, dt: Ignore[datetime]): ...
Declear a function as a dependency by using Ignore
to annotate its return type.
@dataclass
class User:
name: str
role: str
def get_user(config: Config) -> Ignore[User]:
assert isinstance(config, Config)
return User("user", "admin")
def validate_admin(
user: Annotated[User, use(get_user)], service: UserService
) -> Ignore[str]:
assert user.role == "admin"
assert isinstance(service, UserService)
return "ok"
class Route:
def __init__(self, validte_permission: Annotated[str, use(validate_admin)]):
assert validte_permission == "ok"
assert dg.resolve(validate_admin) == "ok"
assert isinstance(dg.resolve(Route), Route)
Since get_user
returns Ignore[User]
instead of User
, it won't be used as factory to resolve User
.
from ididi import use
from typing import NewType
from datetime import datetime, timezone
UserID = NewType("UserID", str)
def utc_factory() -> datetime:
return datetime.now(timezone.utc)
def user_id_factory() -> UserID:
return UserID(str(uuid4()))
class User:
def __init__(self, user_id: UserID, created_at: Annotated[datetime, use(utc_factory)]):
self.user_id = user_id
self.created_at = created_at
user = ididi.resolve(User)
assert user.created_at.tzinfo == timezone.utc
[!TIP]
Graph.node
accepts a wide arrange of types, such as dependent class, sync/async facotry, sync/async resource factory, with typing support.
Scope
is a temporary view of the graph specialized for handling resources.
In a nutshell:
contextlib.AbstractContextManager
or contextlib.AbstractAsyncContextManager
are also considered to be resources and can/should be resolved within scope.[!TIP] If you have two call stack of
a1 -> b1
anda2 -> b2
, Herea1
anda2
are two calls to the same functiona
, then, inb1
, you can only access scope created by thea1
, nota2
.
This is particularly useful when you try to separate resources by route, endpoint, request, etc.
@dg.node
def get_resource() -> ty.Generator[Resource, None, None]:
res = Resource()
with res:
yield res
@dg.node
async def get_asyncresource() -> ty.Generator[AsyncResource, None, None]:
res = AsyncResource()
async with res:
yield res
with dg.scope() as scope:
resource = scope.resolve(Resource)
# For async generator
async with dg.ascope() as scope:
resource = await scope.resolve(AsyncResource)
[!TIP]
dg.node
will leave your class/factory untouched, i.e., you can use it as a function. e.g.dg.node(get_resource, reuse=False)
You can use dg.use_scope to retrive most recent scope, context-wise, this allows your to have access the scope without passing it around, e.g.
async def service_factory():
async with dg.ascope() as scope:
service = scope.resolve(Service)
yield service
@app.get("users")
async def get_user(service: Service = Depends(service_factory))
await service.create_user(...)
Then somewhere deep in your service.create_user call stack
async def create_and_publish():
uow = dg.use_scope().resolve(UnitOfWork)
async with uow.trans():
user_repo.add_user(user)
event_store.add(user_created_event)
Here dg.use_scope()
would return the same scope you created in your service_factory
.
You can create infinite level of scopes by assigning hashable name to scopes
# at the top most entry of a request
async with dg.ascope(request_id) as scope:
...
now scope with name request_id
is accessible everywhere within the request context
request_scope = dg.use_scope(request_id)
[!NOTE] Two or more scopes with the same name would follow most recent rule.
async with dg.ascope(app_name) as app_scope:
async with dg.ascope(router_name) as router_scope:
async with dg.ascope(endpoint_name) as endpoint_scope:
async with dg.ascope(user_id) as user_scope:
async with dg.ascope(request_id) as request_scope:
...
For any functions called within the request_scope, you can get the most recent scope with dg.use_scope()
,
or its parent scopes, i.e. dg.use_scope(app_name)
to get app_scope.
You can control how ididi resolve a dependency during testing, by register the test double of the dependency using
Graph.override
entry.replace
Example: For the following dependent
class UserRepository:
def __init__(self, db: DataBase):
self.db=db
dg = Graph()
assert isinstance(dg.resolve(UserRepository).db, DataBase)
in you test file,
class FakeDB(DataBase): ...
def db_factory() -> DataBase:
return FakeDB()
def test_resolve():
dg = Graph()
assert isinstance(dg.resolve(db_factory).db, DataBase)
dg.override(DataBase, db_factory)
assert isinstance(dg.resolve(UserRepository).db, FakeDB)
Use Graph.override
to replace DataBase
with its test double.
async def test_entry_replace():
@ididi.entry
async def create_user(
user_name: str, user_email: str, service: UserService
) -> UserService:
return service
class FakeUserService(UserService): ...
create_user.replace(UserService, FakeUserService)
res = await create_user("user", "user@email.com")
assert isinstance(res, FakeUserService)
Use entryfunc.replace
to replace a dependency with its test double.
Graph.override
vs entry.replace
Graph.override
applies to the whole graph, entry.replace
applies to only the entry function.
This cheatsheet is designed to give you a quick glance at some of the basic usages of ididi.
from ididi import Graph, Resolver, Ignore
class Base:
def __init__(self, source: str = "class"):
self.source = source
class CTX(Base):
def __init__(self, source: str="class"):
super().__init__(source)
self.status = "init"
async def __aenter__(self):
self.status = "started"
return self
async def __aexit__(self, *args):
self.status = "closed"
class Engine(Base): ...
class Connection(CTX):
def __init__(self, engine: Engine):
super().__init__()
self.engine = engine
def get_engine() -> Engine:
return Engine("factory")
async def get_conn(engine: Engine) -> Connection:
async with Connection(engine) as conn:
yield conn
async def func_dep(engine: Engine, conn: Connection) -> Ignore[int]:
return 69
async def test_ididi_cheatsheet():
dg = Graph()
assert isinstance(dg, Resolver)
engine = dg.resolve(Engine) # resolve a class
assert isinstance(engine, Engine) and engine.source == "class"
faq_engine = dg.resolve(get_engine) # resolve a factory function of a class
assert isinstance(faq_engine, Engine) and faq_engine.source == "factory"
side_effect: list[str] = []
assert not side_effect
async with dg.ascope() as ascope:
ascope.register_exit_callback(random_callback)
# register a callback to be called when scope is exited
assert isinstance(ascope, Resolver)
# NOTE: scopes are also resolvers, thus can have sub-scope
conn = await ascope.aresolve(get_conn)
# generator function will be transformed into context manager and can only be resolved within scope.
assert isinstance(conn, Connection)
assert conn.status == "started"
# context manager is entered when scoped is entered.
res = await ascope.aresolve(func_dep)
assert res == 69
# function dependencies are also supported
assert conn.status == "closed"
# context manager is exited when scope is exited.
assert side_effect[0] == "callbacked"
# registered callback will aslo be called.
For more detailed information, check out Documentation
Tutorial
Usage of factory
Visualize the dependency graph
Circular Dependency Detection
Error context
FAQs
A dependency injection library for Python, Optimized for serverless applications
We found that ididi 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
The ongoing npm phishing campaign escalates as attackers hijack the popular 'is' package, embedding malware in multiple versions.
Security News
A critical flaw in the popular npm form-data package could allow HTTP parameter pollution, affecting millions of projects until patched versions are adopted.
Security News
Bun 1.2.19 introduces isolated installs for smoother monorepo workflows, along with performance boosts, new tooling, and key compatibility fixes.