Feature/sqlalchemy request scoped session (#53)

* feat: add scoped session

docs/en/extensions/data/sqlalchemy.md

@@ -18,33 +18,37 @@ configuration file:
 extensions:
   - selva.ext.data.sqlalchemy # (1)
 
+middleware:
+  - selva.ext.data.sqlalchemy.middleware.scoped_session # (2)
+
 data:
   sqlalchemy:
     connections:
-      default: # (2)
+      default: # (3)
         url: "sqlite+aiosqlite:///var/db.sqlite3"
 
-      postgres: # (3)
+      postgres: # (4)
         url: "postgresql+asyncpg://user:pass@localhost/dbname"
 
-      mysql: # (4)
+      mysql: # (5)
         url: "mysql+aiomysql://user:pass@localhost/dbname"
 
-      oracle: # (5)
+      oracle: # (6)
         url: "oracle+oracledb_async://user:pass@localhost/DBNAME"
         # or "oracle+oracledb_async://user:pass@localhost/?service_name=DBNAME"
 ```
 
 1.  Activate the sqlalchemy extension
-2.  "default" connection will be registered without a name
-3.  Connection registered with name "postgres"
-4.  Connection registered with name "mysql"
-5.  Connection registered with name "oracle"
+2.  Activate the scoped session middleware
+3.  "default" connection will be registered without a name
+4.  Connection registered with name "postgres"
+5.  Connection registered with name "mysql"
+6.  Connection registered with name "oracle"
 
 Once we define the connections, we can inject `AsyncEngine` into our services.
 For each connection, an instance of `AsyncEngine` will be registered, the `default`
-connection will be registered wihout a name, and the other will be registered with
-their respective names:
+connection will be registered without a name, and the other will be registered with
+their respective names.
 
 ```python
 from typing import Annotated
@@ -63,6 +67,33 @@ class MyService:
     engine_oracle: Annotated[AsyncEngine, Inject(name="oracle")]
 ```
 
+## Scoped Session
+
+If the `selva.ext.data.sqlalchemy.middleware.scoped_session` middleware is enabled,
+the `selva.ext.data.sqlalchemy.ScopedSession` service will be registered. It provides
+access to an instance of `AsyncSession` that is available for the duration of the
+request:
+
+```python
+from typing import Annotated
+from sqlalchemy import text
+from selva.di import service, Inject
+from selva.ext.data.sqlalchemy import ScopedSession
+
+
+@service
+class MyService:
+    session: Annotated[ScopedSession, Inject]
+
+    async def method(self) -> int:
+        return await self.session.scalar(text("select 1"))
+```
+
+The `ScopedSession` service is a proxy to an instance of `AsyncSession` that is created
+by the `selva.ext.data.sqlalchemy.middleware.scoped_session` middleware.
+
+## Configuration
+
 Database connections can also be defined with username and password separated from
 the url, or even with individual components:
 

docs/pt/extensions/data/sqlalchemy.md

@@ -18,28 +18,32 @@ Com os drivers instalados, nós podemos definir as conexões no arquivo de confi
 extensions:
   - selva.ext.data.sqlalchemy # (1)
 
+middleware:
+  - selva.ext.data.sqlalchemy.middleware.scoped_session # (2)
+
 data:
   sqlalchemy:
     connections:
-      default: # (2)
+      default: # (3)
         url: "sqlite+aiosqlite:///var/db.sqlite3"
 
-      postgres: # (3)
+      postgres: # (4)
         url: "postgresql+asyncpg://user:pass@localhost/dbname"
 
-      mysql: # (4)
+      mysql: # (5)
         url: "mysql+aiomysql://user:pass@localhost/dbname"
 
-      oracle: # (5)
+      oracle: # (6)
         url: "oracle+oracledb_async://user:pass@localhost/DBNAME"
         # ou "oracle+oracledb_async://user:pass@localhost/?service_name=DBNAME"
 ```
 
 1.  Ativar a extensão sqlalchemy
-2.  A conexão "default" será registrada sem um nome
-3.  Conexão registrada com nome "postgres"
-4.  Conexão registrada com nome "mysql"
-5.  Conexão registrada com nome "oracle"
+2.  Ativar o middleware de scoped session
+3.  A conexão "default" será registrada sem um nome
+4.  Conexão registrada com nome "postgres"
+5.  Conexão registrada com nome "mysql"
+6.  Conexão registrada com nome "oracle"
 
 Uma vez definidas as conexões, nós podemos injetar `AsyncEngine` nos nossos serviços.
 Para cada conexão, uma instância de `AsyncEngine` será registrada, a conexão `default`
@@ -63,6 +67,32 @@ class MyService:
     engine_oracle: Annotated[AsyncEngine, Inject(name="oracle")]
 ```
 
+## Scoped Session
+
+Se o middleware `selva.ext.data.sqlalchemy.middleware.scoped_session` for ativado,
+o serviço `selva.ext.data.sqlalchemy.ScopedSession` será registrado. Ele provê acesso
+a uma instância de `AsyncSession` que fica disponível pela duração da requisição.
+
+```python
+from typing import Annotated
+from sqlalchemy import text
+from selva.di import service, Inject
+from selva.ext.data.sqlalchemy import ScopedSession
+
+
+@service
+class MyService:
+    session: Annotated[ScopedSession, Inject]
+
+    async def method(self) -> int:
+        return await self.session.scalar(text("select 1"))
+```
+
+O serviço `ScopedSession` é um proxy para a instância de `AsyncSession` que é criado
+pelo middeware `selva.ext.data.sqlalchemy.middleware.scoped_session`.
+
+## Configuração
+
 Conexões de bancos de dados podem ser definidas com usuário e senha separados da
 url, ou até com componentes individuais:
 

examples/sqlalchemy/application/service.py

@@ -4,6 +4,7 @@
 from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker
 
 from selva.di import Inject, service
+from selva.ext.data.sqlalchemy.service import ScopedSession
 
 from .model import Base, MyModel, OtherBase, OtherModel
 
@@ -12,6 +13,7 @@
 class DefaultDBService:
     engine: A[AsyncEngine, Inject]
     sessionmaker: A[async_sessionmaker, Inject]
+    session: A[ScopedSession, Inject]
 
     async def initialize(self):
         async with self.engine.connect() as conn:
@@ -27,14 +29,14 @@ async def db_version(self) -> str:
             return await conn.scalar(text("SELECT sqlite_version()"))
 
     async def get_model(self) -> MyModel:
-        async with self.sessionmaker() as session:
-            return await session.scalar(select(MyModel).limit(1))
+        return await self.session.scalar(select(MyModel).limit(1))
 
 
 @service
 class OtherDBService:
     engine: A[AsyncEngine, Inject(name="other")]
     sessionmaker: A[async_sessionmaker, Inject]
+    session: A[ScopedSession, Inject]
 
     async def initialize(self):
         async with self.engine.connect() as conn:
@@ -51,5 +53,4 @@ async def db_version(self) -> str:
             return await conn.scalar(text("SELECT version()"))
 
     async def get_model(self) -> OtherModel:
-        async with self.sessionmaker() as session:
-            return await session.scalar(select(OtherModel).limit(1))
+        return await self.session.scalar(select(OtherModel).limit(1))

examples/sqlalchemy/configuration/settings.yaml

@@ -1,6 +1,9 @@
 extensions:
   - selva.ext.data.sqlalchemy
 
+middleware:
+  - selva.ext.data.sqlalchemy.middleware.scoped_session
+
 data:
   sqlalchemy:
     session:
@@ -27,6 +30,6 @@ data:
 
 logging:
   format: console
-  root: debug
+  root: info
   level:
     application: info

src/selva/ext/data/sqlalchemy/__init__.py

@@ -2,12 +2,17 @@
 
 from selva.configuration.settings import Settings
 from selva.di.container import Container
+from selva.ext.data.sqlalchemy.middleware import scoped_session
 from selva.ext.data.sqlalchemy.service import (
     engine_dict_service,
     make_engine_service,
     sessionmaker_service,
+    ScopedSession,
+    ScopedSessionImpl,
 )
 
+__all__ = ("ScopedSession",)
+
 
 def init_extension(container: Container, settings: Settings):
     if find_spec("sqlalchemy") is None:
@@ -20,3 +25,9 @@ def init_extension(container: Container, settings: Settings):
 
     container.register(engine_dict_service)
     container.register(sessionmaker_service)
+
+    middleware_name = (
+        f"{scoped_session.__module__}.{scoped_session.__name__}"
+    )
+    if middleware_name in settings.middleware:
+        container.register(ScopedSessionImpl)

src/selva/ext/data/sqlalchemy/middleware.py

@@ -0,0 +1,25 @@
+from contextvars import ContextVar
+
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from selva.configuration.settings import Settings
+from selva.di.container import Container
+
+SESSION = ContextVar("sqlalchemy session")
+
+
+async def scoped_session(app, _settings: Settings, container: Container):
+    sessionmaker = await container.get(async_sessionmaker)
+
+    async def middleware(scope, receive, send):
+        async with sessionmaker() as session:
+            token = SESSION.set(session)
+            try:
+                await app(scope, receive, send)
+            except:
+                await session.rollback()
+                raise
+            finally:
+                SESSION.reset(token)
+
+    return middleware

src/selva/ext/data/sqlalchemy/service.py

@@ -1,16 +1,36 @@
+from typing import TypeAlias
+
 import structlog
-from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
 
 from selva.configuration.settings import Settings
 from selva.di.container import Container
 from selva.di.decorator import service
+from selva.ext.data.sqlalchemy.middleware import SESSION
 from selva.ext.data.sqlalchemy.settings import (
     SqlAlchemyEngineSettings,
     SqlAlchemySettings,
 )
 
 logger = structlog.get_logger()
 
+# Type alias to provide autocomplete
+ScopedSession: TypeAlias = AsyncSession
+
+
+@service(provides=ScopedSession)
+class ScopedSessionImpl:
+    def __getattribute__(self, name: str):
+        try:
+            return getattr(SESSION.get(), name)
+        except LookupError:
+            raise RuntimeError("ScopedSession outside request")
+
 
 def make_engine_service(name: str):
     @service(name=name if name != "default" else None)

src/selva/logging.py

@@ -42,6 +42,11 @@ def setup(settings: Settings):
 
     processors.append(renderer)
 
+    extra_loggers = {
+        "sqlalchemy.engine.Engine": {"handlers": ["console"], "propagate": False},
+        "uvicorn": {"handlers": ["console"], "propagate": False},
+    }
+
     logging_config = {
         "version": 1,
         "disable_existing_loggers": False,
@@ -61,7 +66,7 @@ def setup(settings: Settings):
             "handlers": ["console"],
             "level": settings.logging.get("root", "WARN").upper(),
         },
-        "loggers": {
+        "loggers": extra_loggers | {
             module: {"level": level.upper()}
             for module, level in settings.logging.get("level", {}).items()
         },

src/selva/web/middleware/files.py

@@ -100,7 +100,7 @@ async def __call__(self, scope, receive, send):
                 raise
 
 
-async def static_files_middleware(app, settings: Settings, di: Container):
+async def static_files_middleware(app, settings: Settings, _di: Container):
     settings = settings.staticfiles
     path = settings.path.lstrip("/")
     root = Path(settings.root).resolve().absolute()
@@ -135,7 +135,7 @@ async def static_files_middleware(app, settings: Settings, di: Container):
     return StaticFilesMiddleware(app, path, root, filelist, mappings)
 
 
-def uploaded_files_middleware(app, settings: Settings, di: Container):
+def uploaded_files_middleware(app, settings: Settings, _di: Container):
     settings = settings.uploadedfiles
     path = settings.path.lstrip("/")
     root = Path(settings.root).resolve()

src/selva/web/middleware/request_id.py

@@ -7,7 +7,7 @@
 from selva.di.container import Container
 
 
-async def request_id_middleware(app, settings: Settings, di: Container):
+async def request_id_middleware(app, _settings: Settings, _di: Container):
     async def handler(scope, receive, send):
         request = Request(scope, receive, send)
 

tests/ext/data/sqlalchemy/application_scoped_session.py

@@ -0,0 +1,22 @@
+from typing import Annotated
+
+from asgikit.responses import respond_text
+from sqlalchemy import text
+
+from selva.di import Inject, service
+from selva.ext.data.sqlalchemy import ScopedSession
+from selva.web import get
+
+
+@service
+class MyService:
+    session: Annotated[ScopedSession, Inject]
+
+    async def select(self):
+        return await self.session.scalar(text("select 1"))
+
+
+@get
+async def index(request, my_service: Annotated[MyService, Inject]):
+    value = await my_service.select()
+    await respond_text(request.response, str(value))

tests/ext/data/sqlalchemy/application_scoped_session_startup.py

@@ -0,0 +1,9 @@
+from sqlalchemy import text
+
+from selva.ext.data.sqlalchemy import ScopedSession
+from selva.web import startup
+
+
+@startup
+async def startup(session: ScopedSession):
+    await session.scalar(text("select 1"))