用 Litestar Repository 构建作者 CRUD，并扩展示例 Slug

四章原始示例的完整代码对照

以下按各章顺序保留官方示例，包括片段与展开后的完整程序，供对照中文解释和编辑修订。不同章节、同步/异步和Python版本变体不能合并运行；原始例中的吞异常、PATCH空值、Slug并发与长度问题仍须按正文处理。源文第一章在当前版本使用NamedDependency与create_all配置，必须按相同版本选用完整示例。原例debug=True、文件SQLite和删除操作仅限隔离教学环境。未执行任何示例。

基类的其他细节：GUID优先使用数据库原生UUID，否则回退Binary(16)；BigIntIdentity在不支持的后端回退Integer；JsonB优先原生JSONB，否则使用Binary或Blob。EncryptedString支持多个加密后端。这些是源文接口说明，不是兼容性实测。

第1章原代码

来源：Litestar官方第1章，©2025 Litestar-Org，MIT；代码未修改。

示例 1

# The `UUIDBase` class includes a `UUID` based primary key (`id`)
class Author(base.UUIDBase):
    __tablename__ = "author"
    name: Mapped[str]

示例 2

# The `UUIDBase` class includes a `UUID` based primary key (`id`)
class Author(base.UUIDBase):
    __tablename__ = "author"
    name: Mapped[str]
    dob: Mapped[date]
    books: Mapped[List[Book]] = relationship(back_populates="author", lazy="selectin")
# record created, and `updated_at` is the last time the record was modified.
class Book(base.UUIDAuditBase):
    __tablename__ = "book"
    title: Mapped[str]

示例 3

from __future__ import annotations

import uuid
from datetime import date
from uuid import UUID

from sqlalchemy import ForeignKey, func, select
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
from sqlalchemy.orm import Mapped, mapped_column, relationship

from litestar import Litestar, get
from litestar.di import NamedDependency
from litestar.plugins.sqlalchemy import AsyncSessionConfig, SQLAlchemyAsyncConfig, SQLAlchemyPlugin, base


# The SQLAlchemy base includes a declarative model for you to use in your models.
# The `UUIDBase` class includes a `UUID` based primary key (`id`)
class Author(base.UUIDBase):
    __tablename__ = "author"
    name: Mapped[str]
    dob: Mapped[date]
    books: Mapped[list[Book]] = relationship(back_populates="author", lazy="selectin")


# The `UUIDAuditBase` class includes the same UUID` based primary key (`id`) and 2
# additional columns: `created_at` and `updated_at`. `created_at` is a timestamp of when the
# record created, and `updated_at` is the last time the record was modified.
class Book(base.UUIDAuditBase):
    __tablename__ = "book"
    title: Mapped[str]
    author_id: Mapped[UUID] = mapped_column(ForeignKey("author.id"))
    author: Mapped[Author] = relationship(lazy="joined", innerjoin=True, viewonly=True)


session_config = AsyncSessionConfig(expire_on_commit=False)
sqlalchemy_config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///test.sqlite", session_config=session_config, create_all=True
)  # Create 'async_session' dependency.


async def on_startup(app: Litestar) -> None:
    """Adds some dummy data if no data is present."""
    async with sqlalchemy_config.get_session() as session:
        statement = select(func.count()).select_from(Author)
        count = await session.execute(statement)
        if not count.scalar():
            author_id = uuid.uuid4()
            session.add(Author(name="Stephen King", dob=date(1954, 9, 21), id=author_id))
            session.add(Book(title="It", author_id=author_id))
            await session.commit()


@get(path="/authors")
async def get_authors(
    db_session: NamedDependency[AsyncSession], db_engine: NamedDependency[AsyncEngine]
) -> list[Author]:
    """Interact with SQLAlchemy engine and session."""
    return list(await db_session.scalars(select(Author)))


app = Litestar(
    route_handlers=[get_authors],
    on_startup=[on_startup],
    debug=True,
    plugins=[SQLAlchemyPlugin(config=sqlalchemy_config)],
)

第2章原代码

来源：Litestar官方第2章，©2025 Litestar-Org，MIT；代码未修改。

示例 1

from rich import get_console
from sqlalchemy.orm import Mapped

from litestar.plugins.sqlalchemy import base, repository

console = get_console()


# the SQLAlchemy base includes a declarative model for you to use in your models.
# The `Base` class includes a `UUID` based primary key (`id`)
class Author(base.UUIDBase):
    name: Mapped[str]
    dob: Mapped[date]
    dod: Mapped[date | None]


class AuthorRepository(repository.SQLAlchemyAsyncRepository[Author]):
    """Author repository."""

    model_type = Author

示例 2

# let's make a simple context manager as an example here.
@asynccontextmanager
async def repository_factory() -> AsyncIterator[AuthorRepository]:
    async with session_factory() as db_session:
        try:
            yield AuthorRepository(session=db_session)
        except Exception:  # noqa: BLE001
            await db_session.rollback()
        else:

示例 3

    async with repository_factory() as repo:
        obj = await repo.add(
            Author(
                name="F. Scott Fitzgerald",
                dob=datetime.strptime("1896-09-24", "%Y-%m-%d").date(),
            )
        )
        console.print(f"Created Author record for {obj.name} with primary key {obj.id}.")
        return obj

示例 4

    async with repository_factory() as repo:
        obj = await repo.update(obj)
        console.print(f"Updated Author record for {obj.name} with primary key {obj.id}.")
        return obj

示例 5

    async with repository_factory() as repo:
        obj = await repo.delete(id)
        console.print(f"Deleted Author record for {obj.name} with primary key {obj.id}.")
        return obj

