Support context managers

[martinmkhitaryan]

Add opt-in context manager management via manage_context flag

Summary

Adds opt-in support for rodi to enter and exit services that implement the
context manager protocol (__enter__/__exit__ and/or __aenter__/__aexit__).
Behavior is controlled by a new manage_context: bool = False parameter on the
registration methods. The default is False, so all existing behavior is
preserved exactly.

Motivation

The rodi documentation page on
context managers
states the core problem directly:

There is no way to unambiguously know the intentions of the developer:
should a context be entered automatically and disposed automatically?

Today rodi resolves the ambiguity by doing nothing: classes that implement the
context manager protocol are instantiated but never entered or exited. Users
have to wrap every call site themselves with with / async with, or wire
per-request middleware that mirrors what a DI container should already be
doing.

This PR resolves the ambiguity by giving the developer an explicit declaration
of intent. manage_context=True on registration tells rodi "yes, please enter
this on resolve and exit it when the owning scope ends." The default
(manage_context=False) preserves today's behavior verbatim, so no existing
app is affected.

A concrete case where this matters is SQLAlchemy's async session:

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

session_factory = async_sessionmaker(engine, expire_on_commit=False)

# Without this PR: every handler has to do `async with session_factory() as s:`
# itself, or the framework has to wire its own per-request middleware.

# With manage_context=True: rodi enters and exits the session per request scope.
container.add_scoped_by_factory(
    session_factory, AsyncSession, manage_context=True
)

@app.router.post("/users")
async def create_user(session: AsyncSession, payload: UserIn):
    session.add(User(**payload.dict()))
    # On scope close, rodi awaits __aexit__ in LIFO order across all managed
    # services in the request graph.

The same pattern applies to httpx clients, file handles, locks, tracing
spans, and anything else that wants entry on borrow / exit on return.

API

# Sync — managed by ActivationScope's ExitStack
container.add_scoped(MyResource, manage_context=True)
container.add_transient(MyResource, manage_context=True)
container.add_scoped_by_factory(my_factory, manage_context=True)
container.register(MyResource, manage_context=True)

with provider.create_scope() as scope:
    res = scope.get(MyResource)   # __enter__ called here
# __exit__ called here, in LIFO order across all managed instances


# Async — managed by AsyncActivationScope's AsyncExitStack
container.add_scoped(MyAsyncResource, manage_context=True)

async with provider.create_async_scope() as scope:
    res = await scope.aget(MyAsyncResource)   # __aenter__ awaited here
# __aexit__ awaited here, in LIFO order


# Mixed sync + async dependency graphs work in async scope
async with provider.create_async_scope() as scope:
    obj = await scope.aget(SomethingThatDependsOnBothKinds)


# Framework-facing async resolve via Container.aresolve
async with provider.create_async_scope() as scope:
    obj = await container.aresolve(MyAsyncResource, scope)
# `aresolve` requires an AsyncActivationScope and raises TypeError otherwise.
# It is the async counterpart to Container.resolve, declared by the new
# AsyncContainerProtocol so frameworks can opt into async resolution
# structurally without forcing every other DI implementation to follow.

Behavior matrix

Scope	Sync-only CM	Async-only CM	Class with both protocols
provider.create_scope()	entered & exited	TypeError	entered as sync
provider.create_async_scope()	entered as sync	entered as async	entered as async

• Scoped + managed: entered once per scope on first resolve, exited on
  scope end.
• Transient + managed: entered on every resolve, all instances exited on
  scope end (LIFO).
• Singleton + managed: rejected at registration with
  InvalidContextManagerRegistration (rationale below).

Why singleton is not supported (yet)

Singletons live for the lifetime of the Services provider, but rodi has no
explicit teardown hook for the provider/container today. Supporting
manage_context=True for singletons would require introducing one (e.g.
Services.dispose() or making Container itself a context manager), which is
its own design discussion (atexit semantics, shutdown ordering, error
handling).

