Skip to content

Project Structure#

This project is structured as a Python package. The tree structure is as follows (with some files omitted for brevity):

├── Dockerfile  # (1)!
├── alembic.ini  # (2)!
├── docker-compose.yaml  # (3)!
├── migrations  # (4)!
│   ├── env.py  # (5)!
│   ├── script.py.mako  # (6)!
│   └── versions  # (7)!
│       └── 2023_08_20_1729-e5fccc3522ce_initial_migration.py  # (8)!
├── pyproject.toml  # (9)!
├── requirements.txt  # (10)!
├── tests  # (11)!
│   ├── api # (12)!
│   │   └── test_animals_api.py # (13)!
│   └── conftest.py # (14)!
└── zoo # (15)!
    ├── __init__.py # (16)!
    ├── __main__.py # (17)!
    ├── _version.py # (18)!
    ├── api # (19)!
    │   ├── __init__.py # (20)!
    │   └── animals.py # (21)!
    ├── app.py # (22)!
    ├── config.py # (23)!
    ├── db.py # (24)!
    ├── models # (25)!
    │   ├── __init__.py # (26)!
    │   └── animals.py # (27)!
    └── schemas # (28)!
        ├── __init__.py # (29)!
        └── animals.py # (30)!
  1. 🚢 Dockerfile for building the Docker image.
  2. ⚙ Alembic configuration file for managing database migrations.
  3. 🐳 Docker Compose configuration file for running the application.
  4. 🐍 Directory for managing database migrations.
  5. 🐍 Configuration file for Alembic migrations to interface with the application database.
  6. 📄 Template file for Alembic migrations.
  7. 🐍 Directory for Alembic migration scripts.
  8. 🐍 Initial migration script generated by Alembic. This is an example of what a migration script looks like - containing an upgrade and downgrade function.
  9. ⚙ Configuration file for Python Package and Tooling.
  10. ⚙ Python dependencies for the application. This file is created and managed using pip-tools.
  11. 🧪 Directory for unit tests.
  12. 🧪 Directory for API unit tests.
  13. 🧪 Unit tests for the API endpoints.
  14. 🧪 Configuration file for pytest. This contains fixtures that are used by the unit tests. This includes a special client fixture that is used to perform requests against the application with an ephemeral migrated database.
  15. 🐍 Directory for the application source code.
  16. 🐍 Empty file to make the zoo directory a Python package.
  17. 🐍 Command-line entrypoint for the application, built with click. This is used to run the application using python -m zoo.
  18. 🐍 Version file for the application. This is used by hatch to set the version of the application.
  19. 🐍 Directory for the API endpoints.
  20. 🐍 Empty file to make the api directory a Python package.
  21. 🐍 The api module contains the API endpoints for the application. This includes the animals module, which contains the endpoints for the /animals API. These api modules contain the relevant FastAPI endpoints, as well as any supporting functions.
  22. 🐍 The app module contains the FastAPI application instance. This is where the FastAPI application is created, and where the application dependencies are configured.
  23. 🐍 The config module contains the application configuration. This includes the Settings class, which is used to configure the application using environment variables.
  24. 🐍 The db module contains the database configuration. This includes the SessionLocal class, which is used to create a SQLAlchemy session for the application.
  25. 🐍 Directory for the SQLAlchemy models.
  26. 🐍 Empty file to make the models directory a Python package.
  27. 🐍 The models module contains the SQLAlchemy models for the application.
  28. 🐍 The schemas module contains the Pydantic schemas for the application. These schemas are used to validate the request and response data for the API endpoints.
  29. 🐍 Empty file to make the schemas directory a Python package.
  30. 🐍 Pydantic schemas for the application.

Project Files#

🚢 Dockerfile

Dockerfile for building the Docker image. 

FROM python:3.11

WORKDIR /home/zoo

ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

RUN pip install --upgrade pip
COPY ./requirements.txt .
RUN pip install -r requirements.txt && rm requirements.txt

COPY zoo/ zoo/
COPY README.md README.md
COPY pyproject.toml pyproject.toml

⚙ alembic.ini

Configuration file for Alembic project. 

# A generic, single database configuration.

[alembic]
script_location = migrations
file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
prepend_sys_path = .
version_path_separator = os

[post_write_hooks]
hooks = black
black.type = console_scripts
black.entrypoint = black

