Tutorial FastAPI + PyMySQL - Sistema de Biblioteca

Estructura del Proyecto

biblioteca_api/
│
├── app/
│ ├── __init__.py
│ ├── [Link]
│ │
│ ├── api/
│ │ ├── __init__.py
│ │ ├── v1/
│ │ │ ├── __init__.py
│ │ │ ├── endpoints/
│ │ │ │ ├── __init__.py
│ │ │ │ ├── [Link]
│ │ │ │ ├── [Link]
│ │ │ │ ├── [Link]
│ │ │ │ ├── [Link]
│ │ ├── dependencies/
│ │ ├── __init__.py
│ │ ├── [Link]
│ │
│ ├── core/
│ │ ├── __init__.py
│ │ ├── [Link]
│ │
│ ├── db/
│ │ ├── __init__.py
│ │ ├── [Link]
│ │ ├── [Link]
│ │ ├── models/
│ │ ├── __init__.py
│ │ ├── [Link]
│ │ ├── [Link]
│ │ ├── [Link]
│ │ ├── [Link]
│ │
│ ├── schemas/
│ │ ├── __init__.py
│ │ ├── [Link]
│ │ ├── [Link]
│ │ ├── [Link]
│ │ ├── [Link]
│ │
│ ├── services/
│ ├── __init__.py
│ ├── book_service.py
│ ├── author_service.py
 │ ├── loan_service.py
│
├── .env
├── [Link]
└── [Link]

1. Instalación

[Link]

txt

fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
pymysql==1.1.0
cryptography==41.0.7
pydantic==2.5.0
pydantic-settings==2.1.0
python-dotenv==1.0.0

Instalar dependencias

bash

pip install -r [Link]

2. Configuración - app/core/[Link]
 python

from pydantic_settings import BaseSettings

from typing import Optional

class Settings(BaseSettings):
"""Configuración de la aplicación"""

# Database
DB_HOST: str = "localhost"
DB_PORT: int = 3306
DB_USER: str = "root"
DB_PASSWORD: str = "password"
DB_NAME: str = "biblioteca_db"

# Application
APP_NAME: str = "Biblioteca API"
API_V1_PREFIX: str = "/api/v1"
DEBUG: bool = True

@property
def database_url(self) -> str:
return f"mysql+pymysql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME

class Config:
env_file = ".env"

settings = Settings()

.env

env

DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=tu_password
DB_NAME=biblioteca_db

APP_NAME=Biblioteca API
API_V1_PREFIX=/api/v1
DEBUG=True

3. Base de Datos - app/db/

app/db/[Link]

python

from sqlalchemy import create_engine

from [Link] import sessionmaker
from [Link] import settings

# Crear engine de SQLAlchemy con PyMySQL

engine = create_engine(
settings.database_url,
pool_pre_ping=True, # Verifica conexiones antes de usarlas
pool_recycle=3600, # Recicla conexiones cada hora
echo=[Link], # Muestra queries SQL en consola
pool_size=5, # Número de conexiones en el pool
max_overflow=10 # Conexiones adicionales permitidas
)

# Session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

app/db/[Link]

python

from [Link] import declarative_base

Base = declarative_base()

# Importar todos los modelos aquí para que SQLAlchemy los reconozca
from [Link] import Author
from [Link] import Book
from [Link] import Member
from [Link] import Loan

4. Modelos - app/db/models/

app/db/models/[Link]
 python

from sqlalchemy import Column, Integer, String, Text, Date

from [Link] import relationship
from [Link] import Base

class Author(Base):
"""
Modelo de Autor
Un autor puede tener muchos libros (1:N)
"""
__tablename__ = "authors"

id = Column(Integer, primary_key=True, index=True)

name = Column(String(200), nullable=False, index=True)
birth_date = Column(Date)
nationality = Column(String(100))
biography = Column(Text)

# Relación 1:N - Un autor tiene muchos libros

books = relationship("Book", back_populates="author", cascade="all, delete-orphan")

app/db/models/[Link]
 python

from sqlalchemy import Column, Integer, String, Text, ForeignKey, Numeric, Boolean
from [Link] import relationship
from [Link] import Base

class Book(Base):
"""
Modelo de Libro
Relaciones:
- N:1 con Author (muchos libros -> un autor)
- 1:N con Loan (un libro -> muchos préstamos)
"""
__tablename__ = "books"

id = Column(Integer, primary_key=True, index=True)

title = Column(String(300), nullable=False, index=True)
isbn = Column(String(13), unique=True, index=True)
description = Column(Text)
published_year = Column(Integer)
pages = Column(Integer)
price = Column(Numeric(10, 2))
stock = Column(Integer, default=0)
available = Column(Boolean, default=True)

# Foreign Key
author_id = Column(Integer, ForeignKey("[Link]", ondelete="CASCADE"), nullable=False)

