🎮 Biblioteca de Jogos 01 (Python) — CRUD Básico

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    APP_NAME: str = "Biblioteca de Jogos"
    DATABASE_URL: str = "sqlite:///./bibliotecajogos.db"
    ENVIRONMENT: str = "dev"

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore"
    )

settings = Settings()

from typing import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker, Session
from app.config import settings

db_url = settings.DATABASE_URL
if db_url.startswith("postgres://"):
    db_url = db_url.replace("postgres://", "postgresql+psycopg://", 1)
elif db_url.startswith("postgresql://") and not db_url.startswith("postgresql+"):
    db_url = db_url.replace("postgresql://", "postgresql+psycopg://", 1)

connect_args = {"check_same_thread": False} if db_url.startswith("sqlite") else {}

engine = create_engine(db_url, connect_args=connect_args)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

from sqlalchemy import String, Boolean
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base

class Jogo(Base):
    __tablename__ = "jogos"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    titulo: Mapped[str] = mapped_column(String(255), nullable=False)
    plataforma: Mapped[str] = mapped_column(String(100), nullable=False)
    concluido: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)

from pydantic import BaseModel, ConfigDict, Field

class JogoBase(BaseModel):
    titulo: str = Field(..., min_length=1, max_length=255, description="Título do jogo")
    plataforma: str = Field(..., min_length=1, max_length=100, description="Plataforma (ex: PC, PS5, Xbox, Switch)")
    concluido: bool = Field(default=False, description="Indica se o jogo foi concluído/zerado")

class JogoCreate(JogoBase):
    pass

class JogoUpdate(BaseModel):
    titulo: str | None = Field(default=None, min_length=1, max_length=255)
    plataforma: str | None = Field(default=None, min_length=1, max_length=100)
    concluido: bool | None = Field(default=None)

class JogoResponse(JogoBase):
    id: int

    model_config = ConfigDict(from_attributes=True)

from pathlib import Path
from typing import Annotated
from fastapi import APIRouter, Depends, Form, Request, HTTPException, status
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session

from app.database import get_db
from app.models import Jogo
from app.schemas import JogoCreate, JogoUpdate, JogoResponse

router = APIRouter()

TEMPLATES_DIR = Path(__file__).resolve().parent / "templates"
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))

# ==========================================
# Rotas Web (Jinja2 + Bootstrap 5)
# ==========================================

@router.get("/", response_class=HTMLResponse, include_in_schema=False)
def listar_web(request: Request, db: Session = Depends(get_db)):
    jogos = db.query(Jogo).order_by(Jogo.id.asc()).all()
    return templates.TemplateResponse(request=request, name="jogos.html", context={"jogos": jogos})

@router.post("/jogos", response_class=RedirectResponse, status_code=status.HTTP_303_SEE_OTHER, tags=["Web"], summary="Adicionar jogo via Formulário Web")
def adicionar_web(
    titulo: Annotated[str, Form(description="Título do jogo")],
    plataforma: Annotated[str, Form(description="Plataforma do jogo")],
    db: Session = Depends(get_db),
):
    novo_jogo = Jogo(titulo=titulo, plataforma=plataforma, concluido=False)
    db.add(novo_jogo)
    db.commit()
    return RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)

@router.post("/jogos/{id}/concluir", response_class=RedirectResponse, status_code=status.HTTP_303_SEE_OTHER, tags=["Web"], summary="Alternar status do jogo via Formulário Web")
def alternar_status_web(id: int, db: Session = Depends(get_db)):
    jogo = db.query(Jogo).filter(Jogo.id == id).first()
    if jogo:
        jogo.concluido = not jogo.concluido
        db.commit()
    return RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)

@router.post("/jogos/{id}/apagar", response_class=RedirectResponse, status_code=status.HTTP_303_SEE_OTHER, tags=["Web"], summary="Apagar jogo via Formulário Web")
def apagar_web(id: int, db: Session = Depends(get_db)):
    jogo = db.query(Jogo).filter(Jogo.id == id).first()
    if jogo:
        db.delete(jogo)
        db.commit()
    return RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)