示例 6

import json
from pathlib import Path
from typing import Any


def open_fixture(fixtures_path: Path, fixture_name: str) -> Any:
    """Loads JSON file with the specified fixture name

    Args:
        fixtures_path (Path): The path to look for fixtures
        fixture_name (str): The fixture name to load.

    Raises:
        FileNotFoundError: Fixtures not found.

    Returns:
        Any: The parsed JSON data
    """
    fixture = Path(fixtures_path / f"{fixture_name}.json")
    if fixture.exists():
        with fixture.open(mode="r", encoding="utf-8") as f:
            f_data = f.read()
        return json.loads(f_data)
    raise FileNotFoundError(f"Could not find the {fixture_name} fixture")

示例 7

[
  {
    "name": "Alabama",
    "abbreviation": "AL"
  },
  {
    "name": "Alaska",
    "abbreviation": "AK"
  },
  {
    "name": "Arizona",
    "abbreviation": "AZ"
  },
  {
    "name": "Arkansas",
    "abbreviation": "AR"
  },
  {
    "name": "California",
    "abbreviation": "CA"
  },
  {
    "name": "Colorado",
    "abbreviation": "CO"
  },
  {
    "name": "Connecticut",
    "abbreviation": "CT"
  },
  {
    "name": "Delaware",
    "abbreviation": "DE"
  },
  {
    "name": "District Of Columbia",
    "abbreviation": "DC"
  },
  {
    "name": "Florida",
    "abbreviation": "FL"
  },
  {
    "name": "Georgia",
    "abbreviation": "GA"
  },
  {
    "name": "Guam",
    "abbreviation": "GU"
  },
  {
    "name": "Hawaii",
    "abbreviation": "HI"
  },
  {
    "name": "Idaho",
    "abbreviation": "ID"
  },
  {
    "name": "Illinois",
    "abbreviation": "IL"
  },
  {
    "name": "Indiana",
    "abbreviation": "IN"
  },
  {
    "name": "Iowa",
    "abbreviation": "IA"
  },
  {
    "name": "Kansas",
    "abbreviation": "KS"
  },
  {
    "name": "Kentucky",
    "abbreviation": "KY"
  },
  {
    "name": "Louisiana",
    "abbreviation": "LA"
  },
  {
    "name": "Maine",
    "abbreviation": "ME"
  },
  {
    "name": "Maryland",
    "abbreviation": "MD"
  },
  {
    "name": "Massachusetts",
    "abbreviation": "MA"
  },
  {
    "name": "Michigan",
    "abbreviation": "MI"
  },
  {
    "name": "Minnesota",
    "abbreviation": "MN"
  },
  {
    "name": "Mississippi",
    "abbreviation": "MS"
  },
  {
    "name": "Missouri",
    "abbreviation": "MO"
  },
  {
    "name": "Montana",
    "abbreviation": "MT"
  },
  {
    "name": "Nebraska",
    "abbreviation": "NE"
  },
  {
    "name": "Nevada",
    "abbreviation": "NV"
  },
  {
    "name": "New Hampshire",
    "abbreviation": "NH"
  },
  {
    "name": "New Jersey",
    "abbreviation": "NJ"
  },
  {
    "name": "New Mexico",
    "abbreviation": "NM"
  },
  {
    "name": "New York",
    "abbreviation": "NY"
  },
  {
    "name": "North Carolina",
    "abbreviation": "NC"
  },
  {
    "name": "North Dakota",
    "abbreviation": "ND"
  },
  {
    "name": "Ohio",
    "abbreviation": "OH"
  },
  {
    "name": "Oklahoma",
    "abbreviation": "OK"
  },
  {
    "name": "Oregon",
    "abbreviation": "OR"
  },
  {
    "name": "Palau",
    "abbreviation": "PW"
  },
  {
    "name": "Pennsylvania",
    "abbreviation": "PA"
  },
  {
    "name": "Puerto Rico",
    "abbreviation": "PR"
  },
  {
    "name": "Rhode Island",
    "abbreviation": "RI"
  },
  {
    "name": "South Carolina",
    "abbreviation": "SC"
  },
  {
    "name": "South Dakota",
    "abbreviation": "SD"
  },
  {
    "name": "Tennessee",
    "abbreviation": "TN"
  },
  {
    "name": "Texas",
    "abbreviation": "TX"
  },
  {
    "name": "Utah",
    "abbreviation": "UT"
  },
  {
    "name": "Vermont",
    "abbreviation": "VT"
  },
  {
    "name": "Virginia",
    "abbreviation": "VA"
  },
  {
    "name": "Washington",
    "abbreviation": "WA"
  },
  {
    "name": "West Virginia",
    "abbreviation": "WV"
  },
  {
    "name": "Wisconsin",
    "abbreviation": "WI"
  },
  {
    "name": "Wyoming",
    "abbreviation": "WY"
  }
]

示例 8

from rich import get_console
from sqlalchemy import create_engine
from sqlalchemy.orm import Mapped, Session, sessionmaker

from litestar.plugins.sqlalchemy import base, repository

console = get_console()


class USState(base.UUIDBase):
    __tablename__ = "us_state_lookup"  # type: ignore[assignment]
    abbreviation: Mapped[str]
    name: Mapped[str]


class USStateRepository(repository.SQLAlchemySyncRepository[USState]):
    """US State repository."""

    model_type = USState


engine = create_engine(
    "duckdb:///:memory:",
    future=True,
)
session_factory: sessionmaker[Session] = sessionmaker(engine, expire_on_commit=False)