# Relaciones
author = relationship("Author", back_populates="books")
loans = relationship("Loan", back_populates="book")

app/db/models/[Link]
 python

from sqlalchemy import Column, Integer, String, Date, Boolean, DateTime

from [Link] import relationship
from [Link] import func
from [Link] import Base

class Member(Base):
"""
Modelo de Miembro/Socio de la biblioteca
Un miembro puede tener muchos préstamos (1:N)
"""
__tablename__ = "members"

id = Column(Integer, primary_key=True, index=True)

first_name = Column(String(100), nullable=False)
last_name = Column(String(100), nullable=False)
email = Column(String(255), unique=True, index=True, nullable=False)
phone = Column(String(20))
address = Column(String(300))
membership_date = Column(Date)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=[Link]())

# Relación 1:N - Un miembro tiene muchos préstamos

loans = relationship("Loan", back_populates="member")

app/db/models/[Link]
 python

from sqlalchemy import Column, Integer, ForeignKey, Date, Boolean, Numeric

from [Link] import relationship
from [Link] import Base

class Loan(Base):
"""
Modelo de Préstamo
Relaciones N:1 con Book y Member
"""
__tablename__ = "loans"

id = Column(Integer, primary_key=True, index=True)

# Foreign Keys
book_id = Column(Integer, ForeignKey("[Link]", ondelete="CASCADE"), nullable=False)
member_id = Column(Integer, ForeignKey("[Link]", ondelete="CASCADE"), nullable=False)

# Campos
loan_date = Column(Date, nullable=False)
due_date = Column(Date, nullable=False)
return_date = Column(Date)
is_returned = Column(Boolean, default=False)
fine_amount = Column(Numeric(10, 2), default=0.00)

# Relaciones N:1
book = relationship("Book", back_populates="loans")
member = relationship("Member", back_populates="loans")

5. Schemas - app/schemas/

app/schemas/[Link]
 python

from pydantic import BaseModel, Field

from typing import Optional, List
from datetime import date

# Schema base
class AuthorBase(BaseModel):
name: str = Field(..., min_length=1, max_length=200)
birth_date: Optional[date] = None
nationality: Optional[str] = None
biography: Optional[str] = None

# Para crear
class AuthorCreate(AuthorBase):
pass

# Para actualizar
class AuthorUpdate(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=200)
birth_date: Optional[date] = None
nationality: Optional[str] = None
biography: Optional[str] = None

# Para respuestas
class AuthorResponse(AuthorBase):
id: int

class Config:
from_attributes = True

# Autor con sus libros

class AuthorWithBooks(AuthorResponse):
books: List["BookResponse"] = []

app/schemas/[Link]
 python

from pydantic import BaseModel, Field

from typing import Optional
from decimal import Decimal

class BookBase(BaseModel):
title: str = Field(..., min_length=1, max_length=300)
isbn: Optional[str] = Field(None, max_length=13)
description: Optional[str] = None
published_year: Optional[int] = None
pages: Optional[int] = None
price: Optional[Decimal] = None
stock: int = 0
available: bool = True
author_id: int

class BookCreate(BookBase):
pass

class BookUpdate(BaseModel):
title: Optional[str] = Field(None, min_length=1, max_length=300)
isbn: Optional[str] = None
description: Optional[str] = None
published_year: Optional[int] = None
pages: Optional[int] = None
price: Optional[Decimal] = None
stock: Optional[int] = None
available: Optional[bool] = None
author_id: Optional[int] = None

class BookResponse(BookBase):
id: int

class Config:
from_attributes = True

# Libro con información del autor

class BookWithAuthor(BookResponse):
author: "AuthorResponse"

app/schemas/[Link]
 python

from pydantic import BaseModel, EmailStr, Field

from typing import Optional
from datetime import date, datetime

class MemberBase(BaseModel):
first_name: str = Field(..., min_length=1, max_length=100)
last_name: str = Field(..., min_length=1, max_length=100)
email: EmailStr
phone: Optional[str] = None
address: Optional[str] = None

class MemberCreate(MemberBase):
membership_date: date

class MemberUpdate(BaseModel):
first_name: Optional[str] = None
last_name: Optional[str] = None
email: Optional[EmailStr] = None
phone: Optional[str] = None
address: Optional[str] = None
is_active: Optional[bool] = None

class MemberResponse(MemberBase):
id: int
membership_date: date
is_active: bool
created_at: datetime

class Config:
from_attributes = True

app/schemas/[Link]
 python

from pydantic import BaseModel

from typing import Optional
from datetime import date
from decimal import Decimal

class LoanBase(BaseModel):
book_id: int
member_id: int
loan_date: date
due_date: date

class LoanCreate(LoanBase):
pass

class LoanUpdate(BaseModel):
return_date: Optional[date] = None
is_returned: Optional[bool] = None
fine_amount: Optional[Decimal] = None

