[Feature request]
Support for different request-scoped providers and instances
Sometimes there is a need to register a different provider and/or instance for each request context. An example for this would be a request-level parameter, config or state that needs to be shared with other request-scoped services easily.
Currently this is not possible, because
a. only one provider can be registered globally for the request-scope and/or
b. registration of instances is not supported.
One possible workaround at the moment is to manually register data on the current protected request context of the container.
Assumedly, the target solution should be something very similar, but better exposed by the framework, for example:
a. introducing the concept of registering parameters, configs or instances for the different scopes; which automatically means that for request-scope a new, different instance can be bound every time on a request context
b. introducing support for registering a different provider for each request context instead of just globally for the request-scope.
Please see two runnable example snippets for:
The problem statement:
import time
from threading import Thread
from typing import Annotated
import anydi
import requests
import uvicorn
from anydi import Container
from anydi.ext.fastapi import Inject
from anydi.ext.starlette.middleware import RequestScopedMiddleware
from fastapi import FastAPI, Path
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import Response
from starlette.types import ASGIApp
def add_item(self, item: str):
self.items.append(item)
def order_items(self):
return f"Order for shopping cart {self.shopping_cart_id} has been placed. Items: {self.items}"
class ShoppingCartContextMiddleware(BaseHTTPMiddleware):
def __init__(self, app: ASGIApp, container: Container) -> None:
super().__init__(app)
self.container = container
async def dispatch(
self, request: Request, call_next: RequestResponseEndpoint
) -> Response:
path = request.url.path
parts = path.split('/')
shopping_cart_id = parts[1] # a more robust parsing is needed in real-life
# [FEATURE REQUEST]
# THIS IS NOT POSSIBLE AT THE MOMENT
# (to register a new provider, or instance, only in the scope of the current request-context)
self.container.register(Annotated[str, "shopping_cart_id"], lambda: shopping_cart_id, scope="request")
return await call_next(request)
if name == "main":
thread = Thread(target=run_server)
thread.daemon = True
thread.start()
# Give the server a moment to start
print("Waiting for server to start...")
time.sleep(3)
print("Sending test request...")
response = requests.get("http://127.0.0.1:8000/my-shopping-cart-1/order-test")
print(response.text)
response = requests.get("http://127.0.0.1:8000/my-shopping-cart-1/order-test")
print(response.text)
# second invocation raises: LookupError: The provider interface `Annotated[str, "shopping_cart_id"]]` already registered.
# because the provider registration happens globally and not just inside the current scope (request context)
# see ShoppingCartContextMiddleware for the root cause
2. The **workaround**:
```python
import time
from threading import Thread
from typing import Annotated
import anydi
import requests
import uvicorn
from anydi import Container
from anydi.ext.fastapi import Inject
from anydi.ext.starlette.middleware import RequestScopedMiddleware
from fastapi import FastAPI, Path
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import Response
from starlette.types import ASGIApp
class SomeOtherService:
pass
class ShoppingCartService:
def __init__(self,
shopping_cart_id: Annotated[str, "shopping_cart_id"],
some_other_service: SomeOtherService):
self.shopping_cart_id = shopping_cart_id
self.some_other_service = some_other_service
self.items = []
def add_item(self, item: str):
self.items.append(item)
def order_items(self):
return f"Order for shopping cart {self.shopping_cart_id} has been placed. Items: {self.items}"
class ShoppingCartContextMiddleware(BaseHTTPMiddleware):
def __init__(self, app: ASGIApp, container: Container) -> None:
super().__init__(app)
self.container = container
async def dispatch(
self, request: Request, call_next: RequestResponseEndpoint
) -> Response:
path = request.url.path
parts = path.split('/')
shopping_cart_id = parts[1] # a more robust parsing is needed in real-life
# ------ WORKAROUND PART 1 ------
# noinspection PyProtectedMember
rc = self.container._get_request_context()
rc.shopping_cart_id = shopping_cart_id
return await call_next(request)
container = Container()
app = FastAPI(middleware=[
Middleware(RequestScopedMiddleware, container=container),
Middleware(ShoppingCartContextMiddleware, container=container),
])
# ------ WORKAROUND PART 2 ------
@container.provider(scope="request")
def provide_shopping_cart_id() -> Annotated[str, "shopping_cart_id"]:
"""
Provides the `shopping_cart_id` in the current request context, if available, otherwise raises error.
See `ShoppingCartContextMiddleware` as a prerequisite.
"""
shopping_cart_id = None
# noinspection PyProtectedMember
rc = container._get_request_context()
if hasattr(rc, "shopping_cart_id"):
shopping_cart_id = rc.shopping_cart_id
if not shopping_cart_id:
raise LookupError("The shopping_cart_id parameter has not been set up correctly on the request context!")
return shopping_cart_id
@app.get("/{shopping_cart_id}/order-test")
async def order_test(shopping_cart_id: str = Path(),
# instead of manually passing `shopping_cart_id` everywhere, injection would be preferred
cart_service: ShoppingCartService = Inject()) -> str:
cart_service.add_item("test item 1")
cart_service.add_item("test item 2")
result = cart_service.order_items()
return result
anydi.ext.fastapi.install(app, container)
def run_server():
uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__":
thread = Thread(target=run_server)
thread.daemon = True
thread.start()
# Give the server a moment to start
print("Waiting for server to start...")
time.sleep(3)
print("Sending test request...")
response = requests.get("http://127.0.0.1:8000/my-shopping-cart-1/order-test")
print(response.text)
response = requests.get("http://127.0.0.1:8000/my-shopping-cart-1/order-test")
print(response.text)
# both invocations work (even if they were to be invoked in parallel)
[Feature request] Support for different request-scoped providers and instances
Sometimes there is a need to register a different provider and/or instance for each request context. An example for this would be a request-level parameter, config or state that needs to be shared with other request-scoped services easily.
Currently this is not possible, because a. only one provider can be registered globally for the request-scope and/or b. registration of instances is not supported.
One possible workaround at the moment is to manually register data on the current protected request context of the container.
Assumedly, the target solution should be something very similar, but better exposed by the framework, for example: a. introducing the concept of registering parameters, configs or instances for the different scopes; which automatically means that for request-scope a new, different instance can be bound every time on a request context b. introducing support for registering a different provider for each request context instead of just globally for the request-scope.
Please see two runnable example snippets for:
import anydi import requests import uvicorn from anydi import Container from anydi.ext.fastapi import Inject from anydi.ext.starlette.middleware import RequestScopedMiddleware from fastapi import FastAPI, Path from starlette.middleware import Middleware from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint from starlette.requests import Request from starlette.responses import Response from starlette.types import ASGIApp
class SomeOtherService: pass
class ShoppingCartService: def init(self, shopping_cart_id: Annotated[str, "shopping_cart_id"], some_other_service: SomeOtherService): self.shopping_cart_id = shopping_cart_id self.some_other_service = some_other_service self.items = []
class ShoppingCartContextMiddleware(BaseHTTPMiddleware):
container = Container()
app = FastAPI(middleware=[ Middleware(RequestScopedMiddleware, container=container), Middleware(ShoppingCartContextMiddleware, container=container), ])
@app.get("/{shopping_cart_id}/order-test") async def order_test(shopping_cart_id: str = Path(),
instead of manually passing
shopping_cart_ideverywhere, injection would be preferredanydi.ext.fastapi.install(app, container)
def run_server(): uvicorn.run(app, host="0.0.0.0", port=8000)
if name == "main": thread = Thread(target=run_server) thread.daemon = True thread.start()