Refactoring examples

aiohttp_sqlalchemy/decorators.py

@@ -11,6 +11,7 @@
 
 def sa_decorator(key: str = SA_DEFAULT_KEY) -> THandlerWrapper:
     """SQLAlchemy asynchronous handler decorator."""
+
     def wrapper(handler: THandler) -> THandler:
         @wraps(handler)
         async def wrapped(*args: Any, **kwargs: Any) -> StreamResponse:

aiohttp_sqlalchemy/middlewares.py

@@ -7,6 +7,7 @@
 
 def sa_middleware(key: str = SA_DEFAULT_KEY) -> THandler:
     """SQLAlchemy asynchronous middleware factory."""
+
     @middleware
     async def sa_middleware_(request: Request, handler: THandler) -> StreamResponse:
         if key in request:

examples/complex_example.py

@@ -0,0 +1,158 @@
+from datetime import datetime
+from secrets import choice
+from typing import Any
+
+import sqlalchemy as sa
+from aiohttp import web
+from sqlalchemy import orm
+
+import aiohttp_sqlalchemy
+from aiohttp_sqlalchemy import SAView, sa_decorator, sa_session
+
+metadata = sa.MetaData()
+Base: Any = orm.declarative_base(metadata=metadata)
+DB_URL = "sqlite+aiosqlite:///"
+
+
+class MyModel(Base):
+    __tablename__ = "my_table"
+    pk = sa.Column(sa.Integer, primary_key=True)
+    timestamp = sa.Column(sa.DateTime(), default=datetime.now)
+
+
+SECOND_KEY = "sa_second"
+THIRD_KEY = "sa_third"
+FOURTH_KEY = "sa_fourth"
+
+
+def add_instance(session):
+    session.add_all([MyModel()])
+
+
+def select_instances(session):
+    stmt = sa.select(MyModel)
+    result = session.execute(stmt)
+    instances = result.scalars()
+    return {instance.pk: instance.timestamp.isoformat() for instance in instances}
+
+
+@sa_decorator(THIRD_KEY)
+@sa_decorator(FOURTH_KEY)
+async def function_handler(request):
+    data = {}
+    main_session = sa_session(request)
+    second_session = sa_session(request, SECOND_KEY)
+    third_session = sa_session(request, THIRD_KEY)
+    fourth_session = sa_session(request, FOURTH_KEY)
+    random_session = choice(
+        [main_session, second_session, third_session, fourth_session]
+    )
+
+    async with random_session.begin():
+        await random_session.run_sync(add_instance)
+
+    async with main_session.begin():
+        data["main"] = await main_session.run_sync(select_instances)
+
+    async with second_session.begin():
+        data["second"] = await second_session.run_sync(select_instances)
+
+    async with third_session.begin():
+        data["third"] = await third_session.run_sync(select_instances)
+
+    async with fourth_session.begin():
+        data["fourth"] = await fourth_session.run_sync(select_instances)
+
+    return web.json_response(data)
+
+
+class ClassOrganizedHandler:
+    @sa_decorator(THIRD_KEY)
+    @sa_decorator(FOURTH_KEY)
+    async def get(self, request):
+        data = {}
+        main_session = sa_session(request)
+        second_session = sa_session(request, SECOND_KEY)
+        third_session = sa_session(request, THIRD_KEY)
+        fourth_session = sa_session(request, FOURTH_KEY)
+        random_session = choice(
+            [main_session, second_session, third_session, fourth_session]
+        )
+
+        async with random_session.begin():
+            await random_session.run_sync(add_instance)
+
+        async with main_session.begin():
+            data["main"] = await main_session.run_sync(select_instances)
+
+        async with second_session.begin():
+            data["second"] = await second_session.run_sync(select_instances)
+
+        async with third_session.begin():
+            data["third"] = await third_session.run_sync(select_instances)
+
+        async with fourth_session.begin():
+            data["fourth"] = await fourth_session.run_sync(select_instances)
+
+        return web.json_response(data)
+
+
+class ClassBasedView(SAView):
+    @sa_decorator(THIRD_KEY)
+    @sa_decorator(FOURTH_KEY)
+    async def get(self):
+        data = {}
+        main_session = self.sa_session()
+        second_session = self.sa_session(SECOND_KEY)
+        third_session = self.sa_session(THIRD_KEY)
+        fourth_session = self.sa_session(FOURTH_KEY)
+        random_session = choice(
+            [main_session, second_session, third_session, fourth_session]
+        )
+
+        async with random_session.begin():
+            await random_session.run_sync(add_instance)
+
+        async with main_session.begin():
+            data["main"] = await main_session.run_sync(select_instances)
+
+        async with second_session.begin():
+            data["second"] = await second_session.run_sync(select_instances)
+
+        async with third_session.begin():
+            data["third"] = await third_session.run_sync(select_instances)
+
+        async with fourth_session.begin():
+            data["fourth"] = await fourth_session.run_sync(select_instances)
+
+        return web.json_response(data)
+
+
+async def app_factory():
+    app = web.Application()
+    aiohttp_sqlalchemy.setup(
+        app,
+        [
+            aiohttp_sqlalchemy.bind(DB_URL),
+            aiohttp_sqlalchemy.bind(DB_URL, SECOND_KEY),
+            aiohttp_sqlalchemy.bind(DB_URL, THIRD_KEY, middleware=False),
+            aiohttp_sqlalchemy.bind(DB_URL, FOURTH_KEY, middleware=False),
+        ],
+    )
+    await aiohttp_sqlalchemy.init_db(app, metadata)
+    await aiohttp_sqlalchemy.init_db(app, metadata, SECOND_KEY)
+    await aiohttp_sqlalchemy.init_db(app, metadata, THIRD_KEY)
+    await aiohttp_sqlalchemy.init_db(app, metadata, FOURTH_KEY)
+
+    app.add_routes([web.get("/function_handler", function_handler)])
+
+    class_organized_handler = ClassOrganizedHandler()
+    app.add_routes([web.get("/class_organized_handler", class_organized_handler.get)])
+
+    app.add_routes([web.view("/class_based_view", ClassBasedView)])
+
+    return app
+
+
+if __name__ == "__main__":
+    web.run_app(app_factory())