class LoanResponse(LoanBase):
id: int
return_date: Optional[date] = None
is_returned: bool
fine_amount: Decimal

class Config:
from_attributes = True

# Préstamo con relaciones completas

class LoanWithDetails(LoanResponse):
book: "BookResponse"
member: "MemberResponse"

6. Dependency - app/api/dependencies/[Link]
 python

from typing import Generator

from [Link] import Session
from [Link] import SessionLocal

def get_db() -> Generator[Session, None, None]:

"""
Dependency que proporciona una sesión de base de datos.
Se cierra automáticamente después de cada request.
"""
db = SessionLocal()
try:
yield db
finally:
[Link]()

7. Services - app/services/

app/services/book_service.py
python

from typing import List, Optional

from [Link] import Session, joinedload
from sqlalchemy import func, and_, or_
from [Link] import Book
from [Link] import Author
from [Link] import BookCreate, BookUpdate
from fastapi import HTTPException, status

class BookService:
"""Servicio para operaciones con libros"""

@staticmethod
def get_all(db: Session, skip: int = 0, limit: int = 100) -> List[Book]:
"""Obtener todos los libros con paginación"""
return [Link](Book).offset(skip).limit(limit).all()

@staticmethod
def get_by_id(db: Session, book_id: int) -> Book:
"""Obtener libro por ID"""
book = [Link](Book).filter([Link] == book_id).first()
if not book:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Libro no encontrado"
)
return book

@staticmethod
def get_with_author(db: Session, book_id: int) -> Book:
"""
Obtener libro con información del autor
Usa joinedload para evitar N+1 queries
"""
book = [Link](Book)\
.options(joinedload([Link]))\
.filter([Link] == book_id)\
.first()

if not book:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Libro no encontrado"
)
return book
@staticmethod
def create(db: Session, book_data: BookCreate) -> Book:
"""Crear nuevo libro"""
# Verificar que el autor existe
author = [Link](Author).filter([Link] == book_data.author_id).first()
if not author:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Autor no encontrado"
)

# Verificar ISBN único

if book_data.isbn:
existing = [Link](Book).filter([Link] == book_data.isbn).first()
if existing:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Ya existe un libro con este ISBN"
)

db_book = Book(**book_data.model_dump())
[Link](db_book)
[Link]()
[Link](db_book)
return db_book

@staticmethod
def update(db: Session, book_id: int, book_data: BookUpdate) -> Book:
"""Actualizar libro"""
book = BookService.get_by_id(db, book_id)

update_data = book_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(book, field, value)

[Link]()
[Link](book)
return book

@staticmethod
def delete(db: Session, book_id: int) -> dict:
"""Eliminar libro"""
book = BookService.get_by_id(db, book_id)
[Link](book)
[Link]()
return {"message": "Libro eliminado exitosamente"}
# ========== CONSULTAS AVANZADAS ==========

@staticmethod
def search_by_title(db: Session, search: str) -> List[Book]:
"""
Búsqueda parcial por título (LIKE)
"""
return [Link](Book)\
.filter([Link](f"%{search}%"))\
.all()

@staticmethod
def get_available_books(db: Session) -> List[Book]:
"""Obtener solo libros disponibles"""
return [Link](Book)\
.filter([Link] == True, [Link] > 0)\
.all()

@staticmethod
def get_by_author(db: Session, author_id: int) -> List[Book]:
"""
Obtener todos los libros de un autor
Query con JOIN implícito por FK
"""
return [Link](Book)\
.filter(Book.author_id == author_id)\
.order_by(Book.published_year.desc())\
.all()

@staticmethod
def get_by_year_range(db: Session, start_year: int, end_year: int) -> List[Book]:
"""
Libros publicados en un rango de años (BETWEEN)
"""
return [Link](Book)\
.filter(Book.published_year.between(start_year, end_year))\
.all()

@staticmethod
def get_by_price_range(db: Session, min_price: float, max_price: float) -> List[Book]:
"""
Libros en un rango de precio
"""
return [Link](Book)\
.filter(and_(
 [Link] >= min_price,
[Link] <= max_price
))\
.all()

@staticmethod
def get_expensive_books(db: Session, min_price: float = 50) -> List[Book]:
"""
Libros con precio mayor a un valor (operador >)
"""
return [Link](Book)\
.filter([Link] > min_price)\
.order_by([Link]())\
.all()

@staticmethod
def count_by_author(db: Session, author_id: int) -> int:
"""
Contar libros de un autor (COUNT)
"""
return [Link]([Link]([Link]))\
.filter(Book.author_id == author_id)\
.scalar()

