Advanced Alchemy

Common errors

• sqlalchemy.exc.InvalidRequestError: This session is in 'pending' state, and no further operations are allowed.

  cause Attempting to use a SQLAlchemy session that has already been committed, rolled back, or is in an invalid state, often due to improper `session.close()` or `session.commit()` calls.
  fix Ensure your session is managed correctly, typically using `async with session_maker() as session:` for asynchronous operations. Pass an active session to your repository instance.

• ModuleNotFoundError: No module named 'alembic.templates.sync'

  cause This specific error often indicates an issue with `advanced-alchemy`'s Alembic integration or templating, where it might incorrectly reference or generate templates. This was a known bug.
  fix Upgrade `advanced-alchemy` to version `1.9.2` or later, as this specific import issue was fixed in `v1.9.1` for migration templates. If manually creating templates, ensure correct import paths.

• sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint

  cause Attempting to insert a record with a unique constraint violation (e.g., duplicate email address for a unique column) when using a repository `add` or `update` method.
  fix Before adding or updating, check for the existence of the record with the unique key. Wrap `repository.add()` or `repository.update()` calls in a `try...except IntegrityError` block to handle the specific violation gracefully.

Warnings

• gotcha When using `advanced-alchemy` repositories, ensure proper session management. Incorrectly handling sessions (e.g., closing too early or committing at the wrong time) can lead to `InvalidRequestError` or mask original exceptions, making debugging difficult.
  Fix: Always use `session_maker()` within an `async with` block (for async) or `with` block (for sync) to ensure sessions are properly managed and closed. Repositories expect an active session.
• gotcha When integrating with Alembic for migrations, ensure that `advanced-alchemy`'s base model (`BaseORMModel`) is correctly imported in your `env.py` or migration scripts. Older versions had issues with missing imports in generated templates.
  Fix: Ensure `from advanced_alchemy.base import BaseORMModel` (or the equivalent for your custom base model) is present in your migration scripts, especially `env.py`. Update to `v1.9.2` or later for template fixes.
• breaking The `Repository` class in older versions (e.g., <1.0.0, before `advanced-alchemy` was 'litestar-alchemy') was renamed and refactored significantly. Users upgrading from very old versions might encounter import errors or API changes.
  Fix: Refer to the official documentation for the current `advanced-alchemy` API, particularly `SQLAlchemyAsyncRepository` and `SQLAlchemySyncRepository`, and update import paths and method calls accordingly. This is less relevant for recent upgrades but a major historical change.

Install

• pip install advanced-alchemy Install core library
• pip install advanced-alchemy[sqlmodel] Install with SQLModel support
• pip install advanced-alchemy[litestar] Install with Litestar integration
• pip install advanced-alchemy[alembic] Install with Alembic support

Imports

• BaseORMModel

  from advanced_alchemy.base import BaseORMModel

• SQLAlchemyAsyncRepository

  from advanced_alchemy.repository import SQLAlchemyAsyncRepository

• SQLAlchemySyncRepository

  from advanced_alchemy.repository import SQLAlchemySyncRepository

• SQLAlchemyLitestarConfig

  from advanced_alchemy.config import SQLAlchemyLitestarConfig

• SQLAlchemyPlugin

  from advanced_alchemy.extensions.litestar.plugins import SQLAlchemyPlugin

Quickstart

This quickstart demonstrates how to define an ORM model, configure an asynchronous SQLAlchemy session using `AsyncSessionConfig`, and perform CRUD operations using `SQLAlchemyAsyncRepository`. It sets up an in-memory SQLite database for simplicity, creates a `User` model, and then uses the repository to add, list, update, and delete user records. The example uses `asyncio` to run the asynchronous operations.

import os
from typing import AsyncGenerator
from uuid import UUID, uuid4

from advanced_alchemy.base import BaseORMModel
from advanced_alchemy.config import AsyncSessionConfig
from advanced_alchemy.repository import SQLAlchemyAsyncRepository
from sqlalchemy import String
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Mapped, mapped_column

# Define a database URL (using a simple in-memory SQLite for example)
DB_URL = os.environ.get('DB_URL', 'sqlite+aiosqlite:///:memory:')

# 1. Define your ORM Model
class User(BaseORMModel):
    __tablename__ = "users"

    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    name: Mapped[str] = mapped_column(String(length=255))
    email: Mapped[str] = mapped_column(String(length=255), unique=True)

# 2. Configure SQLAlchemy
async_session_config = AsyncSessionConfig(url=DB_URL)

# 3. Create a Repository for your Model
class UserRepository(SQLAlchemyAsyncRepository[User]):
    model_type = User

# 4. Example Usage (e.g., in an async function)
async def main():
    # Ensure the database tables are created
    async with async_session_config.get_engine().begin() as conn:
        await conn.run_sync(BaseORMModel.metadata.create_all)

    # Get a session maker
    session_maker = async_session_config.get_session_maker()

    # Create an instance of the repository with a session
    async with session_maker() as session:
        user_repo = UserRepository(session=session)

        # Create a new user
        new_user = await user_repo.add(User(name='Alice', email='alice@example.com'))
        print(f"Created user: {new_user.name} ({new_user.id})")

        # Fetch all users
        users = await user_repo.list()
        print(f"All users: {[u.name for u in users]}")

        # Update a user
        updated_user = await user_repo.update(User(id=new_user.id, name='Alicia', email='alicia@example.com'))
        print(f"Updated user: {updated_user.name}")

        # Delete a user
        await user_repo.delete(updated_user.id)
        deleted_users = await user_repo.list()
        print(f"Users after deletion: {[u.name for u in deleted_users]}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())