To keep this PR focused and reviewable, singleton support is deferred to a
follow-up issue. The current behavior is to raise
InvalidContextManagerRegistration at registration with a clear message
pointing the user to SCOPED/TRANSIENT or to manual management.

Implementation notes

• New ManagedScopedProvider and ManagedTransientProvider wrap any existing
  inner provider (TypeProvider, ArgsTypeProvider, FactoryTypeProvider).
  This keeps the existing 6 provider classes completely untouched, so users
  who don't opt in pay zero overhead on the hot resolve path.
• ActivationScope gained a lazy _exit_stack: ExitStack | None. Drained
  before dispose() in __exit__, in LIFO order. Standard
  swap-and-call idiom for single-shot consumption.
• AsyncActivationScope is a new subclass that adds a lazy
  _async_exit_stack: AsyncExitStack | None. Resolution stays synchronous —
  aget flips an _async_mode flag for the duration of Services.get,
  causing _enter_context to defer instances into a _pending_aenter
  list. Once sync resolution returns, aget drains the list using
  enter_async_context (for async CMs) or enter_context (for sync CMs)
  on the same AsyncExitStack. Sync get calls on the same async scope
  also route into that unified stack via an overridden
  _register_sync_context, so cross-protocol exit order across mixed
  get + aget calls is strictly LIFO regardless of which entry point
  added the instance.
• AsyncContainerProtocol is a separate Protocol declaring aresolve.
  Container implements both ContainerProtocol and
  AsyncContainerProtocol. The existing ContainerProtocol is unchanged,
  so sync-only third-party DI containers remain compatible. Frameworks that
  want async resolution opt into it structurally by typing against
  AsyncContainerProtocol (or against Container directly).
  Container.aresolve is a thin wrapper around scope.aget that validates
  the scope type up front, so misuse fails with a clear TypeError instead
  of silently leaking managed CMs into an un-awaited throwaway scope.
• TrackingActivationScope.__exit__ now calls super().__exit__(...) instead
  of self.dispose() directly, so the new exit-stack draining runs for users
  who opt into nested-scope tracking. Pre-existing tests confirm dispose
  semantics are unchanged.
• Validation of "is this a context manager?" happens at resolution time inside
  _enter_context rather than at registration. Class-level checks at
  registration are an easy follow-up; deferred because the factory path can't
  reliably introspect the type the factory will return.

Backward compatibility

manage_context defaults to False everywhere. All existing call sites,
fixtures, and tests work without changes. No public symbol was renamed or
removed. New public symbols added:

• AsyncActivationScope
• AsyncContainerProtocol
• Container.aresolve
• InvalidContextManagerRegistration
• ManagedScopedProvider, ManagedTransientProvider (intended internal but
  exposed for parity with the other provider classes already in the public
  module surface)
• Services.create_async_scope, Services.aget

Follow-up (not in this PR)

A separate issue/PR will discuss singleton + manage_context=True. The shape
of that work — provider-level dispose, atexit semantics, container-as-context-
manager — is large enough that bundling it here would make this PR much harder
to review.

[RobertoPrevato]

Hi!
Thank you for your contribution. I reviewed the PR. I like the intent and the opt-in approach. I am
concerned about two things:

1. feature creep: this will set a precedence for refactoring the whole rodi to support async activation of services and this was intentionally left out of the scope because instantiating objects should be fast and not slowed down by async machinery, IMHO)
2. The fact that in your PR description there is an example of footgun: using this for httpx. This would probably cause problems with TCP IP connection pooling (unless the user explicitly handles the connection pool in some other way).

I also used AI, to review (Claude Sonnet 4.6). Giving my input above, it outputted:

Summary

The PR is well-written and the opt-in design (manage_context=False default) is commendable. The sync path is clean and contained. However, there are significant concerns before merging.

---

🔴 Scope / Feature Creep

The sync half (ExitStack in ActivationScope) is modest: ~80 LoC and self-contained. The async half is a different story and is what opens the door you intentionally kept
closed:

