Line-by-line:

• get_book: Fetch by primary key, returns None if not found.
• get_books: Simple pagination with skip & limit.
• create_book: Construct a Book instance from Pydantic model, add, commit, and refresh to get DB-generated fields.
• update_book: Use exclude_unset to only change provided fields.
• delete_book: Delete if exists, return boolean for success.

main.py — FastAPI app and routes

# app/main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from . import models, schemas, crud
from .database import SessionLocal, engine, Base
Create DB tables (for demo). In production use Alembic migrations.
Base.metadata.create_all(bind=engine)
app = FastAPI(title="Books API")
Dependency to get DB session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
@app.post("/books/", response_model=schemas.BookOut, status_code=status.HTTP_201_CREATED)
def create_book_endpoint(book: schemas.BookCreate, db: Session = Depends(get_db)):
    return crud.create_book(db, book)
@app.get("/books/", response_model=list[schemas.BookOut])
def read_books(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud.get_books(db, skip=skip, limit=limit)
@app.get("/books/{book_id}", response_model=schemas.BookOut)
def read_book(book_id: int, db: Session = Depends(get_db)):
    db_book = crud.get_book(db, book_id)
    if not db_book:
        raise HTTPException(status_code=404, detail="Book not found")
    return db_book
@app.patch("/books/{book_id}", response_model=schemas.BookOut)
def update_book_endpoint(book_id: int, changes: schemas.BookUpdate, db: Session = Depends(get_db)):
    updated = crud.update_book(db, book_id, changes)
    if not updated:
        raise HTTPException(status_code=404, detail="Book not found")
    return updated
@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book_endpoint(book_id: int, db: Session = Depends(get_db)):
    success = crud.delete_book(db, book_id)
    if not success:
        raise HTTPException(status_code=404, detail="Book not found")
    return

Line-by-line highlights:

• Base.metadata.create_all: Creates tables automatically (OK for development).
• get_db dependency: yields a session and always closes it (very important to avoid connection leaks).
• Each route uses dependency injection to get DB sessions, returns Pydantic models, and raises HTTP errors appropriately.

• POST /books/ accepts BookCreate JSON; returns created BookOut with id.
• GET /books/ returns paginated list.
• GET/ PATCH / DELETE by id with proper error handling.

Automating Data Entry: Web Scraping to Seed Data

Want to bulk-add books automatically? You can write a small scraper and a CLI command to seed your DB. This ties into "Automating Data Entry with Python: A Hands-On Guide to Web Scraping".

Example scraper (very simple, hypothetical site):

# app/seed_scraper.py
import requests
from bs4 import BeautifulSoup
from typing import List, Dict
def scrape_books_from_page(url: str) -> List[Dict]:
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    soup = BeautifulSoup(resp.text, "html.parser")
    books = []
    for item in soup.select(".book"):
        title = item.select_one(".title").get_text(strip=True)
        author = item.select_one(".author").get_text(strip=True)
        year_txt = item.select_one(".year").get_text(strip=True)
        try:
            year = int(year_txt)
        except ValueError:
            year = None
        books.append({"title": title, "author": author, "published_year": year})
    return books

Line-by-line:

• requests.get: fetch page with timeout.
• resp.raise_for_status: raise HTTPError for non-2xx responses.
• BeautifulSoup: parse HTML.
• CSS selectors to extract book data.
• Cast year to int when possible.

CLI to seed DB with Click

Integrate scraping into a command-line tool using Click — related to "Creating a Command-Line Interface with Click: Best Practices and Tips".

# app/cli.py
import click
from .database import SessionLocal, engine, Base
from .crud import create_book
from .seed_scraper import scrape_books_from_page
@click.group()
def cli():
    pass
@cli.command()
@click.argument("url")
def seed(url):
    """Seed the DB with books scraped from URL."""
    books = scrape_books_from_page(url)
    db = SessionLocal()
    try:
        for b in books:
            create_book(db, b)  # create_book accepts Pydantic-like dict
        click.echo(f"Seeded {len(books)} books")
    finally:
        db.close()
@cli.command()
def initdb():
    Base.metadata.create_all(bind=engine)
    click.echo("Initialized the database")

Line-by-line:

• click.group: CLI group entrypoint.
• seed command: accepts a URL, scrapes books, and uses CRUD to insert them.
• initdb: convenience command to create tables.

Run CLI:

python -m app.cli seed "https://example.com/books"
python -m app.cli initdb

Deploying and Microservices Context

If you're building a microservice, you might containerize your FastAPI app. This guide touches how this differs from "Building Microservices with Flask and Docker: A Practical Approach". While Flask is a great microservice framework, FastAPI is often preferred for async performance and automatic docs.

A simple Dockerfile:

# Dockerfile
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY ./app /app/app
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80", "--workers", "1"]

Notes:

• For high concurrency, tune workers and use Gunicorn + Uvicorn workers.
• Use environment variables for DB URL and secrets instead of hardcoding.
• In microservices, separate responsibilities: one service for books, another for authors, etc.

version: '3.8'
services:
  api:
    build: .
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/appdb
    depends_on:
      - db
    ports:
      - "8000:80"
  db:
    image: postgres:14
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: appdb
    volumes:
      - pgdata:/var/lib/postgresql/data
volumes:
  pgdata:

Best Practices

• Use Alembic for schema migrations — never rely solely on create_all in production.
• Use environment variables and a 12-factor approach for config.
• Keep business logic out of routes; put it in services or CRUD modules.
• Validate inputs with Pydantic and set limits for pagination.
• Use connection pooling and tune pool_size when using SQL databases.
• Implement proper logging and structured error handling.
• Add tests: unit tests for CRUD functions and integration tests for endpoints.

Common Pitfalls

• Connection leaks: Always close sessions or use context managers. The get_db dependency pattern avoids leaks.
• Mixing async route functions with sync DB calls: If you mark a route async def and perform blocking DB calls, you risk blocking the event loop. Use synchronous routes or offload DB operations to threadpool, or use async DB drivers.
• Not handling DB integrity errors: Wrap commits in try/except to translate DB errors to HTTP responses.
• Blindly crawling websites: When scraping to seed data, respect robots.txt and terms of service.

Advanced Tips

• Async SQLAlchemy (1.4+): If you need full async DB access, explore SQLAlchemy's async ORM and async drivers (asyncpg for Postgres). Requires different Session and engine patterns.
• Pagination strategies: Cursor-based pagination scales better than offset-based for large tables.
• Caching: Add response caching (Redis) for read-heavy endpoints.
• Background tasks: Use FastAPI's BackgroundTasks or Celery for long-running imports or expensive scraping jobs.
• Authentication & Authorization: Add OAuth2 with JWT tokens or integrate with API gateways.
• Observability: Add OpenTelemetry, structured logs, and metrics.

Error Handling Example

Catch DB integrity errors and return friendly messages:

from fastapi import Request
from fastapi.responses import JSONResponse
from sqlalchemy.exc import IntegrityError
@app.exception_handler(IntegrityError)
def db_integrity_error_handler(request: Request, exc: IntegrityError):
    return JSONResponse(
        status_code=400,
        content={"detail": "Database integrity error", "error": str(exc.orig)}
    )

This returns a 400 with a concise message. In production, avoid leaking DB internals in error messages.

Testing

• Unit test CRUD functions using a SQLite in-memory DB: sqlite:///:memory: and create tables in setup.
• Use FastAPI's TestClient (starlette.testclient) to test routes.

Conclusion