🚀 Projeto Lista de Tarefas (To-Do List) com Python e FastAPI

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    APP_NAME: str = "API Lista de Tarefas"
    DATABASE_URL: str = "sqlite:///./tarefas.db"
    ENVIRONMENT: str = "dev"
    CORS_ORIGINS: list[str] = ["*"]

    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 Tarefa(Base):
    __tablename__ = "tb_tarefas"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    descricao: Mapped[str] = mapped_column(String(255), nullable=False)
    concluida: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)

from pydantic import BaseModel, ConfigDict, Field

class TarefaBase(BaseModel):
    descricao: str = Field(..., min_length=1, max_length=255, description="Descrição da tarefa")
    concluida: bool = Field(default=False, description="Status de conclusão")

class TarefaCreate(TarefaBase):
    pass

class TarefaUpdate(BaseModel):
    descricao: str | None = Field(default=None, min_length=1, max_length=255)
    concluida: bool | None = Field(default=None)

class TarefaResponse(TarefaBase):
    id: int

    model_config = ConfigDict(from_attributes=True)

from sqlalchemy.orm import Session
from app.models.tarefa import Tarefa
from app.schemas.tarefa import TarefaCreate, TarefaUpdate

class TarefaRepository:
    def __init__(self, db: Session):
        self.db = db

    def find_all(self) -> list[Tarefa]:
        return self.db.query(Tarefa).order_by(Tarefa.id.asc()).all()

    def find_by_id(self, id: int) -> Tarefa | None:
        return self.db.query(Tarefa).filter(Tarefa.id == id).first()

    def create(self, tarefa_in: TarefaCreate) -> Tarefa:
        db_tarefa = Tarefa(
            descricao=tarefa_in.descricao,
            concluida=tarefa_in.concluida,
        )
        self.db.add(db_tarefa)
        self.db.commit()
        self.db.refresh(db_tarefa)
        return db_tarefa

    def update(self, db_tarefa: Tarefa, tarefa_in: TarefaUpdate) -> Tarefa:
        if tarefa_in.descricao is not None:
            db_tarefa.descricao = tarefa_in.descricao
        if tarefa_in.concluida is not None:
            db_tarefa.concluida = tarefa_in.concluida
        
        self.db.commit()
        self.db.refresh(db_tarefa)
        return db_tarefa

    def delete(self, db_tarefa: Tarefa) -> None:
        self.db.delete(db_tarefa)
        self.db.commit()

from fastapi import HTTPException, status
from sqlalchemy.orm import Session
from app.repositories.tarefa_repository import TarefaRepository
from app.schemas.tarefa import TarefaCreate, TarefaUpdate
from app.models.tarefa import Tarefa

class TarefaService:
    def __init__(self, db: Session):
        self.repository = TarefaRepository(db)

    def listar_todas(self) -> list[Tarefa]:
        return self.repository.find_all()

    def buscar_por_id(self, id: int) -> Tarefa:
        tarefa = self.repository.find_by_id(id)
        if not tarefa:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"Tarefa não encontrada com o id: {id}",
            )
        return tarefa

    def criar(self, tarefa_in: TarefaCreate) -> Tarefa:
        return self.repository.create(tarefa_in)

    def atualizar(self, id: int, tarefa_in: TarefaUpdate) -> Tarefa:
        db_tarefa = self.buscar_por_id(id)
        return self.repository.update(db_tarefa, tarefa_in)

    def deletar(self, id: int) -> None:
        db_tarefa = self.buscar_por_id(id)
        self.repository.delete(db_tarefa)

from fastapi import APIRouter, Depends, status
from sqlalchemy.orm import Session
from app.database import get_db
from app.schemas.tarefa import TarefaCreate, TarefaUpdate, TarefaResponse
from app.services.tarefa_service import TarefaService

router = APIRouter(prefix="/api/tarefas", tags=["Tarefas"])

def get_tarefa_service(db: Session = Depends(get_db)) -> TarefaService:
    return TarefaService(db)

@router.get("", response_model=list[TarefaResponse], summary="Listar todas as tarefas")
def listar_tarefas(service: TarefaService = Depends(get_tarefa_service)):
    return service.listar_todas()

@router.get("/{id}", response_model=TarefaResponse, summary="Buscar tarefa por ID")
def buscar_tarefa(id: int, service: TarefaService = Depends(get_tarefa_service)):
    return service.buscar_por_id(id)

@router.post("", response_model=TarefaResponse, status_code=status.HTTP_201_CREATED, summary="Criar nova tarefa")
def criar_tarefa(tarefa_in: TarefaCreate, service: TarefaService = Depends(get_tarefa_service)):
    return service.criar(tarefa_in)

@router.put("/{id}", response_model=TarefaResponse, summary="Atualizar tarefa existente")
def atualizar_tarefa(id: int, tarefa_in: TarefaUpdate, service: TarefaService = Depends(get_tarefa_service)):
    return service.atualizar(id, tarefa_in)

@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT, summary="Deletar tarefa por ID")
def deletar_tarefa(id: int, service: TarefaService = Depends(get_tarefa_service)):
    service.deletar(id)

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
from app.database import Base, engine
from app.routers import tarefas_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Inicializa as tabelas do banco
    Base.metadata.create_all(bind=engine)
    yield

app = FastAPI(
    title=settings.APP_NAME,
    description="API RESTful de Lista de Tarefas com FastAPI e SQLAlchemy",
    version="1.0.0",
    lifespan=lifespan,
)

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

app.include_router(tarefas_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) as test_client:
        yield test_client
    app.dependency_overrides.clear()

def test_listar_tarefas_vazio(client):
    response = client.get("/api/tarefas")
    assert response.status_code == 200
    assert response.json() == []

def test_criar_tarefa(client):
    payload = {"descricao": "Estudar FastAPI", "concluida": False}
    response = client.post("/api/tarefas", json=payload)
    assert response.status_code == 201
    data = response.json()
    assert data["id"] == 1
    assert data["descricao"] == "Estudar FastAPI"
    assert data["concluida"] is False

def test_buscar_tarefa_por_id(client):
    create_resp = client.post("/api/tarefas", json={"descricao": "Comprar pão", "concluida": False})
    tarefa_id = create_resp.json()["id"]

    response = client.get(f"/api/tarefas/{tarefa_id}")
    assert response.status_code == 200
    assert response.json()["descricao"] == "Comprar pão"

    response_404 = client.get("/api/tarefas/999")
    assert response_404.status_code == 404

def test_atualizar_tarefa(client):
    create_resp = client.post("/api/tarefas", json={"descricao": "Lavar o carro", "concluida": False})
    tarefa_id = create_resp.json()["id"]

    update_resp = client.put(f"/api/tarefas/{tarefa_id}", json={"descricao": "Lavar o carro e aspirar", "concluida": True})
    assert update_resp.status_code == 200
    assert update_resp.json()["descricao"] == "Lavar o carro e aspirar"
    assert update_resp.json()["concluida"] is True

def test_deletar_tarefa(client):
    create_resp = client.post("/api/tarefas", json={"descricao": "Tarefa temporária", "concluida": False})
    tarefa_id = create_resp.json()["id"]

    delete_resp = client.delete(f"/api/tarefas/{tarefa_id}")
    assert delete_resp.status_code == 204

    get_resp = client.get(f"/api/tarefas/{tarefa_id}")
    assert get_resp.status_code == 404