A partir de um banco series.db criado, o script abaixo descreve as rotas /series/get/, /series/get/{nome-da-serie} e /series/post:
import sqlite3
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
app = FastAPI(title="Catálogo de Séries - SQLite Edition")
class Serie(BaseModel):
titulo: str = Field(..., min_length=1)
genero: str = Field(..., min_length=1)
ano_lancamento: int = Field(..., gt=1900)
temporadas: int = Field(..., gt=0)
def get_db_connection():
conn = sqlite3.connect('series.db')
conn.row_factory = sqlite3.Row
return conn
@app.post("/series", status_code=201)
async def cadastrar_serie(serie: Serie):
try:
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"INSERT INTO series (titulo, genero, ano_lancamento, temporadas) VALUES (?, ?, ?, ?)",
(serie.titulo, serie.genero, serie.ano_lancamento, serie.temporadas)
)
conn.commit()
return {"mensagem": "Série cadastrada com sucesso!", "serie": serie}
except sqlite3.IntegrityError:
raise HTTPException(status_code=400, detail="Esta série já está cadastrada.")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Erro no banco: {str(e)}")
@app.get("/series/{titulo}", response_model=Serie)
async def buscar_serie(titulo: str):
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM series WHERE LOWER(titulo) = LOWER(?)",
(titulo,)
)
resultado = cursor.fetchone()
if resultado is None:
raise HTTPException(status_code=404, detail="Série não encontrada.")
return dict(resultado)
@app.get("/series")
async def listar_todas():
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT * FROM series")
return [dict(row) for row in cursor.fetchall()]
