chore: add books_cli example project to demo pytest-sqlguard

.github/workflows/sqlguard-ci.yml

@@ -23,7 +23,7 @@ jobs:
         run: |
           export UV_LINK_MODE=copy
           uv sync --all-extras
-          uv run pytest
+          uv run pytest tests
 
   test-tox-sqlguard:
     runs-on: ubuntu-latest
@@ -37,7 +37,7 @@ jobs:
           uv tool install tox --with tox-uv
           echo "$HOME/.local/bin" >> $GITHUB_PATH
       - name: Run Tox suite
-        run: tox run
+        run: tox run -- tests
 
   build-sqlguard:
     needs: [test-sqlguard, test-tox-sqlguard]

examples/books_cli/README.md

@@ -0,0 +1,114 @@
+# 📚 Books CLI — Demo Project for pytest-sqlguard
+
+This sample project showcases how to use [`pytest-sqlguard`](https://pypi.org/project/pytest-sqlguard/) in a realistic application. It includes a minimal CLI interface and a FastAPI web application for managing a books and authors database.
+
+The main goal of this project is to demonstrate how `pytest-sqlguard` can help you detect and manage SQL changes in your codebase using snapshot testing.
+
+---
+
+## 🚀 Quick Start
+
+This project uses [`uv`](https://docs.astral.sh/uv/) as the environment and task manager.
+
+#### 1. Initialize the environment
+
+```bash
+uv init
+```
+
+---
+
+## 🖥️ Command-Line Interface
+
+The CLI is defined in the script [`books.py`](./books.py).
+
+📌 To view CLI options:
+
+```bash
+uv run books.py --help
+```
+
+### CLI Commands
+
+- 🏗️ Create the SQLite database (Run this at least once):
+
+  ```bash
+  uv run books.py init-db
+  ```
+
+- 🧹 Drop the database:
+
+  ```bash
+  uv run books.py drop-db
+  ```
+
+- ✍️ Manage authors (subcommand):
+
+  ```bash
+  uv run books.py authors
+  ```
+
+---
+
+## 🌐 Web API (FastAPI)
+
+You can run the FastAPI app located in [`books_cli/main.py`](./books_cli/main.py):
+
+```bash
+uv run fastapi dev books_cli/main.py
+```
+
+- API documentation is available at:
+  👉 [http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs)
+
+---
+
+## 🧪 Running Tests with pytest-sqlguard
+
+The test suite illustrates how `pytest-sqlguard` detects unexpected SQL changes via snapshot testing.
+
+Run the tests:
+
+```bash
+uv run pytest
+```
+
+Want to see `pytest-sqlguard` in action?
+
+1. Open [`books_cli/operations/search.py`](./books_cli/operations/search.py)
+2. Locate the function `list_books_by_author_name`
+3. Uncomment the version that uses a SQL JOIN
+4. Run tests again:
+
+   ```bash
+   uv run pytest
+   ```
+
+You should see SQL snapshot mismatches for both:
+
+- [`tests/test_author.py`](./tests/test_author.py)
+- [`tests/test_author_api.py`](./tests/test_author_api.py)
+
+To resolve and update the snapshots to the new queries:
+
+```bash
+uv run pytest --sqlguard-overwrite
+```
+
+This will overwrite the following snapshot files:
+
+- [`tests/test_author.queries.yaml`](./tests/test_author.queries.yaml)
+- [`tests/test_author_api.queries.yaml`](./tests/test_author_api.queries.yaml)
+
+---
+
+## 🧰 Tech Stack
+
+This project leverages:
+
+- ⚡️ [FastAPI](https://fastapi.tiangolo.com/) – Web framework
+- 🐍 [Typer](https://typer.tiangolo.com/) – CLI framework
+- 🐘 [SQLAlchemy](https://www.sqlalchemy.org/) – ORM
+- 🔎 [pytest](https://docs.pytest.org/en/stable/) – Test runner
+- 🔐 [pytest-sqlguard](https://pypi.org/project/pytest-sqlguard/) – SQL snapshot testing
+- 🚀 [uv](https://docs.astral.sh/uv/) – Python dependency and virtualenv manager

examples/books_cli/books.py

@@ -0,0 +1,59 @@
+from contextlib import contextmanager
+from decimal import Decimal
+
+import rich
+import typer
+from books_cli.model_base import SQLABase
+from books_cli.operations.author import create_author, list_authors
+from books_cli.operations.books import create_book
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+app = typer.Typer(no_args_is_help=True)
+
+engine = create_engine("sqlite:///books.db", echo=True)
+
+
+@contextmanager
+def session() -> Session:
+    with Session(autoflush=True, bind=engine) as session:
+        yield session
+        session.commit()
+
+
+@app.command()
+def init_db():
+    SQLABase.metadata.create_all(bind=engine)
+    with session() as _session:
+        tolkien = create_author(name="Tolkien", age=99, session=_session)
+        rothfuss = create_author(name="Rothfuss", age=55, session=_session)
+        create_book(title="Lord of the Rings", price=Decimal("50.4"), author_id=tolkien.id, session=_session)
+        create_book(title="Bilbo Baggins", price=Decimal("23.2"), author_id=tolkien.id, session=_session)
+        create_book(title="The Name of the Wind", price=Decimal("19.99"), author_id=rothfuss.id, session=_session)
+
+
+@app.command()
+def drop_db():
+    SQLABase.metadata.drop_all(bind=engine)
+
+
+authors = typer.Typer()
+app.add_typer(authors, name="authors")
+
+
+@authors.command(name="create")
+def create_auth(name: str, age: int):
+    with session() as _session:
+        author = create_author(name=name, age=age, session=_session)
+    rich.print("Created author:", author)
+
+
+@authors.command(name="list")
+def list_auth():
+    with session() as _session:
+        authors = list_authors(_session)
+    rich.print("Authors:", authors)
+
+
+if __name__ == "__main__":
+    app()

examples/books_cli/books_cli/api/__init__.py

@@ -0,0 +1,2 @@
+from .author import router as author_router
+from .search import router as search_router

examples/books_cli/books_cli/api/author.py

@@ -0,0 +1,17 @@
+from books_cli.db import DB
+from books_cli.operations.author import create_author, list_authors
+from books_cli.schemas.author import AuthorCreateSchema, AuthorSchema
+from fastapi import APIRouter
+
+router = APIRouter()
+
+
+@router.post("")
+def new_author(author_info: AuthorCreateSchema, session: DB) -> AuthorSchema:
+    author = create_author(author_info.name, author_info.age, session)
+    return author
+
+
+@router.get("")
+def list_all_authors(session: DB) -> list[AuthorSchema]:
+    return list_authors(session)

examples/books_cli/books_cli/api/search.py

@@ -0,0 +1,11 @@
+from books_cli.db import DB
+from books_cli.operations.search import list_books_by_author_name
+from books_cli.schemas.book import BookSchema
+from fastapi.routing import APIRouter
+
+router = APIRouter()
+
+
+@router.get("/by-author-name/{name}")
+def api_list_books_by_author_name(name: str, session: DB) -> list[BookSchema]:
+    return list_books_by_author_name(name, session)

examples/books_cli/books_cli/db.py

@@ -0,0 +1,16 @@
+from typing import Annotated
+
+from fastapi import Depends
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+engine = create_engine("sqlite:///books.db", echo=True)
+
+
+def get_session() -> Session:
+    with Session(bind=engine, autoflush=True) as session:
+        yield session
+        session.commit()  # Autocommit for demonstration purposes
+
+
+DB = Annotated[Session, Depends(get_session)]

examples/books_cli/books_cli/main.py

@@ -0,0 +1,9 @@
+from books_cli.api import author_router, search_router
+from fastapi import FastAPI
+
+app = FastAPI(
+    title="Books API",
+    version="0.9.9",
+)
+app.include_router(author_router, prefix="/author")
+app.include_router(search_router, prefix="/search")

examples/books_cli/books_cli/model_base.py

@@ -0,0 +1,33 @@
+from datetime import datetime, timezone
+from typing import Type
+
+from pydantic import BaseModel
+from sqlalchemy import Column, DateTime, func
+from sqlalchemy.orm import declarative_base
+from sqlalchemy_utils import UUIDType
+from ulid import ULID
+
+Base = declarative_base()
+
+
+class SQLABase(Base):
+    __abstract__ = True
+
+    def to_schema(self, schema: Type[BaseModel]):
+        return schema.model_validate(self, from_attributes=True)
+
+
+class DBModelBase:
+    id = Column(
+        UUIDType,
+        primary_key=True,
+        default=lambda: ULID().to_uuid(),
+        nullable=False,
+    )
+
+    created_at = Column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(timezone.utc),
+        server_default=func.now(),
+    )

examples/books_cli/books_cli/models/__init__.py

@@ -0,0 +1,2 @@
+from .author import Author
+from .book import Book

examples/books_cli/books_cli/models/author.py

@@ -0,0 +1,12 @@
+from books_cli.model_base import DBModelBase, SQLABase
+from sqlalchemy import Column, Integer, String
+from sqlalchemy.orm import Mapped, relationship
+
+
+class Author(SQLABase, DBModelBase):
+    __tablename__ = "authors"
+
+    name = Column(String, nullable=False)
+    age = Column(Integer, nullable=False, default=20)
+
+    books: Mapped[list["Book"]] = relationship(back_populates="author")  # noqa

examples/books_cli/books_cli/models/book.py

@@ -0,0 +1,17 @@
+from decimal import Decimal
+
+from books_cli.model_base import DBModelBase, SQLABase
+from sqlalchemy import Column, ForeignKey, String
+from sqlalchemy.orm import Mapped, relationship
+from sqlalchemy.sql.sqltypes import Numeric
+from sqlalchemy_utils.types.uuid import UUIDType
+
+
+class Book(SQLABase, DBModelBase):
+    __tablename__ = "books"
+
+    title = Column(String, nullable=False)
+    price = Column(Numeric(10, 2), nullable=False, default=Decimal("20.0"))
+    author_id = Column(UUIDType, ForeignKey("authors.id"), nullable=False)
+
+    author: Mapped["Author"] = relationship(back_populates="books")  # noqa

examples/books_cli/books_cli/operations/author.py

@@ -0,0 +1,15 @@
+from books_cli.models.author import Author
+from books_cli.schemas.author import AuthorSchema
+from sqlalchemy.orm.session import Session
+
+
+def create_author(name: str, age: int, session: Session) -> AuthorSchema:
+    author = Author(name=name, age=age)
+    session.add(author)
+    session.flush()
+    return author.to_schema(AuthorSchema)
+
+
+def list_authors(session: Session) -> list[AuthorSchema]:
+    authors = session.query(Author).all()
+    return [a.to_schema(AuthorSchema) for a in authors]

examples/books_cli/books_cli/operations/books.py

@@ -0,0 +1,13 @@
+from decimal import Decimal
+from uuid import UUID
+
+from books_cli.models import Book
+from books_cli.schemas.book import BookSchema
+from sqlalchemy.orm.session import Session
+
+
+def create_book(title: str, price: Decimal, author_id: UUID, session: Session) -> BookSchema:
+    book = Book(title=title, price=price, author_id=author_id)
+    session.add(book)
+    session.flush()
+    return book.to_schema(BookSchema)

examples/books_cli/books_cli/operations/search.py

@@ -0,0 +1,23 @@
+from books_cli.models.author import Author
+from books_cli.schemas.book import BookSchema
+from sqlalchemy.orm import Session
+
+
+def list_books_by_author_name(author_name: str, session: Session) -> list[BookSchema]:
+    author = session.query(Author).where(Author.name == author_name).first()
+    if author:
+        books = author.books
+        return [b.to_schema(BookSchema) for b in books]
+
+    # Instead of doing two queries like the above,
+    # the same result can be obtained by doing:
+    # ==== Uncomment =====
+    # from books_cli.models.book import Book
+
+    # books = session.query(Book).join(Author, Author.id == Book.author_id).where(Author.name == author_name).all()
+    # return [b.to_schema(BookSchema) for b in books]
+    # === Enf of Uncomment ==
+    # However if you do that, you need to update the guarded SQL present in
+    # test_author.queries.yaml["test_search_author"]
+    # You can do this with the command: pytest tests/test_author.py::test_search_author --sqlguard-overwrite
+    return []

examples/books_cli/books_cli/schemas/author.py

@@ -0,0 +1,16 @@
+from datetime import datetime
+from uuid import UUID
+
+from pydantic import BaseModel
+
+
+class AuthorCreateSchema(BaseModel):
+    name: str
+    age: int = 20
+
+
+class AuthorSchema(BaseModel):
+    id: UUID
+    created_at: datetime
+    name: str
+    age: int