Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -9,3 +9,6 @@

# macOS
.DS_Store

# Environment files containing secrets
.env
10 changes: 6 additions & 4 deletions Dockerfile
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
FROM python:3.11
RUN apt-get update && apt-get install -y libpq-dev
FROM python:3.11-slim
RUN apt-get update && apt-get install -y libpq-dev && rm -rf /var/lib/apt/lists/*
COPY . /app
WORKDIR /app
RUN pip install -r requirements.txt
RUN pip install --upgrade pip && pip install -r requirements.txt
RUN adduser --disabled-password --gecos "" appuser && chown -R appuser /app
USER appuser
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--log-level", "trace"]
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--log-level", "info"]
75 changes: 50 additions & 25 deletions conftest.py
Original file line number Diff line number Diff line change
@@ -1,57 +1,82 @@
"""Pytest configuration and shared fixtures for the Books API test suite."""

import os

# Set DATABASE_URL before any application modules are imported so that
# dependencies.py does not raise EnvironmentError during test collection.
# The actual test fixtures use their own in-memory SQLite engines, so this
# value is only used for the initial module-level import guard.
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")

import pytest
from sqlalchemy import create_engine, inspect, text
from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from fastapi.testclient import TestClient
from main import create_app
from dependencies import get_db, database_url
from dependencies import get_db
from models import Base


def _make_memory_engine():
"""Return a fresh in-memory SQLite engine with the schema applied."""
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(bind=engine)
return engine


@pytest.fixture(scope="session")
def test_engine():
"""Create test database engine"""
test_engine = create_engine(database_url)
Base.metadata.create_all(bind=test_engine)
yield test_engine
Base.metadata.drop_all(bind=test_engine)
"""Session-scoped engine used by application-level tests."""
engine = _make_memory_engine()
yield engine
engine.dispose()


@pytest.fixture(scope="function")
def test_db(test_engine):
"""Create test database session that rolls back after each test"""
connection = test_engine.connect()
transaction = connection.begin()
db = sessionmaker(
autocommit=False, autoflush=False, bind=connection,
join_transaction_mode="create_savepoint"
)()
def test_db():
"""
Function-scoped DB session backed by a fresh in-memory SQLite database.

A new engine (and therefore a new empty database) is created for every
test function, giving full isolation without relying on savepoint rollback
semantics that differ between SQLite and PostgreSQL.
"""
engine = _make_memory_engine()
db = sessionmaker(autocommit=False, autoflush=False, bind=engine)()
try:
yield db
finally:
db.close()
transaction.rollback()
connection.close()
engine.dispose()


@pytest.fixture
def test_app(test_engine):
"""Create test FastAPI application with test database"""
def test_app():
"""
Create a test FastAPI application backed by a fresh in-memory SQLite
database for each test function.
"""
engine = _make_memory_engine()

def override_get_db():
TestingSessionLocal = sessionmaker(
autocommit=False, autoflush=False, bind=test_engine
)
db = TestingSessionLocal()
db = sessionmaker(autocommit=False, autoflush=False, bind=engine)()
try:
yield db
finally:
db.close()

app = create_app()
app.dependency_overrides[get_db] = override_get_db
return app
yield app
engine.dispose()


@pytest.fixture
def client(test_app):
"""Create test client"""
"""Return a TestClient for the test application."""
return TestClient(test_app)
2 changes: 1 addition & 1 deletion docker-compose.yml
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ services:
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
ports:
- "5432:5432"
- "127.0.0.1:5432:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
- ./db.sql:/docker-entrypoint-initdb.d/db.sql
Expand Down
15 changes: 9 additions & 6 deletions models.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
"""Database and Pydantic models for the Books API."""

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String
from pydantic import BaseModel, ConfigDict
Expand All @@ -7,32 +9,33 @@
class Base(DeclarativeBase):
"""Base class for all database models"""

pass


class Book(Base):
"""Book model"""
"""Book ORM model representing the books table."""

__tablename__ = "books"

id: Mapped[int] = mapped_column(primary_key=True, index=True)
title: Mapped[str] = mapped_column(String(255), index=True)
author: Mapped[str] = mapped_column(String(255))
isbn: Mapped[str] = mapped_column(String(255), nullable=False)
isbn: Mapped[str] = mapped_column(String(20), unique=True, index=True)


# Pydantic models
class BookIn(BaseModel):
"""Pydantic model for book input"""
"""Pydantic model for book input (create / update)."""

title: str
author: str
isbn: str


class BookOut(BaseModel):
"""Pydantic model for book output"""
"""Pydantic model for book output (API responses)."""

id: int
title: str
author: str
isbn: str

model_config = ConfigDict(from_attributes=True)
67 changes: 56 additions & 11 deletions repositories.py
Original file line number Diff line number Diff line change
@@ -1,40 +1,85 @@
"""Repository layer: database CRUD operations for books."""

from sqlalchemy.orm import Session
import models


# Create a new book
def create_book(db: Session, book: models.BookIn):
db_book = models.Book(title=book.title, author=book.author)
def create_book(db: Session, book: models.BookIn) -> models.Book:
"""Create and persist a new book record.

Args:
db: Active database session.
book: Validated input data for the new book.

Returns:
The newly created Book ORM instance.
"""
db_book = models.Book(title=book.title, author=book.author, isbn=book.isbn)
db.add(db_book)
db.commit()
db.refresh(db_book)
return db_book


# Get all books
def get_books(db: Session, skip: int = 0, limit: int = 10):
def get_books(db: Session, skip: int = 0, limit: int = 10) -> list[models.Book]:
"""Return a paginated list of books.

Args:
db: Active database session.
skip: Number of records to skip (offset).
limit: Maximum number of records to return.

Returns:
List of Book ORM instances.
"""
return db.query(models.Book).offset(skip).limit(limit).all()


# Get a book by ID
def get_book(db: Session, book_id: int):
def get_book(db: Session, book_id: int) -> models.Book | None:
"""Fetch a single book by primary key.

Args:
db: Active database session.
book_id: Primary key of the book.

Returns:
The matching Book ORM instance, or None if not found.
"""
return db.query(models.Book).filter(models.Book.id == book_id).first()


# Update a book
def update_book(db: Session, book_id: int, book: models.BookIn):
def update_book(db: Session, book_id: int, book: models.BookIn) -> models.Book | None:
"""Update an existing book record.

Args:
db: Active database session.
book_id: Primary key of the book to update.
book: Validated input data with updated fields.

Returns:
The updated Book ORM instance, or None if not found.
"""
db_book = db.query(models.Book).filter(models.Book.id == book_id).first()
if db_book:
db_book.title = book.title
db_book.author = book.author
db_book.isbn = book.isbn
db.commit()
db.refresh(db_book)
return db_book
return None


# Delete a book
def delete_book(db: Session, book_id: int):
def delete_book(db: Session, book_id: int) -> models.Book | None:
"""Delete a book record by primary key.

Args:
db: Active database session.
book_id: Primary key of the book to delete.

Returns:
The deleted Book ORM instance, or None if not found.
"""
db_book = db.query(models.Book).filter(models.Book.id == book_id).first()
if db_book:
db.delete(db_book)
Expand Down
26 changes: 22 additions & 4 deletions routers.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,17 @@
"""FastAPI router: HTTP endpoints for the Books API."""

import logging
from typing import List

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

import models
import repositories
from dependencies import get_db

logger = logging.getLogger(__name__)

# Create router with prefix
router = APIRouter()

Expand All @@ -13,28 +20,37 @@
"/books/", response_model=models.BookOut, status_code=status.HTTP_201_CREATED
)
def create_book(book: models.BookIn, db: Session = Depends(get_db)):
"""Create a new book record."""
try:
return repositories.create_book(db=db, book=book)
except Exception as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
logger.error("Failed to create book: %s", e, exc_info=True)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Could not create book.",
) from e


@router.get(
"/books/", response_model=List[models.BookOut], status_code=status.HTTP_200_OK
)
def get_books(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
"""Return a paginated list of books."""
try:
return repositories.get_books(db=db, skip=skip, limit=limit)
except Exception as e:
logger.error("Failed to retrieve books: %s", e, exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)
)
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Could not retrieve books.",
) from e


@router.get(
"/books/{book_id}", response_model=models.BookOut, status_code=status.HTTP_200_OK
)
def get_book(book_id: int, db: Session = Depends(get_db)):
"""Return a single book by ID."""
db_book = repositories.get_book(db=db, book_id=book_id)
if db_book is None:
raise HTTPException(
Expand All @@ -47,6 +63,7 @@ def get_book(book_id: int, db: Session = Depends(get_db)):
"/books/{book_id}", response_model=models.BookOut, status_code=status.HTTP_200_OK
)
def update_book(book_id: int, book: models.BookIn, db: Session = Depends(get_db)):
"""Update an existing book by ID."""
db_book = repositories.update_book(db=db, book_id=book_id, book=book)
if db_book is None:
raise HTTPException(
Expand All @@ -59,6 +76,7 @@ def update_book(book_id: int, book: models.BookIn, db: Session = Depends(get_db)
"/books/{book_id}", response_model=models.BookOut, status_code=status.HTTP_200_OK
)
def delete_book(book_id: int, db: Session = Depends(get_db)):
"""Delete a book by ID."""
db_book = repositories.delete_book(db=db, book_id=book_id)
if db_book is None:
raise HTTPException(
Expand Down
Loading
Loading