@staticmethod
def get_statistics(db: Session) -> dict:
"""
Estadísticas generales (COUNT, AVG, MAX, MIN, SUM)
"""
stats = [Link](
[Link]([Link]).label('total_books'),
[Link]([Link]).label('avg_price'),
[Link]([Link]).label('max_price'),
[Link]([Link]).label('min_price'),
[Link]([Link]).label('total_stock')
).first()

return {
"total_books": stats.total_books,
"average_price": float(stats.avg_price) if stats.avg_price else 0,
"max_price": float(stats.max_price) if stats.max_price else 0,
"min_price": float(stats.min_price) if stats.min_price else 0,
"total_stock": stats.total_stock
}

@staticmethod
 > List[dict]:
def get_books_with_author_info(db: Session) ->
"""
JOIN explícito - Libros con nombre del autor
"""
results = [Link](
[Link],
[Link],
[Link],
[Link],
[Link]('author_name'),
[Link]
)\
.join(Author, Book.author_id == [Link])\
.all()

return [
{
"id": [Link],
"title": [Link],
"isbn": [Link],
"price": float([Link]) if [Link] else 0,
"author_name": r.author_name,
"nationality": [Link]
}
for r in results
]

@staticmethod
def get_books_grouped_by_year(db: Session):
"""
GROUP BY - Libros agrupados por año de publicación
"""
results = [Link](
Book.published_year,
[Link]([Link]).label('count'),
[Link]([Link]).label('avg_price')
)\
.group_by(Book.published_year)\
.order_by(Book.published_year.desc())\
.all()

return [
{
"year": r.published_year,
"book_count": [Link],
"average_price": float(r.avg_price) if r.avg_price else 0
}
 for r in results
]

@staticmethod
def complex_search(
db: Session,
title: Optional[str] = None,
author_id: Optional[int] = None,
min_price: Optional[float] = None,
available_only: bool = False
) -> List[Book]:
"""
Búsqueda compleja con múltiples filtros opcionales (AND/OR)
"""
query = [Link](Book)

filters = []

if title:
[Link]([Link](f"%{title}%"))

if author_id:
[Link](Book.author_id == author_id)

if min_price:
[Link]([Link] >= min_price)

if available_only:
[Link](and_([Link] == True, [Link] > 0))

if filters:
query = [Link](and_(*filters))

return [Link]()

@staticmethod
def get_books_with_stock_alert(db: Session, min_stock: int = 5) -> List[Book]:
"""
Libros con stock bajo (útil para reposición)
"""
return [Link](Book)\
.filter(and_(
[Link] == True,
[Link] < min_stock,
[Link] > 0
))\
 .all()

@staticmethod
def search_books_by_author_name(db: Session, author_name: str) -> List[Book]:
"""
Buscar libros por nombre del autor (JOIN con búsqueda)
"""
return [Link](Book)\
.join(Author)\
.filter([Link](f"%{author_name}%"))\
.all()

app/services/author_service.py
python

from typing import List

from [Link] import Session, joinedload
from sqlalchemy import func
from [Link] import Author
from [Link] import Book
from [Link] import AuthorCreate, AuthorUpdate
from fastapi import HTTPException, status

class AuthorService:
"""Servicio para operaciones con autores"""

@staticmethod
def get_all(db: Session, skip: int = 0, limit: int = 100) -> List[Author]:
"""Obtener todos los autores"""
return [Link](Author).offset(skip).limit(limit).all()

@staticmethod
def get_by_id(db: Session, author_id: int) -> Author:
"""Obtener autor por ID"""
author = [Link](Author).filter([Link] == author_id).first()
if not author:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Autor no encontrado"
)
return author

@staticmethod
def get_with_books(db: Session, author_id: int) -> Author:
"""
Obtener autor con todos sus libros (eager loading)
Evita el problema N+1
"""
author = [Link](Author)\
.options(joinedload([Link]))\
.filter([Link] == author_id)\
.first()

if not author:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Autor no encontrado"
)
return author
@staticmethod
def create(db: Session, author_data: AuthorCreate) -> Author:
"""Crear nuevo autor"""
db_author = Author(**author_data.model_dump())
[Link](db_author)
[Link]()
[Link](db_author)
return db_author

@staticmethod
def update(db: Session, author_id: int, author_data: AuthorUpdate) -> Author:
"""Actualizar autor"""
author = AuthorService.get_by_id(db, author_id)

update_data = author_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(author, field, value)

[Link]()
[Link](author)
return author

@staticmethod
def delete(db: Session, author_id: int) -> dict:
"""Eliminar autor (también elimina sus libros por cascade)"""
author = AuthorService.get_by_id(db, author_id)
[Link](author)
[Link]()
return {"message": "Autor eliminado exitosamente"}

# ========== CONSULTAS AVANZADAS ==========

@staticmethod
def search_by_name(db: Session, name: str) -> List[Author]:
"""Búsqueda parcial por nombre"""
return [Link](Author)\
.filter([Link](f"%{name}%"))\
.all()

