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)!
- Dockerfile for building the Docker image.
- Alembic configuration file for managing database migrations.
- Docker Compose configuration file for running the application.
- Directory for managing database migrations.
- Configuration file for Alembic migrations to interface with the application database.
- Template file for Alembic migrations.
- Directory for Alembic migration scripts.
- Initial migration script generated by Alembic. This is an example of what
a migration script looks like - containing an
upgrade
anddowngrade
function. - Configuration file for Python Package and Tooling.
- Python dependencies for the application. This file is created and managed using pip-tools.
- Directory for unit tests.
- Directory for API unit tests.
- Unit tests for the API endpoints.
- 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. - Directory for the application source code.
- Empty file to make the
zoo
directory a Python package. - Command-line entrypoint for the application, built with
click. This is
used to run the application using
python -m zoo
. - Version file for the application. This is used by hatch to set the version of the application.
- Directory for the API endpoints.
- Empty file to make the
api
directory a Python package. - The
api
module contains the API endpoints for the application. This includes theanimals
module, which contains the endpoints for the/animals
API. Theseapi
modules contain the relevant FastAPI endpoints, as well as any supporting functions. - The
app
module contains the FastAPI application instance. This is where the FastAPI application is created, and where the application dependencies are configured. - The
config
module contains the application configuration. This includes theSettings
class, which is used to configure the application using environment variables. - The
db
module contains the database configuration. This includes theSessionLocal
class, which is used to create a SQLAlchemy session for the application. - Directory for the SQLAlchemy models.
- Empty file to make the
models
directory a Python package. - The
models
module contains the SQLAlchemy models for the application. - 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. - Empty file to make the
schemas
directory a Python package. - Pydantic schemas for the application.
Project Files#
Dockerfile
Dockerfile for building the Docker image.
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()
)