Blog API: usuários e autenticação

from datetime import datetime

from sqlalchemy import DateTime, ForeignKey, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.database import Base


class Usuario(Base):
    __tablename__ = "usuarios"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    nome: Mapped[str] = mapped_column(String(100))
    senha_hash: Mapped[str] = mapped_column(String(255))
    criado_em: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

    posts: Mapped[list["Post"]] = relationship(back_populates="autor", cascade="all, delete-orphan")


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    titulo: Mapped[str] = mapped_column(String(200))
    conteudo: Mapped[str] = mapped_column(Text)
    autor_id: Mapped[int] = mapped_column(ForeignKey("usuarios.id", ondelete="CASCADE"), index=True)
    criado_em: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    atualizado_em: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), onupdate=func.now(), default=None
    )

    autor: Mapped[Usuario] = relationship(back_populates="posts")

from datetime import datetime
from typing import Generic, TypeVar

from pydantic import BaseModel, ConfigDict, EmailStr, Field

T = TypeVar("T")


class UsuarioCriar(BaseModel):
    email: EmailStr
    nome: str = Field(min_length=1, max_length=100)
    senha: str = Field(min_length=8, max_length=128)


class UsuarioSaida(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: str
    nome: str


class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"


class PostCriar(BaseModel):
    titulo: str = Field(min_length=1, max_length=200)
    conteudo: str = Field(min_length=1)


class PostSaida(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    titulo: str
    conteudo: str
    autor_id: int
    criado_em: datetime
    atualizado_em: datetime | None


class Pagina(BaseModel, Generic[T]):
    pagina: int
    limite: int
    total: int
    paginas: int
    dados: list[T]

from datetime import datetime, timedelta, timezone

import jwt
from pwdlib import PasswordHash

from app.config import Configuracoes

hasher = PasswordHash.recommended()
# Hash de uma senha que ninguém usa: serve para gastar o mesmo tempo quando o e-mail não existe.
SENHA_FALSA = hasher.hash("senha-que-ninguem-usa")


def hash_senha(senha: str) -> str:
    return hasher.hash(senha)


def verificar_senha(senha: str, senha_hash: str) -> bool:
    return hasher.verify(senha, senha_hash)


def criar_token(config: Configuracoes, usuario_id: int, minutos: int | None = None) -> str:
    duracao = config.expira_em_minutos if minutos is None else minutos
    expira = datetime.now(timezone.utc) + timedelta(minutes=duracao)
    carga = {"sub": str(usuario_id), "exp": expira}
    return jwt.encode(carga, config.chave_secreta.get_secret_value(), algorithm=config.algoritmo)


def ler_token(config: Configuracoes, token: str) -> int | None:
    """Devolve o id do usuário, ou None se o token for inválido, adulterado ou vencido."""
    try:
        carga = jwt.decode(
            token, config.chave_secreta.get_secret_value(), algorithms=[config.algoritmo]
        )
        return int(carga["sub"])
    except (jwt.InvalidTokenError, KeyError, ValueError):
        return None

from collections.abc import Iterator
from typing import Annotated

from fastapi import Depends, HTTPException, Request, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

from app.config import Configuracoes
from app.models import Usuario
from app.security import ler_token

oauth2 = OAuth2PasswordBearer(tokenUrl="auth/token")


def obter_config(request: Request) -> Configuracoes:
    config: Configuracoes = request.app.state.config
    return config


def obter_sessao(request: Request) -> Iterator[Session]:
    with request.app.state.fabrica_sessao() as sessao:
        yield sessao


Config = Annotated[Configuracoes, Depends(obter_config)]
Sessao = Annotated[Session, Depends(obter_sessao)]


def usuario_atual(
    token: Annotated[str, Depends(oauth2)], sessao: Sessao, config: Config
) -> Usuario:
    erro = HTTPException(
        status.HTTP_401_UNAUTHORIZED,
        "Token inválido ou expirado",
        headers={"WWW-Authenticate": "Bearer"},
    )
    usuario_id = ler_token(config, token)
    if usuario_id is None:
        raise erro
    usuario = sessao.get(Usuario, usuario_id)
    if usuario is None:
        raise erro
    return usuario


UsuarioAtual = Annotated[Usuario, Depends(usuario_atual)]

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from app.deps import Config, Sessao, UsuarioAtual
from app.models import Usuario
from app.schemas import Token, UsuarioCriar, UsuarioSaida
from app.security import SENHA_FALSA, criar_token, hash_senha, verificar_senha

router = APIRouter(tags=["Autenticação"])


@router.post("/usuarios", response_model=UsuarioSaida, status_code=status.HTTP_201_CREATED)
def registrar(dados: UsuarioCriar, sessao: Sessao) -> Usuario:
    usuario = Usuario(
        email=dados.email.lower(), nome=dados.nome, senha_hash=hash_senha(dados.senha)
    )
    sessao.add(usuario)
    try:
        sessao.commit()
    except IntegrityError:
        sessao.rollback()
        raise HTTPException(status.HTTP_409_CONFLICT, "E-mail já cadastrado") from None
    return usuario


@router.post("/auth/token", response_model=Token)
def entrar(
    formulario: Annotated[OAuth2PasswordRequestForm, Depends()], sessao: Sessao, config: Config
) -> Token:
    usuario = sessao.scalar(select(Usuario).where(Usuario.email == formulario.username.lower()))
    senha_hash = usuario.senha_hash if usuario else SENHA_FALSA
    senha_confere = verificar_senha(formulario.password, senha_hash)
    if usuario is None or not senha_confere:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            "E-mail ou senha inválidos",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return Token(access_token=criar_token(config, usuario.id))


@router.get("/usuarios/eu", response_model=UsuarioSaida)
def meus_dados(usuario: UsuarioAtual) -> Usuario:
    return usuario