# ==========================================
# Endpoints REST API (Swagger / JSON)
# ==========================================

@router.get("/api/jogos", response_model=list[JogoResponse], tags=["API REST - Jogos"], summary="Listar todos os jogos")
def api_listar_jogos(db: Session = Depends(get_db)):
    return db.query(Jogo).order_by(Jogo.id.asc()).all()

@router.get("/api/jogos/{id}", response_model=JogoResponse, tags=["API REST - Jogos"], summary="Buscar jogo por ID")
def api_buscar_jogo(id: int, db: Session = Depends(get_db)):
    jogo = db.query(Jogo).filter(Jogo.id == id).first()
    if not jogo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Jogo com ID {id} não encontrado",
        )
    return jogo

@router.post("/api/jogos", response_model=JogoResponse, status_code=status.HTTP_201_CREATED, tags=["API REST - Jogos"], summary="Cadastrar novo jogo")
def api_criar_jogo(jogo_in: JogoCreate, db: Session = Depends(get_db)):
    novo_jogo = Jogo(
        titulo=jogo_in.titulo,
        plataforma=jogo_in.plataforma,
        concluido=jogo_in.concluido,
    )
    db.add(novo_jogo)
    db.commit()
    db.refresh(novo_jogo)
    return novo_jogo

@router.put("/api/jogos/{id}", response_model=JogoResponse, tags=["API REST - Jogos"], summary="Atualizar jogo por ID")
def api_atualizar_jogo(id: int, jogo_in: JogoUpdate, db: Session = Depends(get_db)):
    jogo = db.query(Jogo).filter(Jogo.id == id).first()
    if not jogo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Jogo com ID {id} não encontrado",
        )
    
    if jogo_in.titulo is not None:
        jogo.titulo = jogo_in.titulo
    if jogo_in.plataforma is not None:
        jogo.plataforma = jogo_in.plataforma
    if jogo_in.concluido is not None:
        jogo.concluido = jogo_in.concluido

    db.commit()
    db.refresh(jogo)
    return jogo

@router.patch("/api/jogos/{id}/concluir", response_model=JogoResponse, tags=["API REST - Jogos"], summary="Alternar status zerado/jogando")
def api_alternar_status(id: int, db: Session = Depends(get_db)):
    jogo = db.query(Jogo).filter(Jogo.id == id).first()
    if not jogo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Jogo com ID {id} não encontrado",
        )
    
    jogo.concluido = not jogo.concluido
    db.commit()
    db.refresh(jogo)
    return jogo

@router.delete("/api/jogos/{id}", status_code=status.HTTP_204_NO_CONTENT, tags=["API REST - Jogos"], summary="Excluir jogo por ID")
def api_deletar_jogo(id: int, db: Session = Depends(get_db)):
    jogo = db.query(Jogo).filter(Jogo.id == id).first()
    if not jogo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Jogo com ID {id} não encontrado",
        )
    db.delete(jogo)
    db.commit()

from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.config import settings
from app.database import Base, engine
from app.routes import router

@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(bind=engine)
    yield

app = FastAPI(
    title="Biblioteca de Jogos API",
    description="Aplicação Web e API REST de Catálogo de Jogos em Python 3.11+, FastAPI e SQLAlchemy 2.0",
    version="1.0.0",
    lifespan=lifespan,
)

app.include_router(router)

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.main import app
from app.database import Base, get_db

SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture(scope="function")
def db_session():
    Base.metadata.create_all(bind=engine)
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
        Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def client(db_session):
    def override_get_db():
        try:
            yield db_session
        finally:
            pass

    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app, follow_redirects=True) as test_client:
        yield test_client
    app.dependency_overrides.clear()