@staticmethod
def get_by_nationality(db: Session, nationality: str) -> List[Author]:
"""Filtrar autores por nacionalidad"""
return [Link](Author)\
.filter([Link] == nationality)\
.all()
@staticmethod
def get_authors_with_book_count(db: Session):
"""
Autores con el número de libros que han escrito
LEFT JOIN + GROUP BY + COUNT
"""
results = [Link](
[Link],
[Link],
[Link],
[Link]([Link]).label('book_count')
)\
.outerjoin(Book, [Link] == Book.author_id)\
.group_by([Link])\
.order_by([Link]([Link]).desc())\
.all()

return [
{
"id": [Link],
"name": [Link],
"nationality": [Link],
"book_count": r.book_count
}
for r in results
]

@staticmethod
def get_most_prolific_authors(db: Session, limit: int = 10):
"""
Top autores por número de libros (HAVING)
"""
results = [Link](
[Link],
[Link],
[Link]([Link]).label('book_count')
)\
.join(Book)\
.group_by([Link])\
.having([Link]([Link]) > 0)\
.order_by([Link]([Link]).desc())\
.limit(limit)\
.all()

return [
{
 "author_id": [Link],
"author_name": [Link],
"total_books": r.book_count
}
for r in results
]

app/services/loan_service.py
python

from typing import List

from datetime import date, datetime, timedelta
from [Link] import Session, joinedload
from sqlalchemy import and_, func
from [Link] import Loan
from [Link] import Book
from [Link] import Member
from [Link] import LoanCreate, LoanUpdate
from fastapi import HTTPException, status

class LoanService:
"""Servicio para gestión de préstamos"""

@staticmethod
def create_loan(db: Session, loan_data: LoanCreate) -> Loan:
"""Crear nuevo préstamo"""
# Verificar que el libro existe y está disponible
book = [Link](Book).filter([Link] == loan_data.book_id).first()
if not book:
raise HTTPException(status_code=404, detail="Libro no encontrado")

if not [Link] or [Link] <= 0:

raise HTTPException(status_code=400, detail="Libro no disponible")

# Verificar que el miembro existe y está activo

member = [Link](Member).filter([Link] == loan_data.member_id).first()
if not member:
raise HTTPException(status_code=404, detail="Miembro no encontrado")

if not member.is_active:
raise HTTPException(status_code=400, detail="Miembro inactivo")

# Crear préstamo
db_loan = Loan(**loan_data.model_dump())
[Link](db_loan)

# Reducir stock del libro

[Link] -= 1
if [Link] == 0:
[Link] = False

[Link]()
[Link](db_loan)
return db_loan
@staticmethod
def return_book(db: Session, loan_id: int) -> Loan:
"""Procesar devolución de libro"""
loan = [Link](Loan).filter([Link] == loan_id).first()
if not loan:
raise HTTPException(status_code=404, detail="Préstamo no encontrado")

if loan.is_returned:
raise HTTPException(status_code=400, detail="Libro ya devuelto")

# Actualizar préstamo
loan.return_date = [Link]()
loan.is_returned = True

# Calcular multa si hay retraso

if loan.return_date > loan.due_date:
days_late = (loan.return_date - loan.due_date).days
loan.fine_amount = days_late * 1.5 # $1.5 por día

# Aumentar stock del libro

book = [Link](Book).filter([Link] == loan.book_id).first()
[Link] += 1
[Link] = True

[Link]()
[Link](loan)
return loan

@staticmethod
def get_active_loans(db: Session) -> List[Loan]:
"""Obtener préstamos activos (no devueltos)"""
return [Link](Loan)\
.filter(Loan.is_returned == False)\
.options(joinedload([Link]), joinedload([Link]))\
.all()

@staticmethod
def get_overdue_loans(db: Session) -> List[Loan]:
"""
Préstamos vencidos (fecha límite pasada y no devueltos)
"""
today = [Link]()
return [Link](Loan)\
.filter(and_(
Loan.is_returned == False,
Loan.due_date < today
 ))\
.options(joinedload([Link]), joinedload([Link]))\
.all()

@staticmethod
def get_member_loans(db: Session, member_id: int) -> List[Loan]:
"""Historial de préstamos de un miembro"""
return [Link](Loan)\
.filter(Loan.member_id == member_id)\
.order_by(Loan.loan_date.desc())\
.all()

@staticmethod
def get_loan_statistics(db: Session) -> dict:
"""
Estadísticas de préstamos (SUBQUERIES y AGREGACIONES)
"""
total = [Link]([Link]([Link])).scalar()
active = [Link]([Link]([Link]))\
.filter(Loan.is_returned == False).scalar()

returned = [Link]([Link]([Link]))\
.filter(Loan.is_returned == True).scalar()

overdue = [Link]([Link]([Link]))\
.filter(and_(
Loan.is_returned == False,
Loan.due_date < [Link]()
)).scalar()

