A Litestar web application is being set up to manage a collection of books, using SQLAlchemy for data storage and Jinja for rendering a template.

app_dto_sqlite.py

from __future__ import annotations
from dataclasses import dataclass
from contextlib import asynccontextmanager
from litestar.exceptions import HTTPException
from litestar import Litestar, get, post, put, patch, delete
from litestar.dto import AbstractDTO, field, DataclassDTO, DTOConfig, DTOData
from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from collections.abc import AsyncGenerator
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from litestar.datastructures import State
from litestar.contrib.jinja import JinjaTemplateEngine
from litestar.response import Template
from litestar.template.config import TemplateConfig
from litestar.static_files.config import StaticFilesConfig

Base = declarative_base()

class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    author: Mapped[str]
    publisher: Mapped[str]
    genre: Mapped[str]
      
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
    engine = getattr(app.state, "engine", None)
    if engine is None:
        engine = create_async_engine("sqlite+aiosqlite:///books.db", echo=True)
        app.state.engine = engine
        async with engine.begin() as conn:
            Base.metadata.bind = engine
            await conn.run_sync(Base.metadata.create_all)
    try:
        yield
    finally:
        await engine.dispose()
        
sessionmaker = async_sessionmaker(expire_on_commit=False)

class ReadDTO(SQLAlchemyDTO[Book]):
    config = DTOConfig()
class WriteDTO(SQLAlchemyDTO[Book]):
    config = DTOConfig(exclude={"id"})
class PatchDTO(SQLAlchemyDTO[Book]):
    config = DTOConfig(exclude={"id"}, partial=True)
    
@get("/")
async def spa() -> Template:
    return Template(template_name="home.html")
  
@get(path="/books", return_dto=ReadDTO)
async def get_books(state: State, request: Request) -> list[Book]:
    async with sessionmaker(bind=state.engine) as session:
        query = select(Book)
        result = await session.execute(query)
        books = result.scalars().all()
        return books
      
@get(path="/book/{book_id:int}", return_dto=ReadDTO)
async def get_book(state: State, request: Request, book_id: int) -> Book:
    async with sessionmaker(bind=state.engine) as session:
        query = select(Book).where(Book.id==book_id)
        result = await session.execute(query)
        try:
            book = result.scalar_one()
        except:
            raise HTTPException(status_code=400, detail=f"book with id [{book_id}] not found")
        return book
      
@post(path="/book", dto=WriteDTO, return_dto=ReadDTO)
async def create_book(state: State, request: Request, data: DTOData[Book]) -> Book:
    book = data.create_instance()
    async with sessionmaker(bind=state.engine) as session, session.begin():
        session.add(book)
        return book
      
@put(path="/book/{book_id:int}", dto=WriteDTO, return_dto=ReadDTO)
async def update_book(state: State, request: Request, book_id: int, data: DTOData[Book]) -> Book:
    async with sessionmaker(bind=state.engine) as session, session.begin():
        query = select(Book).where(Book.id==book_id)
        result = await session.execute(query)
        try:
            book = result.scalar_one()
        except:
            raise HTTPException(status_code=400, detail=f"book with id [{book_id}] not found")
        data.update_instance(book)
        return book
      
@delete(path="/book/{book_id:int}")
async def delete_book(state: State, request: Request, book_id: int) -> None:
    async with sessionmaker(bind=state.engine) as session, session.begin():
        query = select(Book).where(Book.id==book_id)
        result = await session.execute(query)
        try:
            book = result.scalar_one()
        except:
            raise HTTPException(status_code=400, detail=f"book with id [{book_id}] not found")
        await session.delete(book)
        return None
      
app = Litestar(route_handlers=[
    spa,
    create_book, 
    get_book, 
    get_books, 
    update_book,
    delete_book
    ], 
    lifespan=[db_connection],
    template_config=TemplateConfig(directory=".\\templates", engine=JinjaTemplateEngine),
    static_files_config=[
        StaticFilesConfig(directories=["static"], path="/static")
    ]
)