def run_script() -> None:
    """Load data from a fixture."""

    # Initializes the database.
    with engine.begin() as conn:
        USState.metadata.create_all(conn)

    with session_factory() as db_session:
        # 1) Load the JSON data into the US States table.
        repo = USStateRepository(session=db_session)
        fixture = open_fixture(here, USStateRepository.model_type.__tablename__)  # type: ignore
        objs = repo.add_many([USStateRepository.model_type(**raw_obj) for raw_obj in fixture])
        db_session.commit()
        console.print(f"Created {len(objs)} new objects.")

示例 9

from litestar.repository.filters import LimitOffset


        # 2) Select paginated data and total row count.
        created_objs, total_objs = repo.list_and_count(LimitOffset(limit=10, offset=0))
        console.print(f"Selected {len(created_objs)} records out of a total of {total_objs}.")

示例 10

        # 3) Let's remove the batch of records selected.
        deleted_objs = repo.delete_many([new_obj.id for new_obj in created_objs])
        console.print(f"Removed {len(deleted_objs)} records out of a total of {total_objs}.")

示例 11

        # 4) Let's count the remaining rows
        remaining_count = repo.count()
        console.print(f"Found {remaining_count} remaining records after delete.")

示例 12

from __future__ import annotations

from contextlib import asynccontextmanager
from datetime import date, datetime
from typing import AsyncIterator
from uuid import UUID

import anyio
from rich import get_console
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import Mapped

from litestar.plugins.sqlalchemy import base, repository

console = get_console()


# the SQLAlchemy base includes a declarative model for you to use in your models.
# The `Base` class includes a `UUID` based primary key (`id`)
class Author(base.UUIDBase):
    name: Mapped[str]
    dob: Mapped[date]
    dod: Mapped[date | None]


class AuthorRepository(repository.SQLAlchemyAsyncRepository[Author]):
    """Author repository."""

    model_type = Author


engine = create_async_engine(
    "sqlite+aiosqlite:///test.sqlite",
    future=True,
)
session_factory = async_sessionmaker(engine, expire_on_commit=False)


# let's make a simple context manager as an example here.
@asynccontextmanager
async def repository_factory() -> AsyncIterator[AuthorRepository]:
    async with session_factory() as db_session:
        try:
            yield AuthorRepository(session=db_session)
        except Exception:  # noqa: BLE001
            await db_session.rollback()
        else:
            await db_session.commit()