• AsyncActivationScope with _async_exit_stack, _async_mode, _pending_aenter — a non-trivial async state machine
• AsyncContainerProtocol — a new public protocol interface for frameworks
• Container.aresolve — a new public async API method
• Services.aget / Services.create_async_scope — two new public methods on Services
• The PR description itself already names the follow-up: singleton lifecycle hooks, Services.dispose(), container-as-context-manager

The PR author is right that those are separate — but this PR is the foot in the door. Once AsyncActivationScope and AsyncContainerProtocol are public surface, the
pressure to fill in the rest (singleton disposal, Container.aenter/aexit, etc.) becomes much stronger. This is the precedent risk you identified.

---

🔴 The httpx Example Is a Footgun

The PR description lists httpx as a concrete use case:

"The same pattern applies to httpx clients…"

This is incorrect and will cause real performance harm to users who follow it.

Why it's wrong:
httpx.AsyncClient manages a connection pool internally. The whole reason to use it (vs. making one-shot requests) is to reuse TCP connections across multiple HTTP
requests. The correct lifetime is singleton (one client for the lifetime of the application, closed at shutdown).

• As add_scoped(manage_context=True): a new AsyncClient is created and entered (pool opens) on every request scope, then exited (pool closes) when the scope ends. You
  get zero connection reuse — worse than just using httpx.get().
• As add_transient(manage_context=True): even worse, a new client per injection site.
• As a singleton: explicitly rejected by this PR with InvalidContextManagerRegistration.

So for the httpx case, manage_context=True is not just unhelpful — it actively destroys connection pooling. The httpx example should be removed from the PR description
entirely, or replaced with a clear warning.

The PR description should only use the SQLAlchemy async session as the motivating example (where a new session per request is the correct pattern).

---

🟡 Bug: Services.aget Creates an Orphaned Scope

   async def aget(self, desired_type, scope=None, *, default=...):
       if scope is None:
           scope = self.create_async_scope()  # created but never __aexit__'d
       return await scope.aget(desired_type, default=default)

If scope is None, a new AsyncActivationScope is created but its aexit is never awaited. Any managed context managers entered during resolution will never be exited —
a resource leak. Either require the scope explicitly (no default None) or document that this path skips lifecycle management.

---

🟡 Late Validation (Registration vs. Resolution)

The PR validates at resolution time, not registration time:

   # Validation of "is this a context manager?" happens at resolution time inside
   # _enter_context rather than at registration.

This means if you register NotACM with manage_context=True, the error only surfaces the first time the service is resolved (possibly deep in a request handler), not at
app startup. For a library that already validates circular dependencies and abstract types at build time, this is inconsistent. The author acknowledges this as a known
gap.

---

🟡 slots Inheritance in AsyncActivationScope

AsyncActivationScope defines __slots__ = ("_async_exit_stack", "_async_mode", "_pending_aenter") but ActivationScope.__slots__ includes ("scoped_services", "provider",
"_exit_stack"). Since ActivationScope is not object, Python will correctly use both slot descriptors — this is fine — but it also means instances still carry a
_exit_stack from the parent that is never used in the async path (the async path redirects sync CMs into the AsyncExitStack via _register_sync_context). It's not a bug,
but the unused _exit_stack slot is confusing.

---

✅ What's Good

• Default manage_context=False everywhere — zero change to existing behavior.
• LIFO exit ordering is correct and well-tested.
• Singleton rejection with a clear exception message is the right call.
• Test coverage is extensive (542-line test file, covers LIFO, mixed sync/async, exception propagation).
• ManagedScopedProvider / ManagedTransientProvider wrap without touching existing providers — zero hot-path overhead for non-managed services.

---

And also, I argued this is a lot of work and maintenance overhead to achieve something as simple as, for example:

# new - automatic entering and exiting of session
async def get_topic_by_id(self, site_id: int, topic_id: int) -> Topic | None:                                                                                                                                  
        result = await self.session.execute(                                                                                                                          
            select(TopicEntity).filter(                                                                                                                               
                TopicEntity.id == topic_id,                                                                                                                           
                TopicEntity.site_id == site_id,                                                                                                                       
            )                                                                                                                                                         
        )                                                                                                                                                             
        record = result.scalar_one_or_none()                                                                                                                          
        return topic_entity_to_domain_type(record) if record else None                                                                                                

