starlette example

Author: krylosov-aa

Makefile

@@ -5,7 +5,13 @@ lint:
 	flake8 .
 
 test:
-	pytest --cov context_async_sqlalchemy examples/fastapi_example/tests --cov-report=term-missing
+	pytest --cov context_async_sqlalchemy examples/fastapi_example/tests examples/starlette_example/tests --cov-report=term-missing
+
+test_fastapi:
+	pytest examples/fastapi_example/tests
+
+test_starlette:
+	pytest examples/starlette_example/tests
 
 uv:
 	uv sync

examples/starlette_example/README.md

@@ -0,0 +1,3 @@
+# Starlette example
+
+An example of how to use the library with Starlette

examples/starlette_example/__init__.py

@@ -0,0 +1,3 @@
+"""
+An example of how to use the library with Starlette
+"""

examples/starlette_example/database.py

@@ -0,0 +1,48 @@
+"""
+The module that defines the database connection parameters
+"""
+
+from sqlalchemy.ext.asyncio import (
+    async_sessionmaker,
+    AsyncEngine,
+    AsyncSession,
+    create_async_engine,
+)
+
+from context_async_sqlalchemy import DBConnect
+
+
+def create_engine(host: str) -> AsyncEngine:
+    """
+    database connection parameters
+    """
+
+    # In production code, you will probably take these parameters from env
+    pg_user = "krylosov-aa"
+    pg_password = ""
+    pg_port = 6432
+    pg_db = "test"
+    return create_async_engine(
+        f"postgresql+asyncpg://"
+        f"{pg_user}:{pg_password}"
+        f"@{host}:{pg_port}"
+        f"/{pg_db}",
+        future=True,
+        pool_pre_ping=True,
+    )
+
+
+def create_session_maker(
+    engine: AsyncEngine,
+) -> async_sessionmaker[AsyncSession]:
+    """session parameters"""
+    return async_sessionmaker(
+        engine, class_=AsyncSession, expire_on_commit=False
+    )
+
+
+connection = DBConnect(
+    host="127.0.0.1",
+    engine_creator=create_engine,
+    session_maker_creator=create_session_maker,
+)

examples/starlette_example/main.py

@@ -0,0 +1,6 @@
+"""Application entry point"""
+
+from .setup_app import setup_app
+
+
+app = setup_app()

examples/starlette_example/models.py

@@ -0,0 +1,39 @@
+import uuid
+from typing import Any
+from pydantic import GetCoreSchemaHandler
+from pydantic_core.core_schema import IsInstanceSchema, is_instance_schema
+from sqlalchemy import text, UUID, Text
+from sqlalchemy.orm import (
+    DeclarativeBase,
+    Mapped,
+    mapped_column,
+    MappedColumn,
+)
+
+
+class BaseTable(DeclarativeBase):
+    """Base table for all tables"""
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls,
+        _source_type: Any,
+        _handler: GetCoreSchemaHandler,
+    ) -> IsInstanceSchema:
+        return is_instance_schema(cls)
+
+
+def uuid_pk_column() -> MappedColumn[uuid.UUID]:
+    """Allows you to reuse a column definition without inheritance"""
+    return mapped_column(
+        UUID, primary_key=True, server_default=text("gen_random_uuid()")
+    )
+
+
+class ExampleTable(BaseTable):
+    """just a table for example"""
+
+    id: Mapped[uuid.UUID] = uuid_pk_column()
+    text: Mapped[str] = mapped_column(Text)
+
+    __tablename__ = "example"

examples/starlette_example/routes/README.md

@@ -0,0 +1,6 @@
+# Handlers example
+
+A module that contains all the application handlers.
+
+The examples are fictional, but they’re designed to demonstrate the
+various capabilities of the library.

examples/starlette_example/routes/__init__.py

@@ -0,0 +1,3 @@
+"""
+A module that contains all the application handlers.
+"""

examples/starlette_example/routes/atomic_usage.py

@@ -0,0 +1,33 @@
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from context_async_sqlalchemy import atomic_db_session, db_session
+from sqlalchemy import insert
+
+from ..database import connection
+from ..models import ExampleTable
+
+
+async def handler_with_db_session_and_atomic(_: Request) -> JSONResponse:
+    """
+    Let's imagine you already have a function that works with a contextual
+    session, and its use case calls autocommit at the end of the request.
+    You want to reuse this function, but you need to commit immediately,
+        rather than wait for the request to complete.
+    """
+    # the transaction will be committed or rolled back automatically
+    # using the context manager
+    async with atomic_db_session(connection):
+        await _insert_1()
+
+    # This is a new transaction in the same connection
+    await _insert_1()
+
+    return JSONResponse({})
+
+
+async def _insert_1() -> None:
+    session = await db_session(connection)
+    stmt = insert(ExampleTable).values(
+        text="example_with_db_session_and_atomic"
+    )
+    await session.execute(stmt)

examples/starlette_example/routes/early_connection_close.py

@@ -0,0 +1,44 @@
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from context_async_sqlalchemy import (
+    atomic_db_session,
+    close_db_session,
+    db_session,
+)
+from sqlalchemy import insert
+
+from ..database import connection
+from ..models import ExampleTable
+
+
+async def handler_with_early_connection_close(_: Request) -> JSONResponse:
+    """
+    An example when you can return a connection to the connection pool for a
+        long period of work unrelated to the database
+    """
+    async with atomic_db_session(connection):
+        await _insert()
+
+    await close_db_session(connection)
+
+    ...
+    # There's a lot of work going on here.
+    ...
+
+    # new connect and new transaction
+    await _insert()
+
+    return JSONResponse({})
+
+
+async def _insert() -> None:
+    """
+    Let's imagine that the same function is useful to us in another handle,
+        where it is good to use it in a common transaction for the entire
+        request.
+    """
+    session = await db_session(connection)
+    stmt = insert(ExampleTable).values(
+        text="example_with_early_connection_close"
+    )
+    await session.execute(stmt)