async def create_author() -> Author:
    async with repository_factory() as repo:
        obj = await repo.add(
            Author(
                name="F. Scott Fitzgerald",
                dob=datetime.strptime("1896-09-24", "%Y-%m-%d").date(),
            )
        )
        console.print(f"Created Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def update_author(obj: Author) -> Author:
    async with repository_factory() as repo:
        obj = await repo.update(obj)
        console.print(f"Updated Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def remove_author(id: UUID) -> Author:
    async with repository_factory() as repo:
        obj = await repo.delete(id)
        console.print(f"Deleted Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def get_author_if_exists(id: UUID) -> Author | None:
    async with repository_factory() as repo:
        obj = await repo.get_one_or_none(id=id)
        if obj is not None:
            console.print(f"Found Author record for {obj.name} with primary key {obj.id}.")
        else:
            console.print(f"Could not find Author with primary key {id}.")
        return obj


async def run_script() -> None:
    """Load data from a fixture."""
    async with engine.begin() as conn:
        await conn.run_sync(base.UUIDBase.metadata.create_all)

    # 1) create a new Author record.
    console.print("1) Adding a new record")
    author = await create_author()
    author_id = author.id

    # 2) Let's update the Author record.
    console.print("2) Updating a record.")
    author.dod = datetime.strptime("1940-12-21", "%Y-%m-%d").date()
    await update_author(author)

    # 3) Let's delete the record we just created.
    console.print("3) Removing a record.")
    await remove_author(author_id)

    # 4) Let's verify the record no longer exists.
    console.print("4) Select one or none.")
    _should_be_none = await get_author_if_exists(author_id)


if __name__ == "__main__":
    anyio.run(run_script)

示例 13

from __future__ import annotations

from contextlib import asynccontextmanager
from datetime import date, datetime
from collections.abc import AsyncIterator
from uuid import UUID

import anyio
from rich import get_console
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import Mapped

from litestar.plugins.sqlalchemy import base, repository

console = get_console()


# the SQLAlchemy base includes a declarative model for you to use in your models.
# The `Base` class includes a `UUID` based primary key (`id`)
class Author(base.UUIDBase):
    name: Mapped[str]
    dob: Mapped[date]
    dod: Mapped[date | None]


class AuthorRepository(repository.SQLAlchemyAsyncRepository[Author]):
    """Author repository."""

    model_type = Author


engine = create_async_engine(
    "sqlite+aiosqlite:///test.sqlite",
    future=True,
)
session_factory = async_sessionmaker(engine, expire_on_commit=False)


# let's make a simple context manager as an example here.
@asynccontextmanager
async def repository_factory() -> AsyncIterator[AuthorRepository]:
    async with session_factory() as db_session:
        try:
            yield AuthorRepository(session=db_session)
        except Exception:  # noqa: BLE001
            await db_session.rollback()
        else:
            await db_session.commit()


async def create_author() -> Author:
    async with repository_factory() as repo:
        obj = await repo.add(
            Author(
                name="F. Scott Fitzgerald",
                dob=datetime.strptime("1896-09-24", "%Y-%m-%d").date(),
            )
        )
        console.print(f"Created Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def update_author(obj: Author) -> Author:
    async with repository_factory() as repo:
        obj = await repo.update(obj)
        console.print(f"Updated Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def remove_author(id: UUID) -> Author:
    async with repository_factory() as repo:
        obj = await repo.delete(id)
        console.print(f"Deleted Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def get_author_if_exists(id: UUID) -> Author | None:
    async with repository_factory() as repo:
        obj = await repo.get_one_or_none(id=id)
        if obj is not None:
            console.print(f"Found Author record for {obj.name} with primary key {obj.id}.")
        else:
            console.print(f"Could not find Author with primary key {id}.")
        return obj


async def run_script() -> None:
    """Load data from a fixture."""
    async with engine.begin() as conn:
        await conn.run_sync(base.UUIDBase.metadata.create_all)

    # 1) create a new Author record.
    console.print("1) Adding a new record")
    author = await create_author()
    author_id = author.id

    # 2) Let's update the Author record.
    console.print("2) Updating a record.")
    author.dod = datetime.strptime("1940-12-21", "%Y-%m-%d").date()
    await update_author(author)

    # 3) Let's delete the record we just created.
    console.print("3) Removing a record.")
    await remove_author(author_id)

    # 4) Let's verify the record no longer exists.
    console.print("4) Select one or none.")
    _should_be_none = await get_author_if_exists(author_id)


if __name__ == "__main__":
    anyio.run(run_script)

第3章原代码

来源：Litestar官方第3章，©2025 Litestar-Org，MIT；代码未修改。

示例 1

async def provide_authors_repo(db_session: AsyncSession) -> AuthorRepository:
    """This provides the default Authors repository."""
    return AuthorRepository(session=db_session)

示例 2

# we can optionally override the default `select` used for the repository to pass in
# specific SQL options such as join details
async def provide_author_details_repo(db_session: AsyncSession) -> AuthorRepository:
    """This provides a simple example demonstrating how to override the join options for the repository."""
    return AuthorRepository(
        statement=select(AuthorModel).options(selectinload(AuthorModel.books)),
        session=db_session,
    )

示例 3

    @get(path="/authors")
    async def list_authors(
        self,
        authors_repo: AuthorRepository,
        limit_offset: filters.LimitOffset,
    ) -> OffsetPagination[Author]:
        """List authors."""
        results, total = await authors_repo.list_and_count(limit_offset)
        type_adapter = TypeAdapter(list[Author])
        return OffsetPagination[Author](
            items=type_adapter.validate_python(results),
            total=total,
            limit=limit_offset.limit,
            offset=limit_offset.offset,
        )

    @post(path="/authors")
    async def create_author(
        self,
        authors_repo: AuthorRepository,
        data: AuthorCreate,
    ) -> Author:
        """Create a new author."""
        obj = await authors_repo.add(
            AuthorModel(**data.model_dump(exclude_unset=True, exclude_none=True)),
        )
        await authors_repo.session.commit()
        return Author.model_validate(obj)

    # we override the authors_repo to use the version that joins the Books in
    @get(path="/authors/{author_id:uuid}", dependencies={"authors_repo": Provide(provide_author_details_repo)})
    async def get_author(
        self,
        authors_repo: AuthorRepository,
        author_id: Annotated[
            UUID,
            PathParameter(
                title="Author ID",
                description="The author to retrieve.",
            ),
        ],
    ) -> Author:
        """Get an existing author."""
        obj = await authors_repo.get(author_id)
        return Author.model_validate(obj)

    @patch(
        path="/authors/{author_id:uuid}",
        dependencies={"authors_repo": Provide(provide_author_details_repo)},
    )
    async def update_author(
        self,
        authors_repo: AuthorRepository,
        data: AuthorUpdate,
        author_id: Annotated[
            UUID,
            PathParameter(
                title="Author ID",
                description="The author to update.",
            ),
        ],
    ) -> Author:
        """Update an author."""
        raw_obj = data.model_dump(exclude_unset=True, exclude_none=True)
        raw_obj.update({"id": author_id})
        obj = await authors_repo.update(AuthorModel(**raw_obj))
        await authors_repo.session.commit()
        return Author.from_orm(obj)

    @delete(path="/authors/{author_id:uuid}")
    async def delete_author(
        self,
        authors_repo: AuthorRepository,
        author_id: Annotated[
            UUID,
            PathParameter(
                title="Author ID",
                description="The author to delete.",
            ),
        ],

示例 4

from __future__ import annotations

from datetime import date
from typing import TYPE_CHECKING, Annotated
from uuid import UUID

from pydantic import BaseModel as _BaseModel
from pydantic import TypeAdapter
from sqlalchemy import ForeignKey, select
from sqlalchemy.orm import Mapped, mapped_column, relationship, selectinload

from litestar import Litestar, get
from litestar.controller import Controller
from litestar.di import Provide
from litestar.handlers.http_handlers.decorators import delete, patch, post
from litestar.pagination import OffsetPagination
from litestar.params import PathParameter, QueryParameter
from litestar.plugins.sqlalchemy import (
    SQLAlchemyInitPlugin,
    SQLAlchemySyncConfig,
    base,
    repository,
)
from litestar.repository.filters import LimitOffset

if TYPE_CHECKING:
    from sqlalchemy.orm import Session


class BaseModel(_BaseModel):
    """Extend Pydantic's BaseModel to enable ORM mode"""

    model_config = {"from_attributes": True}


# The SQLAlchemy base includes a declarative model for you to use in your models.
# The `UUIDBase` class includes a `UUID` based primary key (`id`)
class AuthorModel(base.UUIDBase):
    # we can optionally provide the table name instead of auto-generating it
    __tablename__ = "author"  #  type: ignore[assignment]
    name: Mapped[str]
    dob: Mapped[date | None]
    books: Mapped[list[BookModel]] = relationship(back_populates="author", lazy="noload")


# The `UUIDAuditBase` class includes the same UUID` based primary key (`id`) and 2
# additional columns: `created_at` and `updated_at`. `created_at` is a timestamp of when the
# record created, and `updated_at` is the last time the record was modified.
class BookModel(base.UUIDAuditBase):
    __tablename__ = "book"  #  type: ignore[assignment]
    title: Mapped[str]
    author_id: Mapped[UUID] = mapped_column(ForeignKey("author.id"))
    author: Mapped[AuthorModel] = relationship(lazy="joined", innerjoin=True, viewonly=True)


# we will explicitly define the schema instead of using DTO objects for clarity.


class Author(BaseModel):
    id: UUID | None
    name: str
    dob: date | None = None


class AuthorCreate(BaseModel):
    name: str
    dob: date | None = None


class AuthorUpdate(BaseModel):
    name: str | None = None
    dob: date | None = None


class AuthorRepository(repository.SQLAlchemySyncRepository[AuthorModel]):
    """Author repository."""

    model_type = AuthorModel


async def provide_authors_repo(db_session: Session) -> AuthorRepository:
    """This provides the default Authors repository."""
    return AuthorRepository(session=db_session)


# we can optionally override the default `select` used for the repository to pass in
# specific SQL options such as join details
async def provide_author_details_repo(db_session: Session) -> AuthorRepository:
    """This provides a simple example demonstrating how to override the join options
    for the repository."""
    return AuthorRepository(
        statement=select(AuthorModel).options(selectinload(AuthorModel.books)),
        session=db_session,
    )


def provide_limit_offset_pagination(
    current_page: Annotated[int, QueryParameter(name="currentPage", ge=1, required=False)] = 1,
    page_size: Annotated[int, QueryParameter(name="pageSize", ge=1, required=False)] = 10,
) -> LimitOffset:
    """Add offset/limit pagination.

    Return type consumed by `Repository.apply_limit_offset_pagination()`.

    Parameters
    ----------
    current_page : int
        LIMIT to apply to select.
    page_size : int
        OFFSET to apply to select.
    """
    return LimitOffset(page_size, page_size * (current_page - 1))


class AuthorController(Controller):
    """Author CRUD"""

    dependencies = {"authors_repo": Provide(provide_authors_repo, sync_to_thread=False)}

    @get(path="/authors")
    def list_authors(
        self,
        authors_repo: AuthorRepository,
        limit_offset: LimitOffset,
    ) -> OffsetPagination[Author]:
        """List authors."""
        results, total = authors_repo.list_and_count(limit_offset)
        type_adapter = TypeAdapter(list[Author])
        return OffsetPagination[Author](
            items=type_adapter.validate_python(results),
            total=total,
            limit=limit_offset.limit,
            offset=limit_offset.offset,
        )

    @post(path="/authors")
    def create_author(
        self,
        authors_repo: AuthorRepository,
        data: AuthorCreate,
    ) -> Author:
        """Create a new author."""
        obj = authors_repo.add(
            AuthorModel(**data.model_dump(exclude_unset=True, exclude_none=True)),
        )
        authors_repo.session.commit()
        return Author.model_validate(obj)

    # we override the authors_repo to use the version that joins the Books in
    @get(
        path="/authors/{author_id:uuid}",
        dependencies={"authors_repo": Provide(provide_author_details_repo, sync_to_thread=False)},
    )
    def get_author(
        self,
        authors_repo: AuthorRepository,
        author_id: Annotated[
            UUID,
            PathParameter(
                title="Author ID",
                description="The author to retrieve.",
            ),
        ],
    ) -> Author:
        """Get an existing author."""
        obj = authors_repo.get(author_id)
        return Author.model_validate(obj)

    @patch(
        path="/authors/{author_id:uuid}",
        dependencies={"authors_repo": Provide(provide_author_details_repo, sync_to_thread=False)},
    )
    def update_author(
        self,
        authors_repo: AuthorRepository,
        data: AuthorUpdate,
        author_id: Annotated[
            UUID,
            PathParameter(
                title="Author ID",
                description="The author to update.",
            ),
        ],
    ) -> Author:
        """Update an author."""
        raw_obj = data.model_dump(exclude_unset=True, exclude_none=True)
        raw_obj.update({"id": author_id})
        obj = authors_repo.update(AuthorModel(**raw_obj))
        authors_repo.session.commit()
        return Author.model_validate(obj)

    @delete(path="/authors/{author_id:uuid}")
    def delete_author(
        self,
        authors_repo: AuthorRepository,
        author_id: Annotated[
            UUID,
            PathParameter(
                title="Author ID",
                description="The author to delete.",
            ),
        ],
    ) -> None:
        """Delete a author from the system."""
        _ = authors_repo.delete(author_id)
        authors_repo.session.commit()


sqlalchemy_config = SQLAlchemySyncConfig(connection_string="sqlite:///test.sqlite")  # Create 'db_session' dependency.
sqlalchemy_plugin = SQLAlchemyInitPlugin(config=sqlalchemy_config)


def on_startup() -> None:
    """Initializes the database."""
    with sqlalchemy_config.get_engine().begin() as conn:
        base.UUIDBase.metadata.create_all(conn)


app = Litestar(
    route_handlers=[AuthorController],
    on_startup=[on_startup],
    plugins=[SQLAlchemyInitPlugin(config=sqlalchemy_config)],
    dependencies={"limit_offset": Provide(provide_limit_offset_pagination)},
)

示例 5

from __future__ import annotations

from datetime import date
from typing import TYPE_CHECKING, Annotated
from uuid import UUID

from pydantic import BaseModel as _BaseModel
from pydantic import TypeAdapter
from sqlalchemy import ForeignKey, select
from sqlalchemy.orm import Mapped, mapped_column, relationship, selectinload

from litestar import Litestar, get
from litestar.controller import Controller
from litestar.di import Provide
from litestar.handlers.http_handlers.decorators import delete, patch, post
from litestar.pagination import OffsetPagination
from litestar.params import PathParameter, QueryParameter
from litestar.plugins.sqlalchemy import (
    AsyncSessionConfig,
    SQLAlchemyAsyncConfig,
    SQLAlchemyInitPlugin,
    base,
    filters,
    repository,
)

if TYPE_CHECKING:
    from sqlalchemy.ext.asyncio import AsyncSession


class BaseModel(_BaseModel):
    """Extend Pydantic's BaseModel to enable ORM mode"""

    model_config = {"from_attributes": True}


# The SQLAlchemy base includes a declarative model for you to use in your models.
# The `UUIDBase` class includes a `UUID` based primary key (`id`)
class AuthorModel(base.UUIDBase):
    # we can optionally provide the table name instead of auto-generating it
    __tablename__ = "author"  #  type: ignore[assignment]
    name: Mapped[str]
    dob: Mapped[date | None]
    books: Mapped[list[BookModel]] = relationship(back_populates="author", lazy="noload")


# The `UUIDAuditBase` class includes the same UUID` based primary key (`id`) and 2
# additional columns: `created_at` and `updated_at`. `created_at` is a timestamp of when the
# record created, and `updated_at` is the last time the record was modified.
class BookModel(base.UUIDAuditBase):
    __tablename__ = "book"  #  type: ignore[assignment]
    title: Mapped[str]
    author_id: Mapped[UUID] = mapped_column(ForeignKey("author.id"))
    author: Mapped[AuthorModel] = relationship(lazy="joined", innerjoin=True, viewonly=True)


# we will explicitly define the schema instead of using DTO objects for clarity.


class Author(BaseModel):
    id: UUID | None
    name: str
    dob: date | None = None


class AuthorCreate(BaseModel):
    name: str
    dob: date | None = None


class AuthorUpdate(BaseModel):
    name: str | None = None
    dob: date | None = None


class AuthorRepository(repository.SQLAlchemyAsyncRepository[AuthorModel]):
    """Author repository."""

    model_type = AuthorModel


async def provide_authors_repo(db_session: AsyncSession) -> AuthorRepository:
    """This provides the default Authors repository."""
    return AuthorRepository(session=db_session)


# we can optionally override the default `select` used for the repository to pass in
# specific SQL options such as join details
async def provide_author_details_repo(db_session: AsyncSession) -> AuthorRepository:
    """This provides a simple example demonstrating how to override the join options for the repository."""
    return AuthorRepository(
        statement=select(AuthorModel).options(selectinload(AuthorModel.books)),
        session=db_session,
    )


def provide_limit_offset_pagination(
    current_page: Annotated[int, QueryParameter(name="currentPage", ge=1, required=False)] = 1,
    page_size: Annotated[int, QueryParameter(name="pageSize", ge=1, required=False)] = 10,
) -> filters.LimitOffset:
    """Add offset/limit pagination.

    Return type consumed by `Repository.apply_limit_offset_pagination()`.

    Parameters
    ----------
    current_page : int
        LIMIT to apply to select.
    page_size : int
        OFFSET to apply to select.
    """
    return filters.LimitOffset(page_size, page_size * (current_page - 1))


class AuthorController(Controller):
    """Author CRUD"""

    dependencies = {"authors_repo": Provide(provide_authors_repo)}

    @get(path="/authors")
    async def list_authors(
        self,
        authors_repo: AuthorRepository,
        limit_offset: filters.LimitOffset,
    ) -> OffsetPagination[Author]:
        """List authors."""
        results, total = await authors_repo.list_and_count(limit_offset)
        type_adapter = TypeAdapter(list[Author])
        return OffsetPagination[Author](
            items=type_adapter.validate_python(results),
            total=total,
            limit=limit_offset.limit,
            offset=limit_offset.offset,
        )

    @post(path="/authors")
    async def create_author(
        self,
        authors_repo: AuthorRepository,
        data: AuthorCreate,
    ) -> Author:
        """Create a new author."""
        obj = await authors_repo.add(
            AuthorModel(**data.model_dump(exclude_unset=True, exclude_none=True)),
        )
        await authors_repo.session.commit()
        return Author.model_validate(obj)

    # we override the authors_repo to use the version that joins the Books in
    @get(path="/authors/{author_id:uuid}", dependencies={"authors_repo": Provide(provide_author_details_repo)})
    async def get_author(
        self,
        authors_repo: AuthorRepository,
        author_id: Annotated[
            UUID,
            PathParameter(
                title="Author ID",
                description="The author to retrieve.",
            ),
        ],
    ) -> Author:
        """Get an existing author."""
        obj = await authors_repo.get(author_id)
        return Author.model_validate(obj)

    @patch(
        path="/authors/{author_id:uuid}",
        dependencies={"authors_repo": Provide(provide_author_details_repo)},
    )
    async def update_author(
        self,
        authors_repo: AuthorRepository,
        data: AuthorUpdate,
        author_id: Annotated[
            UUID,
            PathParameter(
                title="Author ID",
                description="The author to update.",
            ),
        ],
    ) -> Author:
        """Update an author."""
        raw_obj = data.model_dump(exclude_unset=True, exclude_none=True)
        raw_obj.update({"id": author_id})
        obj = await authors_repo.update(AuthorModel(**raw_obj))
        await authors_repo.session.commit()
        return Author.from_orm(obj)

    @delete(path="/authors/{author_id:uuid}")
    async def delete_author(
        self,
        authors_repo: AuthorRepository,
        author_id: Annotated[
            UUID,
            PathParameter(
                title="Author ID",
                description="The author to delete.",
            ),
        ],
    ) -> None:
        """Delete a author from the system."""
        _ = await authors_repo.delete(author_id)
        await authors_repo.session.commit()


session_config = AsyncSessionConfig(expire_on_commit=False)
sqlalchemy_config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///test.sqlite", session_config=session_config
)  # Create 'db_session' dependency.
sqlalchemy_plugin = SQLAlchemyInitPlugin(config=sqlalchemy_config)


async def on_startup() -> None:
    """Initializes the database."""
    async with sqlalchemy_config.get_engine().begin() as conn:
        await conn.run_sync(base.UUIDBase.metadata.create_all)


app = Litestar(
    route_handlers=[AuthorController],
    on_startup=[on_startup],
    plugins=[SQLAlchemyInitPlugin(config=sqlalchemy_config)],
    dependencies={"limit_offset": Provide(provide_limit_offset_pagination)},
)

第4章原代码

来源：Litestar官方第4章，©2025 Litestar-Org，MIT；代码未修改。

示例 1

from sqlalchemy.orm import Mapped, declarative_mixin, mapped_column
from sqlalchemy.types import String

from litestar.plugins.sqlalchemy import (
    base,
)


@declarative_mixin
class SlugKey:
    """Slug unique Field Model Mixin."""

    __abstract__ = True
    slug: Mapped[str] = mapped_column(String(length=100), nullable=False, unique=True, sort_order=-9)


class BlogPost(base.UUIDAuditBase, SlugKey):
    title: Mapped[str]
    content: Mapped[str]

示例 2

from __future__ import annotations

import random
import re
import string
import unicodedata
from typing import Any

from litestar.plugins.sqlalchemy import (
    repository,
)


class SQLAlchemyAsyncSlugRepository(repository.SQLAlchemyAsyncRepository[repository.ModelT]):
    """Extends the repository to include slug model features.."""

    async def get_available_slug(
        self,
        value_to_slugify: str,
        **kwargs: Any,
    ) -> str:
        """Get a unique slug for the supplied value.

        If the value is found to exist, a random 4 digit character is appended to the end.
        There may be a better way to do this, but I wanted to limit the number of
        additional database calls.

        Args:
            value_to_slugify (str): A string that should be converted to a unique slug.
            **kwargs: stuff

        Returns:
            str: a unique slug for the supplied value. This is safe for URLs and other
            unique identifiers.
        """
        slug = self._slugify(value_to_slugify)
        if await self._is_slug_unique(slug):
            return slug
        # generate a random 4 digit alphanumeric string to make the slug unique and
        # avoid another DB lookup.
        random_string = "".join(random.choices(string.ascii_lowercase + string.digits, k=4))
        return f"{slug}-{random_string}"

    @staticmethod
    def _slugify(value: str) -> str:
        """slugify.

        Convert to ASCII if 'allow_unicode' is False. Convert spaces or repeated
        dashes to single dashes. Remove characters that aren't alphanumerics,
        underscores, or hyphens. Convert to lowercase. Also strip leading and
        trailing whitespace, dashes, and underscores.

        Args:
            value (str): the string to slugify

        Returns:
            str: a slugified string of the value parameter
        """
        value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
        value = re.sub(r"[^\w\s-]", "", value.lower())
        return re.sub(r"[-\s]+", "-", value).strip("-_")

    async def _is_slug_unique(
        self,
        slug: str,
        **kwargs: Any,
    ) -> bool:
        return await self.get_one_or_none(slug=slug) is None

示例 3

from __future__ import annotations

from uuid import UUID

from pydantic import BaseModel as _BaseModel
from sqlalchemy.orm import Mapped

from litestar import post
from litestar.plugins.sqlalchemy import (
    base,
)


class BaseModel(_BaseModel):
    """Extend Pydantic's BaseModel to enable ORM mode"""

    model_config = {"from_attributes": True}


class BlogPost(base.UUIDAuditBase, SlugKey):
    title: Mapped[str]
    content: Mapped[str]


class BlogPostRepository(SQLAlchemyAsyncSlugRepository[BlogPost]):
    """Blog Post repository."""

    model_type = BlogPost


class BlogPostDTO(BaseModel):
    id: UUID | None
    slug: str
    title: str
    content: str


class BlogPostCreate(BaseModel):
    title: str
    content: str
@post(path="/")
async def create_blog(
    blog_post_repo: BlogPostRepository,
    data: BlogPostCreate,
) -> BlogPostDTO:
    """Create a new blog post."""
    _data = data.model_dump(exclude_unset=True, by_alias=False, exclude_none=True)
    _data["slug"] = await blog_post_repo.get_available_slug(_data["title"])
    obj = await blog_post_repo.add(BlogPost(**_data))
    await blog_post_repo.session.commit()
    return BlogPostDTO.model_validate(obj)

示例 4

from __future__ import annotations

import random
import re
import string
import unicodedata
from typing import TYPE_CHECKING, Any
from uuid import UUID

from pydantic import BaseModel as _BaseModel
from pydantic import TypeAdapter
from sqlalchemy.orm import Mapped, declarative_mixin, mapped_column
from sqlalchemy.types import String

from litestar import Litestar, get, post
from litestar.di import Provide
from litestar.plugins.sqlalchemy import (
    AsyncSessionConfig,
    SQLAlchemyAsyncConfig,
    SQLAlchemyInitPlugin,
    base,
    repository,
)

if TYPE_CHECKING:
    from sqlalchemy.ext.asyncio import AsyncSession


class BaseModel(_BaseModel):
    """Extend Pydantic's BaseModel to enable ORM mode"""

    model_config = {"from_attributes": True}


# we are going to add a simple "slug" to our model that is a URL safe surrogate key to
# our database record.
@declarative_mixin
class SlugKey:
    """Slug unique Field Model Mixin."""

    __abstract__ = True
    slug: Mapped[str] = mapped_column(String(length=100), nullable=False, unique=True, sort_order=-9)


# this class can be re-used with any model that has the `SlugKey` Mixin
class SQLAlchemyAsyncSlugRepository(repository.SQLAlchemyAsyncRepository[repository.ModelT]):
    """Extends the repository to include slug model features.."""

    async def get_available_slug(
        self,
        value_to_slugify: str,
        **kwargs: Any,
    ) -> str:
        """Get a unique slug for the supplied value.

        If the value is found to exist, a random 4 digit character is appended to the end.
        There may be a better way to do this, but I wanted to limit the number of
        additional database calls.

        Args:
            value_to_slugify (str): A string that should be converted to a unique slug.
            **kwargs: stuff

        Returns:
            str: a unique slug for the supplied value. This is safe for URLs and other
            unique identifiers.
        """
        slug = self._slugify(value_to_slugify)
        if await self._is_slug_unique(slug):
            return slug
        # generate a random 4 digit alphanumeric string to make the slug unique and
        # avoid another DB lookup.
        random_string = "".join(random.choices(string.ascii_lowercase + string.digits, k=4))
        return f"{slug}-{random_string}"

    @staticmethod
    def _slugify(value: str) -> str:
        """slugify.

        Convert to ASCII if 'allow_unicode' is False. Convert spaces or repeated
        dashes to single dashes. Remove characters that aren't alphanumerics,
        underscores, or hyphens. Convert to lowercase. Also strip leading and
        trailing whitespace, dashes, and underscores.

        Args:
            value (str): the string to slugify

        Returns:
            str: a slugified string of the value parameter
        """
        value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
        value = re.sub(r"[^\w\s-]", "", value.lower())
        return re.sub(r"[-\s]+", "-", value).strip("-_")

    async def _is_slug_unique(
        self,
        slug: str,
        **kwargs: Any,
    ) -> bool:
        return await self.get_one_or_none(slug=slug) is None


# The `UUIDAuditBase` class includes the same UUID` based primary key (`id`) and 2
# additional columns: `created_at` and `updated_at`. `created_at` is a timestamp of when the
# record created, and `updated_at` is the last time the record was modified.
class BlogPost(base.UUIDAuditBase, SlugKey):
    title: Mapped[str]
    content: Mapped[str]


class BlogPostRepository(SQLAlchemyAsyncSlugRepository[BlogPost]):
    """Blog Post repository."""

    model_type = BlogPost


class BlogPostDTO(BaseModel):
    id: UUID | None
    slug: str
    title: str
    content: str


class BlogPostCreate(BaseModel):
    title: str
    content: str


# we can optionally override the default `select` used for the repository to pass in
# specific SQL options such as join details
async def provide_blog_post_repo(db_session: AsyncSession) -> BlogPostRepository:
    """This provides a simple example demonstrating how to override the join options
    for the repository."""
    return BlogPostRepository(session=db_session)


session_config = AsyncSessionConfig(expire_on_commit=False)
sqlalchemy_config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///test.sqlite", session_config=session_config
)  # Create 'async_session' dependency.
sqlalchemy_plugin = SQLAlchemyInitPlugin(config=sqlalchemy_config)


async def on_startup() -> None:
    """Initializes the database."""
    async with sqlalchemy_config.get_engine().begin() as conn:
        await conn.run_sync(base.UUIDAuditBase.metadata.create_all)


@get(path="/")
async def get_blogs(
    blog_post_repo: BlogPostRepository,
) -> list[BlogPostDTO]:
    """Interact with SQLAlchemy engine and session."""
    objs = await blog_post_repo.list()
    type_adapter = TypeAdapter(list[BlogPostDTO])
    return type_adapter.validate_python(objs)


@get(path="/{post_slug:str}")
async def get_blog_details(
    post_slug: str,
    blog_post_repo: BlogPostRepository,
) -> BlogPostDTO:
    """Interact with SQLAlchemy engine and session."""
    obj = await blog_post_repo.get_one(slug=post_slug)
    return BlogPostDTO.model_validate(obj)


@post(path="/")
async def create_blog(
    blog_post_repo: BlogPostRepository,
    data: BlogPostCreate,
) -> BlogPostDTO:
    """Create a new blog post."""
    _data = data.model_dump(exclude_unset=True, by_alias=False, exclude_none=True)
    _data["slug"] = await blog_post_repo.get_available_slug(_data["title"])
    obj = await blog_post_repo.add(BlogPost(**_data))
    await blog_post_repo.session.commit()
    return BlogPostDTO.model_validate(obj)


app = Litestar(
    route_handlers=[create_blog, get_blogs, get_blog_details],
    dependencies={"blog_post_repo": Provide(provide_blog_post_repo, sync_to_thread=False)},
    on_startup=[on_startup],
    plugins=[SQLAlchemyInitPlugin(config=sqlalchemy_config)],
)