### Define SQLAlchemy Configuration Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/cli.rst Example of a configuration file setup for Advanced Alchemy. ```python from sqlalchemy import create_engine from advanced_alchemy.config import SQLAlchemyConfig # Create a test config using SQLite config = SQLAlchemyConfig( connection_url="sqlite:///test.db" ) ``` -------------------------------- ### Install Advanced Alchemy using uv Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/getting-started.rst Install the library using the `uv` package manager. ```bash uv add advanced-alchemy ``` -------------------------------- ### Install and Update Dependencies Source: https://github.com/litestar-org/advanced-alchemy/blob/main/CONTRIBUTING.rst Commands to prepare the development environment and verify documentation builds. ```bash make install # Install all dependencies make upgrade # Update dependencies to the latest versions make docs # Verify documentation builds ``` -------------------------------- ### Create User with Profile Example Placeholder Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/services.rst This is a placeholder for an example demonstrating the creation of a user with a profile, leveraging Advanced Alchemy's service layer for recursive model creation from nested dictionaries. ```python from typing import Any async def create_user_with_profile(user_service: Any) -> Any: ``` -------------------------------- ### Sanic Integration Example Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/PYPI_README.md Initialize Advanced Alchemy with Sanic and register it using Sanic-Ext. This example uses an asynchronous configuration for Sanic applications. ```python from sanic import Sanic from sanic_ext import Extend from advanced_alchemy.extensions.sanic import AdvancedAlchemy, SQLAlchemyAsyncConfig app = Sanic("AlchemySanicApp") alchemy = AdvancedAlchemy( sqlalchemy_config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"), ) Extend.register(alchemy) ``` -------------------------------- ### Complete Litestar Session Example with SQLAlchemy Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst A full example demonstrating server-side sessions stored in a SQLite database using SQLAlchemy and Litestar's SessionMiddleware. This includes defining a session model, configuring the database and session backend, and setting up routes for login, logout, and home access. ```python from functools import partial from litestar import Litestar, get, post from litestar.connection import ASGIConnection from litestar.middleware.session import SessionMiddleware from litestar.middleware.session.server_side import ServerSideSessionConfig from advanced_alchemy.extensions.litestar import ( AsyncSessionConfig, SQLAlchemyAsyncConfig, SQLAlchemyPlugin, ) from advanced_alchemy.extensions.litestar.session import ( SessionModelMixin, SQLAlchemyAsyncSessionBackend, ) # Session model class WebSession(SessionModelMixin): __tablename__ = "web_sessions" # Database configuration alchemy_config = SQLAlchemyAsyncConfig( connection_string="sqlite+aiosqlite:///sessions.db", session_config=AsyncSessionConfig(expire_on_commit=False), create_all=True, ) # Session configuration session_config = ServerSideSessionConfig( max_age=3600, # 1 hour ) # Session backend session_backend = SQLAlchemyAsyncSessionBackend( config=session_config, alchemy_config=alchemy_config, model=WebSession, ) # Routes @get("/") async def home(request: ASGIConnection) -> dict: username = request.session.get("username") return {"message": f"Hello {username}!" if username else "Hello stranger!"} @post("/login") async def login(request: ASGIConnection) -> dict: form = await request.form() username = form.get("username") if username: request.set_session({"username": username, "login_time": "now"}) return {"message": f"Welcome {username}!"} return {"error": "Username required"} @post("/logout") async def logout(request: ASGIConnection) -> dict: request.clear_session() return {"message": "Logged out successfully"} # Application app = Litestar( route_handlers=[home, login, logout], plugins=[SQLAlchemyPlugin(config=alchemy_config)], middleware=[partial(SessionMiddleware, backend=session_backend)], ) ``` -------------------------------- ### Install Pre-release Packages with Pip Source: https://github.com/litestar-org/advanced-alchemy/blob/main/CONTRIBUTING.rst PyPI automatically handles pre-release versions. Users typically won't receive them with a standard install but can opt-in using the --pre flag or by specifying the exact version. ```bash pip install --pre advanced-alchemy ``` ```bash pip install advanced-alchemy==1.10.0a1 ``` -------------------------------- ### Install Advanced Alchemy using pipx Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/getting-started.rst Install the library in an isolated environment using `pipx`. ```bash pipx install advanced-alchemy ``` -------------------------------- ### Full Application Setup with Caching Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/caching.rst Shows a complete Litestar application setup including SQLAlchemy configuration with caching enabled for Redis. It defines a UserRepository and an endpoint to retrieve a user. ```python from litestar import Litestar from litestar.contrib.sqlalchemy.plugins import SQLAlchemyPlugin from sqlalchemy.ext.asyncio import AsyncSession from advanced_alchemy.cache import CacheConfig from advanced_alchemy.config import SQLAlchemyAsyncConfig from advanced_alchemy.repository import SQLAlchemyAsyncRepository # Assuming User model is defined elsewhere class User: ... # Configure database with caching db_config = SQLAlchemyAsyncConfig( connection_string="postgresql+asyncpg://user:pass@localhost/db", cache_config=CacheConfig( backend="dogpile.cache.redis", expiration_time=3600, arguments={"host": "localhost", "port": 6379, "db": 0}, key_prefix="myapp:" ) ) class UserRepository(SQLAlchemyAsyncRepository[User]): model_type = User async def get_user(session: AsyncSession, user_id: int) -> User: # Repository auto-retrieves cache_manager from session.info repo = UserRepository(session=session, auto_expunge=True) return await repo.get(user_id) app = Litestar( plugins=[SQLAlchemyPlugin(config=db_config)], # ... other app configurations ) ``` -------------------------------- ### Example JSON Fixture Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/database_seeding.rst This is an example of a JSON file used for seeding database records. It includes fields like name, description, price, and stock status. ```json [ { "name": "Laptop", "description": "High-performance laptop with 16GB RAM and 1TB SSD", "price": 999.99, "in_stock": true }, { "name": "Smartphone", "description": "Latest smartphone model with 5G and advanced camera", "price": 699.99, "in_stock": true } ] ``` -------------------------------- ### Example CSV Fixture Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/database_seeding.rst This is an example of a CSV file used for seeding database records. The header row defines the keys, and all values are initially strings. ```text name,price,in_stock Widget,9.99,true Gadget,19.99,true Thingy,4.99,false ``` -------------------------------- ### Manual Cache Setup Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/caching.rst Demonstrates how to manually set up a CacheManager using memory backend and register cache listeners without using the configuration system. The cache manager is then passed explicitly to repositories. ```python from advanced_alchemy.cache import CacheConfig, CacheManager, setup_cache_listeners from sqlalchemy.ext.asyncio import AsyncSession # Assuming UserRepository and session are defined elsewhere class UserRepository: ... session: AsyncSession # Create cache manager cache_manager = CacheManager(CacheConfig(backend="dogpile.cache.memory")) # Register listeners once at startup setup_cache_listeners() # Pass cache_manager explicitly to repositories repo = UserRepository(session=session, cache_manager=cache_manager) ``` -------------------------------- ### Install Advanced Alchemy using pip Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/getting-started.rst Use this command to install the library with pip. Ensure you are using Python 3. ```bash python3 -m pip install advanced-alchemy ``` -------------------------------- ### Basic Flask Setup with Sync SQLAlchemy Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/flask.rst Demonstrates setting up Advanced Alchemy with a standard Flask application using synchronous SQLAlchemy configurations. Sessions are cached on Flask's application context. ```python from flask import Flask from sqlalchemy import select from sqlalchemy.orm import Mapped, mapped_column from advanced_alchemy.extensions.flask import AdvancedAlchemy, SQLAlchemySyncConfig, base class User(base.BigIntBase): __tablename__ = "flask_user_account" name: Mapped[str] app = Flask(__name__) alchemy = AdvancedAlchemy( SQLAlchemySyncConfig( connection_string="sqlite:///local.db", commit_mode="autocommit", create_all=True, ), app, ) @app.route("/users") def list_users() -> dict[str, list[dict[str, object]]]: session = alchemy.get_sync_session() users = session.execute(select(User)).scalars().all() return {"users": [{"id": user.id, "name": user.name} for user in users]} ``` -------------------------------- ### Create, Read, Update, Delete with Service Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/PYPI_README.md This snippet demonstrates using the `SQLAlchemySyncRepositoryService` to perform the same database operations as the repository example, but with a higher-level abstraction that handles type conversions. ```python from advanced_alchemy import base, repository, filters, service, config from sqlalchemy import create_engine from sqlalchemy.orm import Mapped, sessionmaker class User(base.UUIDBase): # you can optionally override the generated table name by manually setting it. __tablename__ = "user_account" # type: ignore[assignment] email: Mapped[str] name: Mapped[str] class UserService(service.SQLAlchemySyncRepositoryService[User]): """User repository.""" class Repo(repository.SQLAlchemySyncRepository[User]): """User repository.""" model_type = User repository_type = Repo db = config.SQLAlchemySyncConfig(connection_string="duckdb:///:memory:", session_config=config.SyncSessionConfig(expire_on_commit=False)) # Initializes the database. with db.get_engine().begin() as conn: User.metadata.create_all(conn) with db.get_session() as db_session: service = UserService(session=db_session) # 1) Create multiple users with `add_many` objs = service.create_many([ {"email": 'cody@litestar.dev', 'name': 'Cody'}, {"email": 'janek@litestar.dev', 'name': 'Janek'}, {"email": 'peter@litestar.dev', 'name': 'Peter'}, {"email": 'jacob@litestar.dev', 'name': 'Jacob'} ]) print(objs) print(f"Created {len(objs)} new objects.") # 2) Select paginated data and total row count. Pass additional filters as kwargs created_objs, total_objs = service.get_many_and_count(LimitOffset(limit=10, offset=0), name="Cody") print(f"Selected {len(created_objs)} records out of a total of {total_objs}.") # 3) Let's remove the batch of records selected. deleted_objs = service.delete_many([new_obj.id for new_obj in created_objs]) print(f"Removed {len(deleted_objs)} records out of a total of {total_objs}.") # 4) Let's count the remaining rows remaining_count = service.count() print(f"Found {remaining_count} remaining records after delete.") ``` -------------------------------- ### Database Health Check Example Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Demonstrates how to check database health using a session. Requires `alchemy_config` to be defined. ```python async def _check_db_status() -> None: async with alchemy_config.get_session() as db_session: a_value = await db_session.execute(text("SELECT 1")) if a_value.scalar_one() == 1: print("Database is healthy") else: print("Database is not healthy") anyio.run(_check_db_status) ``` -------------------------------- ### Bump to a Pre-release Version Source: https://github.com/litestar-org/advanced-alchemy/blob/main/CONTRIBUTING.rst Use these commands to bump the project to a specific pre-release version like alpha, beta, or release candidate. Ensure you have the 'make' utility installed. ```bash make pre-release version=1.10.0a1 # First alpha ``` ```bash make pre-release version=1.10.0a2 # Second alpha ``` ```bash make pre-release version=1.10.0b1 # First beta ``` ```bash make pre-release version=1.10.0rc1 # First release candidate ``` -------------------------------- ### SQLAlchemyAsyncConfig Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/reference/extensions/litestar/index.rst Configuration for asynchronous SQLAlchemy setup in Litestar. ```APIDOC ## SQLAlchemyAsyncConfig ### Description Configuration for asynchronous SQLAlchemy setup in Litestar. ### Class `advanced_alchemy.extensions.litestar.config.SQLAlchemyAsyncConfig` ``` -------------------------------- ### Registering File Storage Backend Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/fastapi.rst Configure and register an object storage backend for file storage. This example uses an S3-compatible backend. ```python from advanced_alchemy.types import storages from advanced_alchemy.types.file_object.backends.obstore import ObstoreBackend documents_backend = ObstoreBackend( key="documents", fs="s3://company-documents-prod/", ) storages.register_backend(documents_backend) ``` -------------------------------- ### Litestar Application Setup with SQLAlchemy Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Configures the Litestar application with SQLAlchemy asynchronous session management. ```python from litestar import Litestar from litestar.contrib.sqlalchemy.config import SQLAlchemyAsyncConfig, AsyncSessionConfig # Application setup session_config = AsyncSessionConfig(expire_on_commit=False) alchemy_config = SQLAlchemyAsyncConfig( connection_string="sqlite+aiosqlite:///test.sqlite", session_config=session_config, ) app = Litestar([], config=alchemy_config) ``` -------------------------------- ### Install Advanced Alchemy using Poetry Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/getting-started.rst Add the library to your project dependencies using Poetry. ```bash poetry add advanced-alchemy ``` -------------------------------- ### Sticky-After-Write Consistency Example Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/routing.rst Demonstrates 'sticky-after-write' consistency where reads after a write operation are routed to the primary until the transaction commits. This ensures read-your-writes. ```python async with session_maker() as session: repo = UserRepository(session=session) # Read routes to replica users = await repo.get_many() # Write routes to primary new_user = await repo.add(User(name="Alice")) # Read now routes to primary (sticky-after-write) user = await repo.get(new_user.id) # Commit resets stickiness await session.commit() # Read can use replica again users = await repo.get_many() ``` -------------------------------- ### Custom Msgpack Serialization Example Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/caching.rst Provides custom serializer and deserializer functions for Msgpack format, demonstrating how to configure the cache backend with these custom functions. ```python import msgpack from sqlalchemy import inspect from advanced_alchemy.cache import CacheConfig def msgpack_serializer(model): # Convert model to dict and serialize with msgpack mapper = inspect(model.__class__) data = {col.key: getattr(model, col.key) for col in mapper.columns} return msgpack.packb(data) def msgpack_deserializer(data, model_class): unpacked = msgpack.unpackb(data) return model_class(**unpacked) config = CacheConfig( backend="dogpile.cache.redis", serializer=msgpack_serializer, deserializer=msgpack_deserializer ) ``` -------------------------------- ### Cache Listener Setup Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/reference/cache.rst Function to set up cache listeners for automatic invalidation. ```APIDOC ## advanced_alchemy.cache._listeners.setup_cache_listeners ### Description Sets up cache listeners to automatically invalidate cache entries when model instances are modified. ### Function advanced_alchemy.cache._listeners.setup_cache_listeners(cache_manager: CacheManager, session_maker: Callable) -> None ``` -------------------------------- ### Install Advanced Alchemy using PDM Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/getting-started.rst Add the library to your project dependencies using PDM. ```bash pdm add advanced-alchemy ``` -------------------------------- ### SQLAlchemySyncConfig Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/reference/extensions/litestar/index.rst Configuration for synchronous SQLAlchemy setup in Litestar. ```APIDOC ## SQLAlchemySyncConfig ### Description Configuration for synchronous SQLAlchemy setup in Litestar. ### Class `advanced_alchemy.extensions.litestar.config.SQLAlchemySyncConfig` ``` -------------------------------- ### Flask Integration Example Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/PYPI_README.md Initialize Advanced Alchemy with Flask using a synchronous configuration. This is useful for standard Flask applications. ```python from flask import Flask from advanced_alchemy.extensions.flask import AdvancedAlchemy, SQLAlchemySyncConfig app = Flask(__name__) alchemy = AdvancedAlchemy( config=SQLAlchemySyncConfig(connection_string="duckdb:///:memory:"), app=app, ) ``` -------------------------------- ### FastAPI Integration Example Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/PYPI_README.md Initialize Advanced Alchemy with FastAPI using an asynchronous configuration. Suitable for FastAPI applications requiring async database operations. ```python from advanced_alchemy.extensions.fastapi import AdvancedAlchemy, SQLAlchemyAsyncConfig from fastapi import FastAPI app = FastAPI() alchemy = AdvancedAlchemy( config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"), app=app, ) ``` -------------------------------- ### Connection Pooling Configuration Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Example of configuring SQLAlchemy's QueuePool for session workloads to manage database connections efficiently. ```python from sqlalchemy.pool import QueuePool ``` -------------------------------- ### Starlette Integration Example Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/PYPI_README.md Initialize Advanced Alchemy with Starlette using an asynchronous configuration. This is for Starlette-based applications needing async database support. ```python from advanced_alchemy.extensions.starlette import AdvancedAlchemy, SQLAlchemyAsyncConfig from starlette.applications import Starlette app = Starlette() alchemy = AdvancedAlchemy( config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"), app=app, ) ``` -------------------------------- ### Basic Read/Write Routing Configuration Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/routing.rst Configure basic routing with a primary connection string and a list of read replica connection strings. This setup automatically directs reads to replicas and writes to the primary. ```python from advanced_alchemy.config import SQLAlchemyAsyncConfig from advanced_alchemy.config.routing import RoutingConfig config = SQLAlchemyAsyncConfig( routing_config=RoutingConfig( primary_connection_string="postgresql+asyncpg://user:pass@primary:5432/db", read_replicas=[ "postgresql+asyncpg://user:pass@replica1:5432/db", ], ), ) # Create session factory session_maker = config.create_session_maker() # Use with repository - reads automatically go to replica async with session_maker() as session: repo = UserRepository(session=session) users = await repo.get_many() # Routes to replica ``` -------------------------------- ### Litestar Application Setup with SQLAlchemy Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Configures the Litestar application with SQLAlchemy asynchronous support, including database connection string and session configuration. It registers the DocumentController and the SQLAlchemyPlugin. ```python alchemy_config = SQLAlchemyAsyncConfig( connection_string="sqlite+aiosqlite:///test.sqlite", session_config=AsyncSessionConfig(expire_on_commit=False), before_send_handler="autocommit", create_all=True, ) app = Litestar( route_handlers=[DocumentController], plugins=[SQLAlchemyPlugin(config=alchemy_config)] ) ``` -------------------------------- ### Concrete Table Inheritance (CTI) Example Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/modeling/inheritance.rst Maps each class to an independent table containing all its columns. Use when subclasses do not share columns and are queried independently. ```python from sqlalchemy.orm import Mapped, mapped_column from advanced_alchemy.base import UUIDAuditBase class Vehicle(UUIDAuditBase): __abstract__ = True name: Mapped[str] class Car(Vehicle): __tablename__ = "car" engine_type: Mapped[str] class Bicycle(Vehicle): __tablename__ = "bicycle" has_basket: Mapped[bool] ``` -------------------------------- ### Basic FastAPI and Advanced Alchemy Setup Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/fastapi.rst Configure SQLAlchemy with FastAPI using AdvancedAlchemy. This snippet shows the necessary imports and initialization for connecting to a SQLite database and integrating with a FastAPI app. ```python from typing import AsyncGenerator from fastapi import FastAPI from advanced_alchemy.extensions.fastapi import AdvancedAlchemy, AsyncSessionConfig, SQLAlchemyAsyncConfig alchemy_config = SQLAlchemyAsyncConfig( connection_string="sqlite+aiosqlite:///test.sqlite", session_config=AsyncSessionConfig(expire_on_commit=False), create_all=True, commit_mode="autocommit", ) app = FastAPI() alchemy = AdvancedAlchemy(config=alchemy_config, app=app) ``` -------------------------------- ### Litestar App with SQLAlchemy Plugin Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Configure and create a Litestar application with the SQLAlchemyPlugin. This setup includes defining route handlers, plugins, and pagination dependencies. ```python session_config=session_config, create_all=True, ) sa_plugin = SQLAlchemyPlugin(config=alchemy_config) app = Litestar( route_handlers=[AuthorController], plugins=[sa_plugin], dependencies={"limit_offset": Provide(provide_limit_offset_pagination, sync_to_thread=False)}, ) ``` -------------------------------- ### Litestar Routes for Session Management Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Provides example Litestar route handlers for logging in, accessing profile data, and logging out, demonstrating session data manipulation. ```python from litestar import Litestar, get, post from litestar.connection import ASGIConnection from litestar.response import Response @get("/login") async def login_form() -> str: return "
" @post("/login") async def login(request: ASGIConnection) -> Response: form = await request.form() username = form.get("username") # Set session data request.set_session({"user_id": 123, "username": username}) return Response("Logged in!", status_code=200) @get("/profile") async def profile(request: ASGIConnection) -> dict: # Access session data user_id = request.session.get("user_id") username = request.session.get("username") if not user_id: return {"error": "Not logged in"} return {"user_id": user_id, "username": username} @post("/logout") async def logout(request: ASGIConnection) -> str: # Clear session request.clear_session() return "Logged out!" ``` -------------------------------- ### Msgspec Struct Conversion Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/services.rst Shows how to convert a BlogPost model to a Msgspec struct. This example includes a fallback for when Msgspec is not installed. ```python try: from msgspec import Struct except ModuleNotFoundError: # pragma: no cover - optional dependency in docs examples class Struct: # type: ignore[no-redef] pass class BlogPostStruct(Struct): id: int title: str content: str published: bool def to_msgspec_schema(post_service: BlogPostService, post_model: BlogPost) -> BlogPostStruct: return post_service.to_schema(post_model, schema_type=BlogPostStruct) ``` -------------------------------- ### Create and List Heroes with SQLModel Repository Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/modeling/sqlmodel.rst Demonstrates creating an engine, session factory, and using a SQLModel repository to add and retrieve Hero instances. Ensure the database schema is created before operations. ```python from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine async def create_and_list_heroes() -> list[Hero]: engine = create_async_engine("sqlite+aiosqlite:///:memory:") async_session_factory = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession) async with engine.begin() as conn: await conn.run_sync(SQLModel.metadata.create_all) try: async with async_session_factory() as session: repo = HeroRepository(session=session) hero = Hero(name="Deadpool", secret_name="Dive Wilson") await repo.add(hero) await session.commit() return await repo.get_many() finally: await engine.dispose() ``` -------------------------------- ### FastAPI Author Controller Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/fastapi.rst Create a FastAPI router for managing authors using Advanced Alchemy's service layer. This example demonstrates GET, POST, PATCH, and DELETE operations. ```python from fastapi import APIRouter, Depends from uuid import UUID from advanced_alchemy.extensions.fastapi import filters author_router = APIRouter() @author_router.get(path="/authors", response_model=filters.OffsetPagination[Author]) async def list_authors( authors_service: Authors, limit_offset: Annotated[filters.LimitOffset, Depends(provide_limit_offset_pagination)], ) -> filters.OffsetPagination[Author]: """List authors.""" results, total = await authors_service.get_many_and_count(limit_offset) return authors_service.to_schema(results, total, filters=[limit_offset], schema_type=Author) @author_router.post(path="/authors", response_model=Author) async def create_author( authors_service: Authors, data: AuthorCreate, ) -> Author: """Create a new author.""" obj = await authors_service.create(data) return authors_service.to_schema(obj, schema_type=Author) @author_router.get(path="/authors/{author_id}", response_model=Author) async def get_author( authors_service: Authors, author_id: UUID, ) -> Author: """Get an existing author.""" obj = await authors_service.get(author_id) return authors_service.to_schema(obj, schema_type=Author) @author_router.get(path="/authors/{author_id}", response_model=Author) async def get_author( authors_service: Authors, author_id: UUID, ) -> Author: """Get an existing author.""" obj = await authors_service.get(author_id) return authors_service.to_schema(obj, schema_type=Author) @author_router.patch(path="/authors/{author_id}", response_model=Author) async def update_author( authors_service: Authors, data: AuthorUpdate, author_id: UUID, ) -> Author: """Update an author.""" obj = await authors_service.update(data, item_id=author_id) return authors_service.to_schema(obj, schema_type=Author) @author_router.delete(path="/authors/{author_id}") async def delete_author( authors_service: Authors, author_id: UUID, ) -> None: """Delete an author from the system.""" _ = await authors_service.delete(author_id) ``` -------------------------------- ### Cache Installation Check Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/reference/cache.rst Constant indicating whether dogpile.cache is installed. ```APIDOC ## advanced_alchemy.cache.DOGPILE_CACHE_INSTALLED ### Description Boolean constant indicating whether the optional ``dogpile.cache`` dependency is installed. ### Data advanced_alchemy.cache.DOGPILE_CACHE_INSTALLED (bool) ``` -------------------------------- ### Create, Read, Update, Delete with Repository Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/PYPI_README.md This snippet shows how to initialize a database, create multiple users using `add_many`, retrieve paginated data with a count using `get_many_and_count`, delete records using `delete_many`, and count remaining records. ```python from advanced_alchemy import base, repository, filters, config from sqlalchemy.orm import Mapped class User(base.UUIDBase): email: Mapped[str] name: Mapped[str] db = config.SQLAlchemySyncConfig(connection_string="duckdb:///:memory:") # Initializes the database. with db.get_engine().begin() as conn: User.metadata.create_all(conn) with db.get_session() as db_session: repo = UserRepository(session=db_session) # 1) Create multiple users with `add_many` bulk_users = [ {"email": 'cody@litestar.dev', 'name': 'Cody'}, {"email": 'janek@litestar.dev', 'name': 'Janek'}, {"email": 'peter@litestar.dev', 'name': 'Peter'}, {"email": 'jacob@litestar.dev', 'name': 'Jacob'} ] objs = repo.add_many([User(**raw_user) for raw_user in bulk_users]) db_session.commit() print(f"Created {len(objs)} new objects.") # 2) Select paginated data and total row count. Pass additional filters as kwargs created_objs, total_objs = repo.get_many_and_count(LimitOffset(limit=10, offset=0), name="Cody") print(f"Selected {len(created_objs)} records out of a total of {total_objs}.") # 3) Let's remove the batch of records selected. deleted_objs = repo.delete_many([new_obj.id for new_obj in created_objs]) print(f"Removed {len(deleted_objs)} records out of a total of {total_objs}.") # 4) Let's count the remaining rows remaining_count = repo.count() print(f"Found {remaining_count} remaining records after delete.") ``` -------------------------------- ### GUID Type for Platform-Independent IDs Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/modeling/types.rst Demonstrates the use of the GUID type for primary keys, which adapts to different database backends for storing UUIDs. Requires importing UUID from the 'uuid' module and GUID from advanced_alchemy.types. ```python from uuid import UUID from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from advanced_alchemy.base import CommonTableAttributes, orm_registry from advanced_alchemy.types import GUID class Base(CommonTableAttributes, DeclarativeBase): registry = orm_registry class ExternalIdentity(Base): __tablename__ = "external_identity" id: Mapped[UUID] = mapped_column(GUID, primary_key=True) ``` -------------------------------- ### Graduate from Pre-release to Stable Source: https://github.com/litestar-org/advanced-alchemy/blob/main/CONTRIBUTING.rst To move from a release candidate to a stable version, use the 'make release' command with the 'pre' bump. Alternatively, you can skip directly to the next patch or minor stable version. ```bash make release bump=pre # e.g. 1.10.0rc1 → 1.10.0 ``` ```bash make release bump=patch # From any version → next patch ``` ```bash make release bump=minor # From any version → next minor ``` -------------------------------- ### Install Advanced Alchemy CLI Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/cli.rst Install the CLI using common Python package managers. ```bash python3 -m pip install advanced-alchemy[cli] ``` ```bash uv add advanced-alchemy[cli] ``` ```bash pipx install advanced-alchemy[cli] ``` ```bash pdm add advanced-alchemy[cli] ``` ```bash poetry add advanced-alchemy[cli] ``` -------------------------------- ### Install Advanced Alchemy Cache dependency Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/reference/cache.rst Install the optional dogpile.cache dependency to enable caching functionality. ```bash pip install advanced-alchemy[dogpile] ``` -------------------------------- ### Create a Blog Post using Service Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/services.rst Demonstrates creating a new blog post using the BlogPostService. It converts the created post to a BlogPostResponse schema. ```python async def create_post(post_service: BlogPostService, data: BlogPostCreate) -> BlogPostResponse: post = await post_service.create(data=data, auto_commit=True) return post_service.to_schema(post, schema_type=BlogPostResponse) ``` -------------------------------- ### Initialize Alembic Migrations (Custom Path) Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Initializes the Alembic migrations directory using a custom path './alembic'. ```shell $ litestar database init ./alembic ``` -------------------------------- ### Get User with Row Locking Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/services.rst Use the `with_for_update` parameter in the `get` method to acquire a row lock. This is useful for preventing race conditions when modifying data. ```python from typing import Any async def get_user_for_update(user_service: Any, user_id: Any) -> Any: return await user_service.get(item_id=user_id, with_for_update=True) ``` -------------------------------- ### List Alembic Migration Templates Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/cli.rst Lists available Alembic migration templates. Use this to discover templates for the 'init' command or for selecting templates for new projects. ```bash alchemy list-templates --config path.to.alchemy-config.config ``` -------------------------------- ### Get Document Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Retrieves a specific document by its ID. ```APIDOC ## GET /documents/{document_id} ### Description Retrieves a specific document by its ID. ### Method GET ### Endpoint /documents/{document_id} ### Parameters #### Path Parameters - **document_id** (uuid) - Required - The unique identifier of the document to retrieve. ### Response #### Success Response (200) - **id** (UUID) - The unique identifier of the document. - **name** (str) - The name of the document. - **file_url** (Optional[Union[str, list[str]]]) - The signed URL for accessing the uploaded file, if any. ### Response Example { "id": "a1b2c3d4-e5f6-7890-1234-567890abcdef", "name": "Example Document", "file_url": "http://example.com/signed/url" } ``` -------------------------------- ### Initialize Alembic Migrations (Default Path) Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Initializes the Alembic migrations directory using the default path './migrations'. ```shell $ litestar database init ./migrations ``` -------------------------------- ### Prepare Release Script Source: https://github.com/litestar-org/advanced-alchemy/blob/main/CONTRIBUTING.rst Command to run the release preparation script with specific version parameters. ```bash uv run tools/prepare_release.py -c -i --base v{current_version} {new_version} ``` -------------------------------- ### EngineConfig Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/reference/extensions/litestar/index.rst Configuration for SQLAlchemy engine setup in Litestar. ```APIDOC ## EngineConfig ### Description Configuration for SQLAlchemy engine setup in Litestar. ### Class `advanced_alchemy.extensions.litestar.engine.EngineConfig` ``` -------------------------------- ### Create a GitHub Pre-release Source: https://github.com/litestar-org/advanced-alchemy/blob/main/CONTRIBUTING.rst Use the GitHub CLI to create a pre-release tag on GitHub. This is a manual step after pushing your code. ```bash gh release create v1.10.0a1 --prerelease --title "v1.10.0a1" ``` -------------------------------- ### Repository Usage for Base Class Queries Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/modeling/inheritance.rst Demonstrates creating a repository for a base class to query across all inherited types. Ensure the repository is typed with the base model. ```python from sqlalchemy.ext.asyncio import AsyncSession from advanced_alchemy.repository import SQLAlchemyAsyncRepository class EmployeeRepository(SQLAlchemyAsyncRepository[Employee]): model_type = Employee async def list_employees(db_session: AsyncSession) -> list[Employee]: repository = EmployeeRepository(session=db_session) return await repository.get_many() ``` -------------------------------- ### Changelog Placeholder Comments Source: https://github.com/litestar-org/advanced-alchemy/blob/main/CONTRIBUTING.rst Example of placeholder comments to be removed from the changelog during release. ```rst ``` -------------------------------- ### SQLAlchemy Engine Configuration Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/reference/extensions/starlette/index.rst Configuration classes for SQLAlchemy engine setup in Advanced Alchemy. ```APIDOC ## EngineConfig ### Description Base configuration class for SQLAlchemy engines. ### Class EngineConfig ### Attributes None explicitly documented in the provided text. ## SQLAlchemyAsyncConfig ### Description Configuration class for asynchronous SQLAlchemy engines. ### Class SQLAlchemyAsyncConfig ### Attributes None explicitly documented in the provided text. ## SQLAlchemySyncConfig ### Description Configuration class for synchronous SQLAlchemy engines. ### Class SQLAlchemySyncConfig ### Attributes None explicitly documented in the provided text. ``` -------------------------------- ### Get Author Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Retrieves a specific author by their unique ID, including associated book details. ```APIDOC ## GET /authors/{author_id} ### Description Get an existing author with details. ### Method GET ### Endpoint /authors/{author_id} ### Parameters #### Path Parameters - **author_id** (uuid) - Required - The unique identifier of the author to retrieve. ### Response #### Success Response (200) - **id** (UUID) - The unique identifier of the author. - **name** (str) - The name of the author. - **dob** (datetime.date) - The date of birth of the author. - **books** (list) - A list of books associated with the author. #### Response Example { "id": "123e4567-e89b-12d3-a456-426614174000", "name": "John Doe", "dob": "1990-01-01", "books": [ { "id": "f0e1d2c3-b4a5-6789-0123-456789abcdef", "title": "The Great Novel" } ] } ``` -------------------------------- ### FileObject Metadata Handling Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/modeling/types.rst Demonstrates how to create a FileObject with initial metadata and how to update its metadata. This shows the flexibility in associating custom data with stored files. ```python from advanced_alchemy.types import FileObject file_obj = FileObject( backend="documents", filename="test.txt", metadata={ "category": "document", "tags": ["important", "review"], }, ) # Update metadata file_obj.update_metadata({"priority": "high"}) ``` -------------------------------- ### Database Initialization Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/cli.rst Commands related to database initialization and schema generation. ```APIDOC ## Generate SQL Output ### Description Generates SQL output instead of executing commands directly. Useful for database initialization workflows, manual setup, or DBA review. ### Method CLI Command ### Endpoint N/A ### Parameters #### Query Parameters - **--sql** (flag) - Optional - Generate SQL output instead of executing. ### Request Example ```bash alchemy --sql ``` ### Response SQL script output. ``` -------------------------------- ### Cache Manager Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/reference/cache.rst Manages the cache region and provides methods for getting, setting, and deleting cached items. ```APIDOC ## advanced_alchemy.cache.CacheManager ### Description Manages the cache region and provides methods for getting, setting, and deleting cached items. ### Class advanced_alchemy.cache.CacheManager ### Members - **region**: (dogpile.cache.api.CacheRegion) - The configured cache region. ### Methods - **get(key: str)**: Retrieves an item from the cache. - **set(key: str, value: Any)**: Adds an item to the cache. - **delete(key: str)**: Removes an item from the cache. - **clear()**: Clears the entire cache. ``` -------------------------------- ### Invoke CLI Help Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/cli.rst Display the help menu for the alchemy command. ```bash alchemy --help ``` -------------------------------- ### Get Published Posts using NotNullFilter Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/repositories/filtering.rst Fetches posts where the 'published_at' field is NOT NULL. This implies the posts have been published. ```python async def get_published_posts(db_session: AsyncSession) -> list[FilteringPost]: repository = FilteringPostRepository(session=db_session) return await repository.get_many(NotNullFilter(field_name="published_at")) ``` -------------------------------- ### Multiple Database Configurations Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Shows how to configure multiple SQLAlchemy instances with distinct bind keys and session/engine dependency keys for use in Litestar. This allows for managing different databases. ```python primary_config = SQLAlchemyAsyncConfig( connection_string="sqlite+aiosqlite:///primary.sqlite", bind_key="primary", session_dependency_key="primary_session", engine_dependency_key="primary_engine", ) reporting_config = SQLAlchemyAsyncConfig( connection_string="sqlite+aiosqlite:///reporting.sqlite", bind_key="reporting", session_dependency_key="reporting_session", engine_dependency_key="reporting_engine", ) app = Litestar( route_handlers=[], plugins=[SQLAlchemyPlugin(config=[primary_config, reporting_config])], ) @get("/reports") async def get_reports(reporting_session: AsyncSession) -> str: return "Reporting database is available" ``` -------------------------------- ### Get Unpublished Posts using NullFilter Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/repositories/filtering.rst Retrieves posts where the 'published_at' field is NULL. This indicates posts that have not been published. ```python async def get_unpublished_posts(db_session: AsyncSession) -> list[FilteringPost]: repository = FilteringPostRepository(session=db_session) return await repository.get_many(NullFilter(field_name="published_at")) ``` -------------------------------- ### Litestar Plugin Configuration Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/reference/extensions/litestar/plugins.rst Configuration options for integrating Advanced Alchemy with Litestar, including database engine setup. ```APIDOC ## Advanced Alchemy Litestar Plugin Configuration ### Description This section covers the configuration classes for integrating Advanced Alchemy's database features with the Litestar framework. It includes options for both asynchronous and synchronous SQLAlchemy configurations. ### Classes - **EngineConfig** (Base class for engine configurations) - **SQLAlchemyAsyncConfig** (Configuration for asynchronous SQLAlchemy engines) - **SQLAlchemySyncConfig** (Configuration for synchronous SQLAlchemy engines) ### Usage These classes are used to define and manage your SQLAlchemy database connection within a Litestar application. You would typically instantiate `SQLAlchemyAsyncConfig` or `SQLAlchemySyncConfig` and pass it to the Litestar application or relevant plugin. #### Example (Conceptual) ```python from litestar import Litestar from advanced_alchemy.extensions.litestar.plugins import SQLAlchemyAsyncConfig config = SQLAlchemyAsyncConfig( connection_string="postgresql+asyncpg://user:password@host:port/database" ) app = Litestar(plugins=[config.plugin]) ``` ### Parameters #### SQLAlchemyAsyncConfig - **connection_string** (str) - Required - The database connection URL. - **engine_options** (dict) - Optional - Additional options to pass to the SQLAlchemy engine. - **session_config** (dict) - Optional - Configuration for the SQLAlchemy session. #### SQLAlchemySyncConfig - **connection_string** (str) - Required - The database connection URL. - **engine_options** (dict) - Optional - Additional options to pass to the SQLAlchemy engine. - **session_config** (dict) - Optional - Configuration for the SQLAlchemy session. ### Response This plugin primarily affects application startup and dependency injection. Successful configuration leads to a Litestar application with database capabilities enabled. Errors during configuration will typically raise exceptions during application startup. ``` -------------------------------- ### Pydantic Schema Conversion Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/services.rst Demonstrates converting a BlogPost model to a Pydantic schema using a BlogPostService. Ensure Pydantic is installed. ```python from pydantic import BaseModel class BlogPostSchema(BaseModel): id: int title: str content: str published: bool model_config = {"from_attributes": True} def to_pydantic_schema(post_service: BlogPostService, post_model: BlogPost) -> BlogPostSchema: return post_service.to_schema(post_model, schema_type=BlogPostSchema) ``` -------------------------------- ### Create a Basic Async Repository Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/repositories/basics.rst Implement a basic asynchronous repository for a given model using SQLAlchemyAsyncRepository. This snippet also shows how to create a new record. ```python from advanced_alchemy.repository import SQLAlchemyAsyncRepository from sqlalchemy.ext.asyncio import AsyncSession class PostRepository(SQLAlchemyAsyncRepository[Post]): """Repository for managing blog posts.""" model_type = Post async def create_post(db_session: AsyncSession, title: str, content: str) -> Post: repository = PostRepository(session=db_session) return await repository.add(Post(title=title, content=content), auto_commit=True) ``` -------------------------------- ### Create Author Endpoint Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/flask.rst Example of a Flask POST endpoint to create a new author using AuthorService and Advanced Alchemy. ```python @app.route("/authors", methods=["POST"]) def create_author(): author_service = AuthorService(session=alchemy.get_sync_session()) author = author_service.create(data=request.get_json()) return author_service.jsonify(author_service.to_schema(author, schema_type=AuthorSchema)) ``` -------------------------------- ### Run Database Migrations Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/database_seeding.rst Command to execute database schema upgrades via the CLI. ```text uv run app database upgrade ``` -------------------------------- ### Author Controller Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/frameworks/litestar.rst Example of a Litestar controller for Author CRUD operations, demonstrating dependency injection for services and built-in pagination and filtering. ```APIDOC ## GET /authors ### Description List all authors with pagination. ### Method GET ### Endpoint /authors ### Parameters #### Query Parameters - **filters** (Annotated[list[filters.FilterTypes], Dependency(skip_validation=True)]) - Required - List of filters to apply. ### Response #### Success Response (200) - **results** (list[Author]) - List of authors. - **total** (int) - Total number of authors. #### Response Example { "results": [ { "id": "123e4567-e89b-12d3-a456-426614174000", "name": "John Doe", "books": [] } ], "total": 1 } ## POST /authors ### Description Create a new author. ### Method POST ### Endpoint /authors ### Parameters #### Request Body - **data** (AuthorCreate) - Required - Author data to create. ### Response #### Success Response (200) - **Author** - The created author object. #### Response Example { "id": "123e4567-e89b-12d3-a456-426614174000", "name": "Jane Doe", "books": [] } ## GET /authors/{author_id:uuid} ### Description Get an existing author. ### Method GET ### Endpoint /authors/{author_id:uuid} ### Parameters #### Path Parameters - **author_id** (uuid) - Required - The author to retrieve. ### Response #### Success Response (200) - **Author** - The retrieved author object. #### Response Example { "id": "123e4567-e89b-12d3-a456-426614174000", "name": "John Doe", "books": [] } ## PATCH /authors/{author_id:uuid} ### Description Update an author. ### Method PATCH ### Endpoint /authors/{author_id:uuid} ### Parameters #### Path Parameters - **author_id** (uuid) - Required - The author to update. #### Request Body - **data** (AuthorUpdate) - Required - Author data to update. ### Response #### Success Response (200) - **Author** - The updated author object. #### Response Example { "id": "123e4567-e89b-12d3-a456-426614174000", "name": "John Doe Updated", "books": [] } ## DELETE /authors/{author_id:uuid} ### Description Delete an author from the system. ### Method DELETE ### Endpoint /authors/{author_id:uuid} ### Parameters #### Path Parameters - **author_id** (uuid) - Required - The author to delete. ### Response #### Success Response (200) - **None** - Indicates successful deletion. ``` -------------------------------- ### Manage Database Migrations Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/cli.rst Commands for interacting with database revisions and migration files. ```bash alchemy show-current-revision --config path.to.alchemy-config.config ``` ```bash alchemy downgrade --config path.to.alchemy-config.config [REVISION] ``` ```bash alchemy upgrade --config path.to.alchemy-config.config [REVISION] ``` ```bash alchemy stamp --config path.to.alchemy-config.config REVISION ``` ```bash alchemy init --config path.to.alchemy-config.config [DIRECTORY] ``` -------------------------------- ### Coerce and Seed Data from CSV Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/database_seeding.rst Example of transforming raw CSV rows into typed models and persisting them using an asynchronous repository. ```python return { "name": row["name"], "price": float(row["price"]), "in_stock": row["in_stock"].lower() == "true", } async def seed_from_csv(repository: AsyncProductRepository, fixtures_path: Path) -> None: raw_rows = await open_fixture_async(fixtures_path, "products") products = [AsyncProduct(**coerce_product_row(item)) for item in raw_rows] await repository.add_many(products, auto_commit=True) ``` -------------------------------- ### Define Filtering Models and Repository Source: https://github.com/litestar-org/advanced-alchemy/blob/main/docs/usage/repositories/filtering.rst Defines the SQLAlchemy model 'FilteringPost' and its corresponding repository 'FilteringPostRepository'. This serves as the base for filtering and pagination examples. ```python import datetime from typing import Optional from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Mapped, mapped_column from advanced_alchemy.base import BigIntAuditBase from advanced_alchemy.filters import CollectionFilter, LimitOffset, NotNullFilter, NullFilter, SearchFilter from advanced_alchemy.repository import SQLAlchemyAsyncRepository class FilteringPost(BigIntAuditBase): __tablename__ = "filtering_post" title: Mapped[str] content: Mapped[str] published: Mapped[bool] = mapped_column(default=False) published_at: Mapped[Optional[datetime.datetime]] = mapped_column(default=None) class FilteringPostRepository(SQLAlchemyAsyncRepository[FilteringPost]): model_type = FilteringPost ```