total_fines = [Link]([Link](Loan.fine_amount)).scalar() or 0

return {
"total_loans": total,
"active_loans": active,
"returned_loans": returned,
"overdue_loans": overdue,
"total_fines": float(total_fines)
}

@staticmethod
def get_most_borrowed_books(db: Session, limit: int = 10):
"""
Libros más prestados (JOIN + GROUP BY + ORDER BY)
"""
results = [Link](
 [Link],
[Link],
[Link]([Link]).label('times_borrowed')
)\
.join(Loan, [Link] == Loan.book_id)\
.group_by([Link])\
.order_by([Link]([Link]).desc())\
.limit(limit)\
.all()

return [
{
"book_id": [Link],
"title": [Link],
"times_borrowed": r.times_borrowed
}
for r in results
]

@staticmethod
def get_most_active_members(db: Session, limit: int = 10):
"""
Miembros más activos (con más préstamos)
"""
results = [Link](
[Link],
Member.first_name,
Member.last_name,
[Link],
[Link]([Link]).label('total_loans')
)\
.join(Loan, [Link] == Loan.member_id)\
.group_by([Link])\
.order_by([Link]([Link]).desc())\
.limit(limit)\
.all()

return [
{
"member_id": [Link],
"name": f"{r.first_name} {r.last_name}",
"email": [Link],
"total_loans": r.total_loans
}
for r in results
]
app/api/v1/endpoints/[Link]
8. Endpoints - app/api/v1/endpoints/
python

from typing import List

from fastapi import APIRouter, Depends, Query, status
from [Link] import Session
from [Link] import get_db
from [Link].book_service import BookService
from [Link] import BookCreate, BookUpdate, BookResponse, BookWithAuthor

router = APIRouter(prefix="/books", tags=["Books"])

@[Link]("/", response_model=List[BookResponse])
def get_books(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db)
):
"""Obtener todos los libros con paginación"""
return BookService.get_all(db, skip=skip, limit=limit)

@[Link]("/search", response_model=List[BookResponse])
def search_books(
title: str = Query(..., min_length=1),
db: Session = Depends(get_db)
):
"""Buscar libros por título (LIKE)"""
return BookService.search_by_title(db, title)

@[Link]("/available", response_model=List[BookResponse])
def get_available_books(db: Session = Depends(get_db)):
"""Obtener solo libros disponibles"""
return BookService.get_available_books(db)

@[Link]("/expensive", response_model=List[BookResponse])
def get_expensive_books(
min_price: float = Query(50, gt=0),
db: Session = Depends(get_db)
):
"""Libros con precio mayor a un valor"""
return BookService.get_expensive_books(db, min_price)

@[Link]("/price-range", response_model=List[BookResponse])
def get_books_by_price(
min_price: float = Query(..., ge=0),
max_price: float = Query(..., gt=0),
db: Session = Depends(get_db)
):
"""Libros en un rango de precio (BETWEEN)"""
return BookService.get_by_price_range(db, min_price, max_price)

@[Link]("/year-range", response_model=List[BookResponse])
def get_books_by_year(
start_year: int,
end_year: int,
db: Session = Depends(get_db)
):
"""Libros publicados en un rango de años"""
return BookService.get_by_year_range(db, start_year, end_year)

@[Link]("/statistics")
def get_book_statistics(db: Session = Depends(get_db)):
"""Estadísticas generales (COUNT, AVG, MAX, MIN, SUM)"""
return BookService.get_statistics(db)

@[Link]("/with-authors")
def get_books_with_authors(db: Session = Depends(get_db)):
"""Libros con información del autor (JOIN)"""
return BookService.get_books_with_author_info(db)

@[Link]("/grouped-by-year")
def get_books_grouped(db: Session = Depends(get_db)):
"""Libros agrupados por año (GROUP BY)"""
return BookService.get_books_grouped_by_year(db)

@[Link]("/stock-alert")
def get_stock_alerts(
min_stock: int = Query(5, ge=1),
db: Session = Depends(get_db)
):
"""Libros con stock bajo"""
return BookService.get_books_with_stock_alert(db, min_stock)

@[Link]("/complex-search", response_model=List[BookResponse])
def complex_search(
title: str = None,
author_id: int = None,
min_price: float = None,
available_only: bool = False,
db: Session = Depends(get_db)
):
"""Búsqueda compleja con múltiples filtros"""
return BookService.complex_search(
db, title, author_id, min_price, available_only
 )

@[Link]("/{book_id}", response_model=BookWithAuthor)
def get_book(book_id: int, db: Session = Depends(get_db)):
"""Obtener libro por ID con autor"""
return BookService.get_with_author(db, book_id)

@[Link]("/", response_model=BookResponse, status_code=status.HTTP_201_CREATED)