def test_web_listar_jogos_vazio(client):
    response = client.get("/")
    assert response.status_code == 200
    assert "Nenhum jogo cadastrado" in response.text

def test_web_fluxo_completo(client):
    # 1. Adicionar jogo via form
    response = client.post("/jogos", data={"titulo": "Super Mario Odyssey", "plataforma": "Nintendo Switch"})
    assert response.status_code == 200
    assert "Super Mario Odyssey" in response.text
    assert "Jogando" in response.text

    # 2. Alternar para Zerado
    response_concluir = client.post("/jogos/1/concluir")
    assert response_concluir.status_code == 200
    assert "Zerado" in response_concluir.text

    # 3. Apagar jogo
    response_apagar = client.post("/jogos/1/apagar")
    assert response_apagar.status_code == 200
    assert "Super Mario Odyssey" not in response_apagar.text

def test_api_rest_completa(client):
    # 1. Listar vazio
    resp = client.get("/api/jogos")
    assert resp.status_code == 200
    assert resp.json() == []

    # 2. Criar via POST JSON
    post_resp = client.post("/api/jogos", json={"titulo": "God of War Ragnarok", "plataforma": "PS5", "concluido": False})
    assert post_resp.status_code == 201
    jogo_id = post_resp.json()["id"]

    # 3. Buscar por ID
    get_resp = client.get(f"/api/jogos/{jogo_id}")
    assert get_resp.status_code == 200
    assert get_resp.json()["titulo"] == "God of War Ragnarok"

    # 4. Atualizar via PUT
    put_resp = client.put(f"/api/jogos/{jogo_id}", json={"titulo": "God of War (2018)", "plataforma": "PC", "concluido": True})
    assert put_resp.status_code == 200
    assert put_resp.json()["titulo"] == "God of War (2018)"

    # 5. Alternar status via PATCH
    patch_resp = client.patch(f"/api/jogos/{jogo_id}/concluir")
    assert patch_resp.status_code == 200
    assert patch_resp.json()["concluido"] is False

    # 6. Deletar via DELETE
    del_resp = client.delete(f"/api/jogos/{jogo_id}")
    assert del_resp.status_code == 204

    # 7. Buscar deletado (404)
    resp_404 = client.get(f"/api/jogos/{jogo_id}")
    assert resp_404.status_code == 404


def test_api_nao_encontrado_404(client):
    id_inexistente = 99999
    assert client.get(f"/api/jogos/{id_inexistente}").status_code == 404
    assert client.put(f"/api/jogos/{id_inexistente}", json={"titulo": "X"}).status_code == 404
    assert client.patch(f"/api/jogos/{id_inexistente}/concluir").status_code == 404
    assert client.delete(f"/api/jogos/{id_inexistente}").status_code == 404


def test_api_validacao_campos(client):
    # Título vazio não deve ser aceito (min_length=1)
    resp = client.post("/api/jogos", json={"titulo": "", "plataforma": "PC"})
    assert resp.status_code == 422

    # Plataforma ausente não deve ser aceita
    resp = client.post("/api/jogos", json={"titulo": "Zelda"})
    assert resp.status_code == 422


def test_api_atualizacao_parcial(client):
    post_resp = client.post("/api/jogos", json={"titulo": "Hollow Knight", "plataforma": "PC", "concluido": False})
    jogo_id = post_resp.json()["id"]

    # Atualiza apenas a plataforma
    put_resp = client.put(f"/api/jogos/{jogo_id}", json={"plataforma": "Switch"})
    assert put_resp.status_code == 200
    assert put_resp.json()["titulo"] == "Hollow Knight"
    assert put_resp.json()["plataforma"] == "Switch"


def test_web_acoes_jogo_inexistente(client):
    resp_concluir = client.post("/jogos/99999/concluir")
    assert resp_concluir.status_code == 200

    resp_apagar = client.post("/jogos/99999/apagar")
    assert resp_apagar.status_code == 200