Chapter 32, Projects
Blog API: users and authentication
First, who is who. The models of the two tables, the entry and exit contracts, and the security: password hash, token, and the dependency that delivers the logged-in user to any route.
The tables
Two models, linked by a foreign key: one Usuario has many Post. The email is unique and indexed (the search in the login is by it). The ondelete="CASCADE" deletes the posts of a removed user, and the server_default=func.now() makes the database fill in the creation date:
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")
The contracts
The database model and the API model are different things (chapter 17). Here is the separation in practice: UsuarioCriar receives the password, and UsuarioSaida does not have password or hash fields, so there is no way to leak them by accident. The EmailStr validates the email format, and the password has a minimum and maximum (the maximum prevents someone from sending a huge text just to overload the hash). The Pagina[T] is the generic model from chapter 27:
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]
Hash and token
The security.py combines what we saw in chapters 19 and 20. A decision to observe: the sub of the token stores the id of the user (an identifier that does not change), and not the email (which the user can change). The function ler_token returns the id or None, hiding from the route the difference between "expired", "tampered" and "malformed":
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
The dependencies
The deps.py gathers the three pieces that the routes require. The session comes from the factory stored in app.state (so each application, including the test one, has its own database). The configuration comes from the same place. And the current user reads the token, fetches the user in the database and returns the object. Fetching from the database, instead of relying only on the token, ensures that a deleted user loses access even with a still valid token:
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)]
The authentication routes
Three routes: registration, login, and "who am I". The registration normalizes the email to lowercase, and treats the IntegrityError of the database (the uniqueness constraint) as 409, which also resolves the run of two simultaneous registrations with the same email (a "exists?" check before inserting wouldn't solve it). The login uses both defenses from chapter 20: the same message for non-existent email and wrong password, and hash calculated even when the user does not exist:
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
`raise ... from None`
Inside a
except, raising another exception withoutfrommakes Python chain the two in the error message ("during the handling of the above exception, another occurred"). Thefrom Nonediscards the original, which here is an internal detail of the database. Theruffpoints to the oversight (rule B904), and it was this that made me correct it.