def create_book(book: BookCreate, db: Session = Depends(get_db)):
"""Crear nuevo libro"""
return [Link](db, book)

@[Link]("/{book_id}", response_model=BookResponse)
def update_book(
book_id: int,
book: BookUpdate,
db: Session = Depends(get_db)
):
"""Actualizar libro"""
return [Link](db, book_id, book)

@[Link]("/{book_id}")
def delete_book(book_id: int, db: Session = Depends(get_db)):
"""Eliminar libro"""
return [Link](db, book_id)

@[Link]("/author/{author_id}", response_model=List[BookResponse])
def get_books_by_author(author_id: int, db: Session = Depends(get_db)):
"""Obtener todos los libros de un autor"""
return BookService.get_by_author(db, author_id)

@[Link]("/author-name/search", response_model=List[BookResponse])
def search_by_author_name(
author_name: str = Query(..., min_length=1),
db: Session = Depends(get_db)
):
"""Buscar libros por nombre del autor"""
return BookService.search_books_by_author_name(db, author_name)

app/api/v1/endpoints/[Link]
python

from typing import List

from fastapi import APIRouter, Depends, Query, status
from [Link] import Session
from [Link] import get_db
from [Link].author_service import AuthorService
from [Link] import AuthorCreate, AuthorUpdate, AuthorResponse, AuthorWithBooks

router = APIRouter(prefix="/authors", tags=["Authors"])

@[Link]("/", response_model=List[AuthorResponse])
def get_authors(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db)
):
"""Obtener todos los autores"""
return AuthorService.get_all(db, skip, limit)

@[Link]("/search", response_model=List[AuthorResponse])
def search_authors(
name: str = Query(..., min_length=1),
db: Session = Depends(get_db)
):
"""Buscar autores por nombre"""
return AuthorService.search_by_name(db, name)

@[Link]("/nationality/{nationality}", response_model=List[AuthorResponse])
def get_by_nationality(
nationality: str,
db: Session = Depends(get_db)
):
"""Filtrar autores por nacionalidad"""
return AuthorService.get_by_nationality(db, nationality)

@[Link]("/with-book-count")
def get_authors_with_counts(db: Session = Depends(get_db)):
"""Autores con cantidad de libros (LEFT JOIN + GROUP BY)"""
return AuthorService.get_authors_with_book_count(db)

@[Link]("/most-prolific")
def get_most_prolific(
limit: int = Query(10, ge=1, le=50),
db: Session = Depends(get_db)
):
 """Top autores por número de libros"""
return AuthorService.get_most_prolific_authors(db, limit)

@[Link]("/{author_id}", response_model=AuthorResponse)
def get_author(author_id: int, db: Session = Depends(get_db)):
"""Obtener autor por ID"""
return AuthorService.get_by_id(db, author_id)

@[Link]("/{author_id}/books", response_model=AuthorWithBooks)
def get_author_with_books(author_id: int, db: Session = Depends(get_db)):
"""Obtener autor con todos sus libros"""
return AuthorService.get_with_books(db, author_id)

@[Link]("/", response_model=AuthorResponse, status_code=status.HTTP_201_CREATED)

def create_author(author: AuthorCreate, db: Session = Depends(get_db)):
"""Crear nuevo autor"""
return [Link](db, author)

@[Link]("/{author_id}", response_model=AuthorResponse)
def update_author(
author_id: int,
author: AuthorUpdate,
db: Session = Depends(get_db)
):
"""Actualizar autor"""
return [Link](db, author_id, author)

@[Link]("/{author_id}")
def delete_author(author_id: int, db: Session = Depends(get_db)):
"""Eliminar autor"""
return [Link](db, author_id)

app/api/v1/endpoints/[Link]
python

from typing import List

from fastapi import APIRouter, Depends, Query, status
from [Link] import Session
from [Link] import get_db
from [Link].loan_service import LoanService
from [Link] import LoanCreate, LoanResponse, LoanWithDetails

router = APIRouter(prefix="/loans", tags=["Loans"])

@[Link]("/", response_model=LoanResponse, status_code=status.HTTP_201_CREATED)

def create_loan(loan: LoanCreate, db: Session = Depends(get_db)):
"""Crear nuevo préstamo"""
return LoanService.create_loan(db, loan)

@[Link]("/{loan_id}/return", response_model=LoanResponse)
def return_book(loan_id: int, db: Session = Depends(get_db)):
"""Devolver libro prestado"""
return LoanService.return_book(db, loan_id)

@[Link]("/active", response_model=List[LoanWithDetails])
def get_active_loans(db: Session = Depends(get_db)):
"""Obtener préstamos activos"""
return LoanService.get_active_loans(db)

@[Link]("/overdue", response_model=List[LoanWithDetails])
def get_overdue_loans(db: Session = Depends(get_db)):
"""Obtener préstamos vencidos"""
return LoanService.get_overdue_loans(db)