instead of (currently requires entering the async session):

    async def get_topic_by_id(self, site_id: int, topic_id: int) -> Topic | None:                                                                                     
        async with self.session as session:                                                                                                                           
            result = await session.execute(                                                                                                                           
                select(TopicEntity).filter(                                                                                                                           
                    TopicEntity.id == topic_id,                                                                                                                       
                    TopicEntity.site_id == site_id,                                                                                                                   
                )                                                                                                                                                     
            )                                                                                                                                                         
            record = result.scalar_one_or_none()                                                                                                                      
            return topic_entity_to_domain_type(record) if record else None

To which the AI commented:

Yes, exactly. The entire PR — ~900 lines added, 5 new public symbols, a new async state machine — is motivated by saving the async with wrapper at the call site. And that wrapper is actually doing something meaningful: it makes the session lifecycle explicit and visible in the code.

The tradeoff is poor:

• Complexity added to rodi: significant (async state machine, new protocols, new provider wrappers)
• Complexity removed from user code: one async with block per repository method — or more realistically, one async with at the handler/unit-of-work boundary, not per method

The second version in your example (current) is arguably better code: it's obvious where the session begins and ends. The first version hides that the session has a lifecycle at
all — it looks like self.session is just a plain object, with the enter/exit happening invisibly somewhere in the DI machinery.

The right answer for that use-case is almost certainly a factory dependency — inject async_sessionmaker (or a Callable[[], AsyncSession]) and let the handler/unit-of-work own the async with. That's zero new rodi infrastructure, zero magic, and the lifecycle stays in plain sight.

---

I thank you for the time you put into this and interesting idea, but I prefer not merging this and keep handling of async context explicit in user's code for the reasons pointed out above by Claude Sonnet 4.6. This way, all objects activated by rodi look the same from DI perspective.

[martinmkhitaryan]

Hi! Thank you for the detailed review and for taking the time to read through it so thoroughly.

You make excellent points. Regarding the httpx example: that is a very fair point about connection pooling, and you are right that manage_context=True would actively harm performance there. That was a bad example on my part.

Regarding the concerns about feature creep and singletons: I just wanted to clarify that I did actually consider them (as mentioned in the PR description). I deliberately deferred that work to keep this initial PR smaller and easier to review. My intention was to create a separate follow-up PR for singleton support if this initial implementation was merged. However, I completely understand your reasoning—even if done in a separate PR, adding this feature sets a precedent and forces the library to eventually adopt complex DI lifecycle management, which goes against the goal of keeping rodi simple and explicit.

I also wanted to quickly touch on three of your technical notes:

1. Late Validation: You are completely right that for standard Container usage, early validation (at registration time) is much better so it fails fast. The reason I used late validation (at resolution time inside _enter_context) is because ActivationScope is decoupled and can be used standalone without a container. In that standalone flow, there is no register() step, so _enter_context is the only place it can be checked. A robust implementation would probably check at registration and resolution (perhaps passing a cached validation flag from the container down to the scope to avoid double-checking), but I deferred registration checks for simplicity in this first pass.

2. The aget Orphaned Scope Bug: You spotted that scope=None creates a scope that is never exited. It turns out the exact same bug exists in the synchronous Services.get method as well under this PR! If scope=None, it calls self.create_scope() but never calls __exit__ on it, meaning any managed context managers would leak there too. It just highlights exactly what you said: trying to do this implicitly without the user providing an explicit with scope is a recipe for leaks.

3. __slots__ Inheritance: You are completely right that the _exit_stack inherited from the parent is unused since _async_exit_stack handles both sync and async context managers. I chose to inherit from ActivationScope simply because the async scope conceptually extends the base scope to support both types of context managers. I agree though, carrying over an unused slot is a bit confusing, and avoiding inheritance might have been cleaner.

Thanks again for the feedback.