[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

🐳 docker-compose.yaml

Docker Compose configuration file for running the application. 

services:
    server:
        build:
            context: .
            dockerfile: Dockerfile
        command: |
            bash -c "alembic upgrade head &&
            gunicorn zoo.app:app"
        volumes:
            - ./:/home/zoo
        ports:
            - 8000:8000
        environment:
            ZOO_PRODUCTION: true
            ZOO_DOCKER: true
            ZOO_JWT_EXPIRATION: 3600
            ZOO_DATABASE_DRIVER: postgresql+asyncpg
            ZOO_DATABASE_HOST: db
            ZOO_DATABASE_PORT: 5432
            ZOO_DATABASE_USER: postgres
            ZOO_DATABASE_PASSWORD: postgres
            ZOO_DATABASE_NAME: zoo
            GUNICORN_CMD_ARGS: >
                --workers 4
                --worker-class uvicorn.workers.UvicornWorker
                --bind 0.0.0.0:8000
                --preload
                --log-level=info
                --access-logfile "-"
        depends_on:
            - db

    db:
        image: postgres:15.3
        ports:
            - 5432:5432
        environment:
            POSTGRES_USER: postgres
            POSTGRES_PASSWORD: postgres
            POSTGRES_DB: zoo

🐍 migrations/env.py

Configuration file for Alembic migrations to interface with the application database. 

"""
Alembic migration environment.
"""

import asyncio
import logging
from logging.config import fileConfig
from os import getenv

import sqlalchemy.engine
from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine

from zoo.config import app_config
from zoo.models import __all_models__
from zoo.models.base import Base

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# Register all Database Models with Alembic
known_models = __all_models__
target_metadata = Base.metadata

if (
    not app_config.DOCKER and getenv("PYTEST_CURRENT_TEST", None) is None
):  # pragma: no cover
    app_config.rich_logging(loggers=[logging.getLogger()])


def run_migrations_offline() -> None:
    """
    Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = app_config.connection_string
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def sync_run_migrations(connection: sqlalchemy.engine.Connection) -> None:
    """
    Run migrations in 'sync' mode.
    """
    context.configure(
        connection=connection, target_metadata=target_metadata, include_schemas=True
    )

    with context.begin_transaction():
        context.run_migrations()


async def run_migrations_online() -> None:
    """
    Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.
    """
    engine = create_async_engine(
        app_config.connection_string, echo=app_config.DEBUG, future=True
    )
    async with engine.connect() as connection:
        await connection.run_sync(sync_run_migrations)
    await engine.dispose()


if context.is_offline_mode():
    run_migrations_offline()
else:
    asyncio.run(run_migrations_online())

📄 migrations/script.py.mako

Template file for Alembic migrations. 

"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}

"""

from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
${imports if imports else ""}

revision: str = ${repr(up_revision)}
down_revision: Union[str, None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}


def upgrade() -> None:
    """
    Upgrade the database
    """
    ${upgrades if upgrades else "pass"}


def downgrade() -> None:
    """
    Rollback the database upgrade
    """
    ${downgrades if downgrades else "pass"}

🐍 migrations/versions/2023_08_20_1729-e5fccc3522ce_initial_migration.py

Initial migration script generated by Alembic. This is an example of what a migration script looks like - containing an upgrade and downgrade function. 

"""Initial Migration

Revision ID: e5fccc3522ce
Revises:
Create Date: 2023-08-20 17:29:33.690933

"""

from typing import Sequence, Union

import fastapi_users_db_sqlalchemy
import sqlalchemy as sa
from alembic import op

revision: str = "e5fccc3522ce"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    """
    Upgrade the database
    """
    op.create_table(
        "exhibits",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("name", sa.String(), nullable=False),
        sa.Column("description", sa.String(), nullable=True),
        sa.Column("location", sa.String(), nullable=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("(CURRENT_TIMESTAMP)"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("(CURRENT_TIMESTAMP)"),
            nullable=False,
        ),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_table(
        "user",
        sa.Column("id", fastapi_users_db_sqlalchemy.generics.GUID(), nullable=False),
        sa.Column("email", sa.String(length=320), nullable=False),
        sa.Column("hashed_password", sa.String(length=1024), nullable=False),
        sa.Column("is_active", sa.Boolean(), nullable=False),
        sa.Column("is_superuser", sa.Boolean(), nullable=False),
        sa.Column("is_verified", sa.Boolean(), nullable=False),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("(CURRENT_TIMESTAMP)"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("(CURRENT_TIMESTAMP)"),
            nullable=False,
        ),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_user_email"), "user", ["email"], unique=True)
    op.create_table(
        "access_token",
        sa.Column(
            "user_id", fastapi_users_db_sqlalchemy.generics.GUID(), nullable=False
        ),
        sa.Column("token", sa.String(length=43), nullable=False),
        sa.Column(
            "created_at",
            fastapi_users_db_sqlalchemy.generics.TIMESTAMPAware(timezone=True),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("(CURRENT_TIMESTAMP)"),
            nullable=False,
        ),
        sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="cascade"),
        sa.PrimaryKeyConstraint("token"),
    )
    op.create_index(
        op.f("ix_access_token_created_at"), "access_token", ["created_at"], unique=False
    )
    op.create_table(
        "animals",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("name", sa.String(), nullable=False),
        sa.Column("description", sa.String(), nullable=True),
        sa.Column("species", sa.String(), nullable=True),
        sa.Column("exhibit_id", sa.Integer(), nullable=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("(CURRENT_TIMESTAMP)"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("(CURRENT_TIMESTAMP)"),
            nullable=False,
        ),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
        sa.ForeignKeyConstraint(
            ["exhibit_id"],
            ["exhibits.id"],
        ),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_table(
        "staff",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("name", sa.String(), nullable=False),
        sa.Column("job_title", sa.String(), nullable=True),
        sa.Column("email", sa.String(), nullable=True),
        sa.Column("phone", sa.String(), nullable=True),
        sa.Column("notes", sa.String(), nullable=True),
        sa.Column("exhibit_id", sa.Integer(), nullable=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("(CURRENT_TIMESTAMP)"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("(CURRENT_TIMESTAMP)"),
            nullable=False,
        ),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
        sa.ForeignKeyConstraint(
            ["exhibit_id"],
            ["exhibits.id"],
        ),
        sa.PrimaryKeyConstraint("id"),
    )


def downgrade() -> None:
    """
    Rollback the database upgrade
    """
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table("staff")
    op.drop_table("animals")
    op.drop_index(op.f("ix_access_token_created_at"), table_name="access_token")
    op.drop_table("access_token")
    op.drop_index(op.f("ix_user_email"), table_name="user")
    op.drop_table("user")
    op.drop_table("exhibits")

⚙ pyproject.toml

Configuration file for Python Package and Tooling. 

[build-system]
build-backend = "hatchling.build"
requires = ["hatchling"]

[project]
authors = [
  {name = "Justin Flannery", email = "[email protected]"}
]
classifiers = [
  "Development Status :: 4 - Beta",
  "Operating System :: OS Independent",
  "Programming Language :: Python",
  "Programming Language :: Python :: 3.8",
  "Programming Language :: Python :: 3.9",
  "Programming Language :: Python :: 3.10",
  "Programming Language :: Python :: 3.11",
  "Programming Language :: Python :: 3.12",
  "Programming Language :: Python :: Implementation :: CPython",
  "Programming Language :: Python :: Implementation :: PyPy"
]
dependencies = [
  "fastapi~=0.109.2",
  "fastapi-users[sqlalchemy]~=12.1.3",
  "pydantic[email]==2.6.1",
  "pydantic-settings~=2.1.0",
  "pydantic-extra-types~=2.5.0",
  "sqlalchemy[asyncio]~=2.0.27",
  "aiosqlite~=0.19.0",
  "asyncpg~=0.29.0",
  "greenlet~=3.0.3",
  "uvicorn[standard]~=0.27.1",
  "gunicorn~=21.2.0",
  "alembic~=1.13.1",
  "httpx~=0.26.0",
  "rich~=13.7.0",
  "click~=8.1.7",
  # https://github.com/fastapi-users/fastapi-users/issues/1301
  "setuptools; python_version == '3.12'"
]
description = "An asynchronous zoo API, powered by FastAPI and SQLModel"
dynamic = ["version"]
keywords = [
  "fastapi",
  "sqlalchemy",
  "async",
  "alembic"
]
license = "MIT"
name = "zoo"
readme = "README.md"
requires-python = ">=3.8"

[project.urls]
Documentation = "https://github.com/juftin/zoo#readme"
Issues = "https://github.com/juftin/zoo/issues"
Source = "https://github.com/juftin/zoo"

[tool.coverage.paths]
tests = ["tests", "*/zoo/tests"]
zoo = ["zoo", "*/zoo/zoo"]

[tool.coverage.report]
exclude_lines = [
  "no cov",
  "if __name__ == .__main__.:",
  "if TYPE_CHECKING:"
]
show_missing = true

[tool.coverage.run]
branch = true
omit = [
  "zoo/__init__.py",
  "zoo/__main__.py",
  "migrations/**"
]
parallel = true
source_pkgs = ["zoo", "tests"]

[tool.hatch.env]
requires = ["hatch-pip-compile", "hatch-mkdocs"]

[tool.hatch.env.collectors.mkdocs.docs]
path = "mkdocs.yaml"

[tool.hatch.envs.all]
pip-compile-args = [
  "--allow-unsafe"
]
pip-compile-constraint = ""
template = "test"

[[tool.hatch.envs.all.matrix]]
python = ["3.8", "3.9", "3.10", "3.11", "3.12"]

[tool.hatch.envs.all.scripts]
migrate = "alembic upgrade head"

[tool.hatch.envs.app]
detached = false

[tool.hatch.envs.app.scripts]
container = "docker compose up --build"
container-reset = ["docker compose down", "container"]
migrate = "alembic upgrade head"
reset = ["rm -f zoo/zoo.sqlite", "serve"]
serve = [
  "migrate",
  "uvicorn zoo.app:app --reload --host 0.0.0.0 --port 8000"
]

[tool.hatch.envs.default]
pip-compile-constraint = "default"
post-install-commands = [
  "- pre-commit install"
]
type = "pip-compile"

[tool.hatch.envs.default.scripts]
cov = "hatch run test:cov"
test = "hatch run test:test"

[tool.hatch.envs.docs]
detached = false
pip-compile-constraint = "default"
template = "docs"
type = "pip-compile"

[tool.hatch.envs.gen]
detached = false

[tool.hatch.envs.gen.scripts]
all = ["docs"]
docs = ["openapi"]
openapi = "python -m zoo openapi"
release = [
  "npm install --prefix .github/semantic_release/",
  "npx --prefix .github/semantic_release/ semantic-release {args:}"
]

[tool.hatch.envs.lint]
dependencies = [
  "mypy>=1.6.1",
  "ruff~=0.1.7"
]
detached = true
type = "pip-compile"

[tool.hatch.envs.lint.scripts]
all = [
  "style",
  "typing"
]
fmt = [
  "ruff format {args:.}",
  "ruff --fix {args:.}",
  "style"
]
precommit = [
  "pre-commit run --all-files"
]
style = [
  "ruff {args:.}",
  "ruff format --check --diff {args:.}"
]
typing = "mypy --install-types --non-interactive {args:zoo tests migrations}"

[tool.hatch.envs.test]
dependencies = [
  "pytest",
  "pytest-cov"
]

[tool.hatch.envs.test.scripts]
cov = "pytest --cov --cov-config=pyproject.toml {args:tests}"
test = "pytest {args:tests}"

[tool.hatch.version]
path = "zoo/_version.py"

[tool.mypy]
check_untyped_defs = true
disallow_any_generics = true
disallow_untyped_defs = true
follow_imports = "silent"
ignore_missing_imports = true
no_implicit_reexport = true
warn_redundant_casts = true
warn_unused_ignores = true

[tool.ruff]
ignore = [
  # Ignore checks for possible passwords
  "S105",
  "S106",
  "S107",
  # Ignore complexity
  "C901",
  "PLR0911",
  "PLR0912",
  "PLR0913",
  "PLR0915",
  # Boolean-typed positional argument in function definition
  "FBT001",
  # Boolean default positional argument in function definition
  "FBT002",
  # Allow boolean positional values in function calls, like `dict.get(... True)`
  "FBT003",
  # Exception must not use a string literal, assign to variable first
  "EM101",
  # Ignore Depends on FastAPI
  "B008"
]
line-length = 88
select = [
  "A",  # flake8-builtins
  "ARG",  # flake8-unused-arguments
  "B",  # flake8-bugbear
  "C",  # mccabe
  "DTZ",  # flake8-datetimez
  "E",  # pycodestyle (Error)
  "EM",  # flake8-errmsg
  "F",  # Pyflakes
  "FBT",  # flake8-boolean-trap
  "I",  # isort
  "ICN",  # flake8-import-conventions
  "N",  # pep8-naming
  "PLC",  # Pylint (Convention message)
  "PLE",  # Pylint (Error message)
  "PLR",  # Pylint (Refactor message)
  "PLW",  # Pylint (Warning message)
  "Q",  # flake8-quotes
  "RUF",  # Ruff-specific rules
  "S",  # flake8-bandit
  "T",  # flake8-debugger (T10) and flake8-print (T20)
  "TID",  # flake8-tidy-imports
  "UP",  # pyupgrade
  "W",  # pycodestyle (Warning)
  "YTT"  # flake8-2020
]
target-version = "py38"

[tool.ruff.flake8-tidy-imports]
ban-relative-imports = "all"

[tool.ruff.isort]
known-first-party = ["zoo"]

[tool.ruff.per-file-ignores]
# Tests can use magic values, assertions, and relative imports
"tests/**/*" = ["PLR2004", "S101", "TID252"]

[tool.ruff.pydocstyle]
convention = "numpy"

⚙ requirements.txt

Python dependencies for the application. This file is created and managed using pip-tools. 

#
# This file is autogenerated by hatch-pip-compile with Python 3.11
#
# - aiosqlite~=0.19.0
# - alembic~=1.13.1
# - asyncpg~=0.29.0
# - click~=8.1.7
# - fastapi-users[sqlalchemy]~=12.1.3
# - fastapi~=0.109.2
# - greenlet~=3.0.3
# - gunicorn~=21.2.0
# - httpx~=0.26.0
# - pydantic-extra-types~=2.5.0
# - pydantic-settings~=2.1.0
# - pydantic[email]==2.6.1
# - rich~=13.7.0
# - setuptools; python_version == "3.12"
# - sqlalchemy[asyncio]~=2.0.27
# - uvicorn[standard]~=0.27.1
#

aiosqlite==0.19.0
    # via hatch.envs.default
alembic==1.13.1
    # via hatch.envs.default
annotated-types==0.6.0
    # via pydantic
anyio==4.2.0
    # via
    #   httpx
    #   starlette
    #   watchfiles
async-timeout==4.0.3
    # via asyncpg
asyncpg==0.29.0
    # via hatch.envs.default
bcrypt==4.1.2
    # via passlib
certifi==2023.11.17
    # via
    #   httpcore
    #   httpx
cffi==1.16.0
    # via cryptography
click==8.1.7
    # via
    #   hatch.envs.default
    #   uvicorn
cryptography==41.0.7
    # via pyjwt
dnspython==2.4.2
    # via email-validator
email-validator==2.0.0.post2
    # via
    #   fastapi-users
    #   pydantic
fastapi==0.109.2
    # via
    #   hatch.envs.default
    #   fastapi-users
fastapi-users==12.1.3
    # via
    #   hatch.envs.default
    #   fastapi-users
    #   fastapi-users-db-sqlalchemy
fastapi-users-db-sqlalchemy==6.0.1
    # via fastapi-users
greenlet==3.0.3
    # via
    #   hatch.envs.default
    #   sqlalchemy
gunicorn==21.2.0
    # via hatch.envs.default
h11==0.14.0
    # via
    #   httpcore
    #   uvicorn
httpcore==1.0.3
    # via httpx
httptools==0.6.1
    # via uvicorn
httpx==0.26.0
    # via hatch.envs.default
idna==3.6
    # via
    #   anyio
    #   email-validator
    #   httpx
makefun==1.15.2
    # via fastapi-users
mako==1.3.0
    # via alembic
markdown-it-py==3.0.0
    # via rich
markupsafe==2.1.3
    # via mako
mdurl==0.1.2
    # via markdown-it-py
packaging==23.2
    # via gunicorn
passlib==1.7.4
    # via
    #   fastapi-users
    #   passlib
pycparser==2.21
    # via cffi
pydantic==2.6.1
    # via
    #   hatch.envs.default
    #   fastapi
    #   pydantic
    #   pydantic-extra-types
    #   pydantic-settings
pydantic-core==2.16.2
    # via pydantic
pydantic-extra-types==2.5.0
    # via hatch.envs.default
pydantic-settings==2.1.0
    # via hatch.envs.default
pygments==2.17.2
    # via rich
pyjwt==2.8.0
    # via
    #   fastapi-users
    #   pyjwt
python-dotenv==1.0.0
    # via
    #   pydantic-settings
    #   uvicorn
python-multipart==0.0.7
    # via fastapi-users
pyyaml==6.0.1
    # via uvicorn
rich==13.7.0
    # via hatch.envs.default
sniffio==1.3.0
    # via
    #   anyio
    #   httpx
sqlalchemy==2.0.27
    # via
    #   hatch.envs.default
    #   alembic
    #   fastapi-users-db-sqlalchemy
    #   sqlalchemy
starlette==0.36.3
    # via fastapi
typing-extensions==4.9.0
    # via
    #   alembic
    #   fastapi
    #   pydantic
    #   pydantic-core
    #   sqlalchemy
uvicorn==0.27.1
    # via
    #   hatch.envs.default
    #   uvicorn
uvloop==0.19.0
    # via uvicorn
watchfiles==0.21.0
    # via uvicorn
websockets==12.0
    # via uvicorn

🧪 tests/api/test_animals_api.py

Unit tests for the API endpoints. 

"""
API Tests: /animals
"""

import datetime
from os import environ

from fastapi.testclient import TestClient

from zoo.schemas.animals import AnimalsCreate, AnimalsRead, AnimalsUpdate


def test_get_animals(migrated_client: TestClient) -> None:
    """
    Test GET /animals
    """
    response = migrated_client.get("/animals")
    assert response.status_code == 200
    response_data = response.json()
    first_animal = AnimalsRead(**response_data[0])
    assert isinstance(first_animal.updated_at, datetime.datetime)


def test_get_animal(migrated_client: TestClient) -> None:
    """
    Test GET /animals/{animal_id}
    """
    response = migrated_client.get("/animals/1")
    assert response.status_code == 200
    response_data = response.json()
    first_animal = AnimalsRead(**response_data)
    assert isinstance(first_animal.updated_at, datetime.datetime)


def test_get_animal_failure(migrated_client: TestClient) -> None:
    """
    Test GET /animals/{animal_id} - failure
    """
    response = migrated_client.get("/animals/100")
    assert response.status_code == 404
    assert response.json() == {
        "detail": "Error: `Animals` data not found or deleted - ID: 100"
    }


def test_create_animal(migrated_client: TestClient) -> None:
    """
    Test POST /animals
    """
    test_name = environ["PYTEST_CURRENT_TEST"].split(":")[-1].split(" ")[0]
    animal_body = AnimalsCreate(
        name=test_name,
        description="test",
    )
    response = migrated_client.post(
        "/animals",
        json=animal_body.model_dump(exclude_unset=True),
    )
    assert response.status_code == 200
    response_data = response.json()
    animal = AnimalsRead(**response_data)
    assert animal.name == test_name
    assert animal.description == "test"


def test_update_animal(migrated_client: TestClient) -> None:
    """
    Test POST /animals/{animal_id}
    """
    test_name = environ["PYTEST_CURRENT_TEST"].split(":")[-1].split(" ")[0]
    animal_body = AnimalsUpdate(
        description=test_name,
    )
    response = migrated_client.patch(
        "/animals/2",
        json=animal_body.model_dump(exclude_unset=True),
    )
    # assert response.status_code == 200
    response_data = response.json()
    animal = AnimalsRead(**response_data)
    assert animal.description == test_name


def test_delete_animal(migrated_client: TestClient) -> None:
    """
    Test DELETE /animals/{animal_id}
    """
    response = migrated_client.delete("/animals/5")
    assert response.status_code == 200
    response_data = response.json()
    animal = AnimalsRead(**response_data)
    assert animal.deleted_at is not None

🧪 tests/conftest.py

Configuration file for pytest. This contains fixtures that are used by the unit tests. This includes a special client fixture that is used to perform requests against the application with an ephemeral migrated database. 

"""
Test fixtures for the project.
"""

import pathlib
from tempfile import TemporaryDirectory
from typing import Generator

import pytest
from alembic import command
from alembic.config import Config
from fastapi.testclient import TestClient

from zoo.config import app_config


@pytest.fixture(scope="session")
def migrated_client() -> Generator[TestClient, None, None]:
    """
    Test client for the FastAPI app with a shared temporary database.

    This fixture is session-scoped, so it will only be created once for
    the entire test session. It creates a temporary directory and database
    file, and runs the Alembic migrations to set up the database schema.
    It then imports the FastAPI app and returns a test client for it.
    """
    with pytest.MonkeyPatch.context() as monkeypatch, TemporaryDirectory() as temp_dir:
        # Create a temporary database file and update the config
        temp_path = pathlib.Path(temp_dir)
        temp_db = temp_path / "zoo.sqlite"
        monkeypatch.setattr(target=app_config, name="DATABASE_FILE", value=str(temp_db))
        monkeypatch.setenv("ZOO_DATABASE_FILE", str(temp_db))
        # Set the seed data flag to True
        monkeypatch.setattr(target=app_config, name="SEED_DATA", value=True)
        monkeypatch.setenv("ZOO_SEED_DATA", "True")
        # Change directory to the root of the project
        zoo_dir = pathlib.Path(__file__).parent.parent
        monkeypatch.chdir(zoo_dir)
        # Run migrations before importing app
        config = Config(zoo_dir / "alembic.ini")
        command.upgrade(config, "head")
        # Import app after monkeypatching
        from zoo.app import app as fastapi_app

        # Return the test client
        yield TestClient(fastapi_app)

🐍 zoo/main.py

Command-line entrypoint for the application, built with click. This is used to run the application using python -m zoo. 

"""
Zoo CLI
"""

import asyncio
import json
import logging
import pathlib
from dataclasses import dataclass

import click
import uvicorn
from click import Context
from fastapi_users.exceptions import UserAlreadyExists

from zoo._version import __application__, __version__
from zoo.app import ZooFastAPI, app
from zoo.config import ZooSettings, app_config
from zoo.models.users import create_user

logger = logging.getLogger(__name__)


@dataclass
class ZooContext:
    """
    Context Object Passed Around Application
    """

    app: ZooFastAPI
    config: ZooSettings


@click.group(invoke_without_command=True, name=__application__)
@click.version_option(version=__version__, prog_name=__application__)
@click.option("-h", "--host", default="localhost", help="Host to listen on")
@click.option("-p", "--port", default=8000, help="Port to listen on", type=int)
@click.option("-d", "--debug", is_flag=True, help="Enable debug mode", default=False)
@click.pass_context
def cli(context: click.Context, host: str, port: int, *, debug: bool) -> None:
    """
    Zoo CLI
    """
    context.obj = ZooContext(app=app, config=app_config)
    if context.obj.config.DEBUG is True or debug is True:
        context.obj.config.DEBUG = True
        logging.root.setLevel(logging.DEBUG)
    if context.invoked_subcommand is None:
        uvicorn.run(context.obj.app, host=host, port=port)


@cli.command()
@click.pass_obj
def openapi(context: ZooContext) -> None:
    """
    Generate openapi.json
    """
    openapi_body = context.app.openapi()
    logger.debug(openapi_body)
    json_file = pathlib.Path(__file__).parent.parent / "docs" / "openapi.json"
    logger.info("Generating OpenAPI Spec: %s", json_file)
    json_file.write_text(json.dumps(openapi_body, indent=2))


@cli.command()
@click.option("-e", "--email", help="User email to create", type=str, required=True)
@click.option(
    "-p",
    "--password",
    default="admin",
    help="Password to create",
    type=click.STRING,
    required=True,
)
@click.pass_context
def users(context: Context, email: str, password: str) -> None:
    """
    Create users
    """
    _ = context
    logger.info("Creating user: %s", email)
    try:
        user = asyncio.run(create_user(email=email, password=password))
        logger.info("Created user: %s", user.id)
    except UserAlreadyExists:
        logger.info("User already exists: %s", email)
        context.exit(1)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    if not app_config.DOCKER:
        app_config.rich_logging(loggers=[logging.getLogger()])
    cli()

🐍 zoo/_version.py

Version file for the application. This is used by hatch to set the version of the application. 

"""
Application Info
"""

from textwrap import dedent

__version__ = "0.2.0"
__application__ = "zoo"
__description__ = (
    "An asynchronous zoo API, powered by FastAPI, "
    "SQLAlchemy 2.0, Pydantic v2, and Alembic"
)
__favicon__ = "https://raw.githubusercontent.com/juftin/juftin/main/static/juftin.png"

_md_desc = f"""
### {__description__}

[<img src="{__favicon__}" width="120" height="120"  alt="juftin logo">](https://juftin.com)
"""

__markdown_description__ = dedent(_md_desc).strip()

🐍 zoo/api/animals.py

The api module contains the API endpoints for the application. This includes the animals module, which contains the endpoints for the /animals API. These api modules contain the relevant FastAPI endpoints, as well as any supporting functions. 

"""
Animals Router app
"""

import logging
from typing import List, Optional, Sequence

from fastapi import APIRouter, Depends, Query
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from zoo.api.utils import check_model
from zoo.db import get_async_session
from zoo.models.animals import Animals
from zoo.schemas.animals import AnimalsCreate, AnimalsRead, AnimalsUpdate

logger = logging.getLogger(__name__)

animals_router = APIRouter(tags=["animals"])


@animals_router.get("/animals", response_model=List[AnimalsRead])
async def get_animals(
    offset: int = 0,
    limit: int = Query(default=100, le=100),
    session: AsyncSession = Depends(get_async_session),
) -> List[AnimalsRead]:
    """
    Get animals from the database
    """
    result = await session.execute(
        select(Animals)
        .where(Animals.deleted_at.is_(None))
        .order_by(Animals.id)
        .offset(offset)
        .limit(limit)
    )
    animals: Sequence[Animals] = result.scalars().all()
    animals_models = [AnimalsRead.model_validate(animal) for animal in animals]
    return animals_models


@animals_router.post("/animals", response_model=AnimalsRead)
async def create_animal(
    animal: AnimalsCreate, session: AsyncSession = Depends(get_async_session)
) -> AnimalsRead:
    """
    Create a new animal in the database
    """
    new_animal = Animals(**animal.model_dump(exclude_unset=True))
    session.add(new_animal)
    await session.commit()
    await session.refresh(new_animal)
    new_animal_model = AnimalsRead.model_validate(new_animal)
    return new_animal_model


@animals_router.get("/animals/{animal_id}", response_model=AnimalsRead)
async def get_animal(
    animal_id: int, session: AsyncSession = Depends(get_async_session)
) -> AnimalsRead:
    """
    Get an animal from the database
    """
    animal: Optional[Animals] = await session.get(Animals, animal_id)
    animal = check_model(model_instance=animal, model_class=Animals, id=animal_id)
    animal_model = AnimalsRead.model_validate(animal)
    return animal_model


@animals_router.delete("/animals/{animal_id}", response_model=AnimalsRead)
async def delete_animal(
    animal_id: int, session: AsyncSession = Depends(get_async_session)
) -> AnimalsRead:
    """
    Delete an animal from the database
    """
    animal: Optional[Animals] = await session.get(Animals, animal_id)
    animal = check_model(model_instance=animal, model_class=Animals, id=animal_id)
    animal.deleted_at = func.current_timestamp()
    session.add(animal)
    await session.commit()
    await session.refresh(animal)
    animal_model = AnimalsRead.model_validate(animal)
    return animal_model


@animals_router.patch("/animals/{animal_id}", response_model=AnimalsRead)
async def update_animal(
    animal_id: int,
    animal: AnimalsUpdate,
    session: AsyncSession = Depends(get_async_session),
) -> AnimalsRead:
    """
    Update an animal in the database
    """
    db_animal: Optional[Animals] = await session.get(Animals, animal_id)
    db_animal = check_model(model_instance=db_animal, model_class=Animals, id=animal_id)
    for field, value in animal.model_dump(exclude_unset=True).items():
        setattr(db_animal, field, value)
    session.add(db_animal)
    await session.commit()
    await session.refresh(db_animal)
    animal_model = AnimalsRead.model_validate(db_animal)
    return animal_model

🐍 zoo/app.py

The app module contains the FastAPI application instance. This is where the FastAPI application is created, and where the application dependencies are configured. 

"""
zoo app
"""

import uvicorn
from fastapi import FastAPI

from zoo._version import __application__, __markdown_description__, __version__
from zoo.api.animals import animals_router
from zoo.api.exhibits import exhibits_router
from zoo.api.staff import staff_router
from zoo.api.utils import utils_router
from zoo.config import app_config
from zoo.models.users import bootstrap_fastapi_users

if not app_config.DOCKER:
    app_config.rich_logging(
        loggers=[
            "uvicorn",
            "uvicorn.access",
        ]
    )


class ZooFastAPI(FastAPI):
    """
    Zoo FastAPI
    """


app = ZooFastAPI(
    title=__application__,
    description=__markdown_description__,
    version=__version__,
    debug=False,
    docs_url=None,  # Custom Swagger UI @ utils_router
    redoc_url=None,
    generate_unique_id_function=app_config.custom_generate_unique_id,
)
# Routers
app_routers = [
    utils_router,
    animals_router,
    exhibits_router,
    staff_router,
]
for router in app_routers:
    app.include_router(router)
# FastAPI Users Setup
bootstrap_fastapi_users(app=app)


if __name__ == "__main__":
    uvicorn.run(app)

🐍 zoo/config.py

The config module contains the application configuration. This includes the Settings class, which is used to configure the application using environment variables. 

"""
Application configuration
"""

import asyncio
import logging
import pathlib
from typing import List, Optional, Union

import fastapi
import starlette
import uvicorn
from fastapi.routing import APIRoute
from pydantic_settings import BaseSettings, SettingsConfigDict
from rich.logging import RichHandler
from sqlalchemy.engine import URL

from zoo._version import __application__

rich_handler = RichHandler(
    rich_tracebacks=True,
    tracebacks_show_locals=True,
    show_time=True,
    omit_repeated_times=False,
    tracebacks_suppress=[
        starlette,
        fastapi,
        uvicorn,
        asyncio,
    ],
    log_time_format="[%Y-%m-%d %H:%M:%S]",
)


class ZooSettings(BaseSettings):
    """
    Application configuration
    """

    PRODUCTION: bool = False
    DOCKER: bool = False
    DEBUG: bool = False

    DATABASE_FILE: str = str(pathlib.Path(__file__).resolve().parent / "zoo.sqlite")
    DATABASE_DRIVER: str = "sqlite+aiosqlite"
    DATABASE_HOST: Optional[str] = None
    DATABASE_PORT: Optional[int] = None
    DATABASE_USER: Optional[str] = None
    DATABASE_PASSWORD: Optional[str] = None
    DATABASE_NAME: Optional[str] = None
    JWT_EXPIRATION: Optional[int] = None
    SEED_DATA: bool = True

    DATABASE_SECRET: str = __application__

    model_config = SettingsConfigDict(
        env_prefix="ZOO_",
        case_sensitive=True,
    )

    @property
    def connection_string(self) -> str:
        """
        Get the database connection string
        """
        database_url = URL.create(
            drivername=self.DATABASE_DRIVER,
            username=self.DATABASE_USER,
            password=self.DATABASE_PASSWORD,
            host=self.DATABASE_HOST or self.DATABASE_FILE,
            port=self.DATABASE_PORT,
            database=self.DATABASE_NAME,
        ).render_as_string(hide_password=False)
        if all(
            [
                self.DATABASE_HOST is None,
                "sqlite" in self.DATABASE_DRIVER.lower(),
                "////" not in database_url,
            ]
        ):
            database_url = str(database_url).replace("///", "////", 1)
        return database_url

    @classmethod
    def rich_logging(cls, loggers: List[Union[str, logging.Logger]]) -> None:
        """
        Configure logging for development
        """
        for logger in loggers:
            if isinstance(logger, str):
                logger_inst = logging.getLogger(logger)
            else:
                logger_inst = logger
            logger_inst.handlers = [rich_handler]

    @classmethod
    def custom_generate_unique_id(cls, route: APIRoute) -> str:
        """
        Custom function to generate unique id for each route
        """
        return f"{route.tags[0]}-{route.name}"


app_config = ZooSettings()

🐍 zoo/db.py

The db module contains the database configuration. This includes the SessionLocal class, which is used to create a SQLAlchemy session for the application. 

"""
Database Connections
"""

from typing import AsyncGenerator

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from zoo.config import app_config

async_engine = create_async_engine(
    app_config.connection_string, echo=app_config.DEBUG, future=True
)
async_session = async_sessionmaker(
    async_engine,
    class_=AsyncSession,
    autocommit=False,
    expire_on_commit=False,
    autoflush=False,
)


async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    """
    Yield an AsyncSession

    Used by FastAPI Depends
    """
    try:
        async with async_session() as session:
            yield session
    finally:
        await session.close()

🐍 zoo/models/animals.py

The models module contains the SQLAlchemy models for the application. 

"""
Animals Database Model
"""
from typing import TYPE_CHECKING

from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

from zoo.models.base import Base, CreatedUpdatedMixin, DeletedAtMixin, IDMixin

if TYPE_CHECKING:  # pragma: no cover
    from zoo.models.exhibits import Exhibits


class Animals(IDMixin, CreatedUpdatedMixin, DeletedAtMixin, Base):
    """
    Animals Database Model
    """

    __tablename__ = "animals"

    name: Mapped[str]
    description: Mapped[str] = mapped_column(default=None, nullable=True)
    species: Mapped[str] = mapped_column(default=None, nullable=True)
    exhibit_id: Mapped[int] = mapped_column(
        ForeignKey("exhibits.id"), nullable=True, default=None
    )

    exhibit: Mapped["Exhibits"] = relationship(back_populates="animals")

🐍 zoo/schemas/animals.py

The schemas module contains the Pydantic schemas for the application. These schemas are used to validate the request and response data for the API endpoints. 

"""
Animal models
"""

from typing import Any, ClassVar, Dict, Optional

from pydantic import ConfigDict, Field

from zoo.schemas.base import (
    CreatedModifiedMixin,
    DeletedMixin,
    RequiredIdMixin,
    ZooModel,
)


class AnimalsBase(ZooModel):
    """
    Animals model base
    """

    name: str = Field(description="The name of the animal")
    description: Optional[str] = Field(
        default=None, description="The description of the animal"
    )
    species: Optional[str] = Field(
        default=None, description="The species of the animal"
    )

    exhibit_id: Optional[int] = Field(description="The id of the exhibit", default=None)

    __example__: ClassVar[Dict[str, Any]] = {
        "name": "Lion",
        "description": "Ferocious kitty",
        "species": "Panthera leo",
        "exhibit_id": 1,
    }


class AnimalsCreate(AnimalsBase):
    """
    Animals model: create
    """

    model_config = ConfigDict(
        json_schema_extra=AnimalsBase.get_openapi_create_example()
    )


class AnimalsRead(
    DeletedMixin,
    CreatedModifiedMixin,
    AnimalsBase,
    RequiredIdMixin,
):
    """
    Animals model: read
    """

    model_config = ConfigDict(
        from_attributes=True,
        json_schema_extra=AnimalsBase.get_openapi_read_example(),
    )


class AnimalsUpdate(ZooModel):
    """
    Animals model: update
    """

    name: Optional[str] = Field(default=None, description="The name of the animal")
    description: Optional[str] = Field(
        default=None, description="The description of the animal"
    )
    species: Optional[str] = Field(
        default=None, description="The species of the animal"
    )
    exhibit_id: Optional[int] = Field(description="The id of the exhibit", default=None)

    model_config = ConfigDict(
        json_schema_extra=AnimalsBase.get_openapi_update_example()
    )