@[Link]("/statistics")
def get_statistics(db: Session = Depends(get_db)):
"""Estadísticas de préstamos"""
return LoanService.get_loan_statistics(db)

@[Link]("/most-borrowed")
def get_most_borrowed(
limit: int = Query(10, ge=1, le=50),
db: Session = Depends(get_db)
):
"""Libros más prestados"""
return LoanService.get_most_borrowed_books(db, limit)

@[Link]("/most-active-members")
def get_active_members(
 limit: int = Query(10, ge=1, le=50),
db: Session = Depends(get_db)
):
"""Miembros más activos"""
return LoanService.get_most_active_members(db, limit)

@[Link]("/member/{member_id}", response_model=List[LoanResponse])
def get_member_loans(member_id: int, db: Session = Depends(get_db)):
"""Historial de préstamos de un miembro"""
return LoanService.get_member_loans(db, member_id)

app/api/v1/endpoints/[Link]
python

from typing import List

from fastapi import APIRouter, Depends, status
from [Link] import Session
from [Link] import get_db
from [Link] import Member
from [Link] import MemberCreate, MemberUpdate, MemberResponse

router = APIRouter(prefix="/members", tags=["Members"])

@[Link]("/", response_model=List[MemberResponse])
def get_members(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db)
):
"""Obtener todos los miembros"""
return [Link](Member).offset(skip).limit(limit).all()

@[Link]("/{member_id}", response_model=MemberResponse)
def get_member(member_id: int, db: Session = Depends(get_db)):
"""Obtener miembro por ID"""
member = [Link](Member).filter([Link] == member_id).first()
if not member:
raise HTTPException(status_code=404, detail="Miembro no encontrado")
return member

@[Link]("/", response_model=MemberResponse, status_code=status.HTTP_201_CREATED)

def create_member(member: MemberCreate, db: Session = Depends(get_db)):
"""Crear nuevo miembro"""
db_member = Member(**member.model_dump())
[Link](db_member)
[Link]()
[Link](db_member)
return db_member

@[Link]("/{member_id}", response_model=MemberResponse)
def update_member(
member_id: int,
member: MemberUpdate,
db: Session = Depends(get_db)
):
"""Actualizar miembro"""
db_member = [Link](Member).filter([Link] == member_id).first()
if not db_member:
 raise HTTPException(status_code=404, detail="Miembro no encontrado")

update_data = member.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(db_member, field, value)

[Link]()
[Link](db_member)
return db_member

app/api/v1/[Link]

python

from fastapi import APIRouter

from [Link] import books, authors, loans, members

api_router = APIRouter()

api_router.include_router([Link])
api_router.include_router([Link])
api_router.include_router([Link])
api_router.include_router([Link])

9. Main Application - app/[Link]

python

from fastapi import FastAPI

from [Link] import CORSMiddleware
from [Link] import settings
from [Link] import Base
from [Link] import engine
from [Link].v1 import api_router

# Crear tablas
[Link].create_all(bind=engine)

# Crear aplicación
app = FastAPI(
title=settings.APP_NAME,
description="API REST para gestión de biblioteca con FastAPI y PyMySQL",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc"
)

# CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

# Incluir routers
app.include_router(api_router, prefix=settings.API_V1_PREFIX)

@[Link]("/")
def root():
return {
"message": "Biblioteca API",
"docs": "/docs",
"version": "1.0.0"
}

@[Link]("/health")
def health_check():
return {"status": "healthy"}

if __name__ == "__main__":
 import uvicorn
[Link]("[Link]:app", host="[Link]", port=8000, reload=True)

10. Ejecutar la Aplicación

bash

# Crear base de datos en MySQL

mysql -u root -p
CREATE DATABASE biblioteca_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

# Ejecutar aplicación
uvicorn [Link]:app --reload

# Acceder a:
# [Link] - Documentación Swagger
# [Link] - Documentación ReDoc

11. Resumen de Consultas SQL Implementadas

Consultas Básicas

• SELECT simple: get_all() , get_by_id()

• WHERE: Filtros por campos

• LIKE: Búsquedas parciales

• BETWEEN: Rangos de valores

• IN: Múltiples valores

• AND/OR: Condiciones múltiples

Consultas Agregadas

• COUNT: Contar registros

• AVG: Promedios

• MAX/MIN: Valores extremos

• SUM: Sumas totales

Consultas con JOIN

 • INNER JOIN: Relaciones directas

• LEFT JOIN: Incluir registros sin relación

• JOIN + GROUP BY: Agrupaciones

• HAVING: Filtros después de agrupar

Consultas Avanzadas

• Subqueries: Consultas anidadas

• DISTINCT: Valores únicos

• ORDER BY: Ordenamiento

• LIMIT/OFFSET: Paginación

¡Tutorial completo!