Curso de FastAPI – Rest API com Python (Backend Completo)

Bem-vindo ao Curso de FastAPI – Rest API com Python (Backend Completo), uma série de aulas que vão te guiar no desenvolvimento de um backend completo de uma aplicação de delivery de pizzaria usando Python.

Ao longo deste curso, exploraremos desde os fundamentos até a implementação prática do sistema, cobrindo tópicos como integração com banco de dados, autenticação, rotas assíncronas e documentação automática.

É um curso ideal para iniciantes e desenvolvedores intermediários, oferecendo exemplos reais e materiais de suporte para que você construa uma API profissional e otimizada até o final!

Vamos começar essa jornada juntos? 🚀

Aula 1: Apresentação do projeto

Esta primeira parte do curso traz a introdução do curso, os fundamentos do FastAPI, o planejamento das estruturas do projeto e os primeiros passos para colocar a API no ar!

Para receber por e-mail o(s) arquivo(s) utilizados na aula, preencha:

Contexto do Projeto: Backend de Pizzaria Delivery

Neste curso, vamos desenvolver um backend para gerenciar um sistema de delivery de pizzaria. O foco será criar rotas para gerenciar pedidos, como criar um pedido, adicionar ou remover produtos, escolher opções, finalizar e entregar pedidos, além de autenticar usuários.

Essas funcionalidades serão definidas em arquivos de rotas, que dependem de estruturas como banco de dados, modelos, dependências e esquemas para funcionar eficientemente.

Estruturas Essenciais do Projeto

Para que o backend funcione, precisamos de algumas estruturas principais:

• Integração com Banco de Dados no FastAPI: Um banco de dados será criado para armazenar pedidos, usuários e itens de pedidos, garantindo que todas as informações sejam persistentes e acessíveis.
• Modelos de Dados (Models) no FastAPI: Arquivos de modelos definirão as tabelas do banco de dados, como tabelas para usuários, pedidos e itens de pedidos, estruturando os dados de forma organizada.
• Arquivo de Dependências: Regras de Acesso e Autenticação: Este arquivo permitirá restringir o acesso (ex.: apenas usuários autenticados podem ver ordens) e gerenciar requisições ao banco de dados de forma eficiente, especialmente em cenários de múltiplas requisições simultâneas.
• Schemas no FastAPI: Estruturando Dados de Entrada e Resposta: Os esquemas estruturam os dados de entrada e saída, garantindo consistência e eficiência no processamento.
• Tipagem de Dados no FastAPI: Segurança e Performance: A tipagem estática do FastAPI melhora a performance e a segurança, evitando erros de tipo durante a execução.

Benefícios do FastAPI com Funções Assíncronas

Uma das grandes vantagens do FastAPI é o uso de funções assíncronas, que permitem processar múltiplas requisições simultaneamente. Imagine que uma requisição de um usuário leve 15 segundos para consultar o banco de dados. Em vez de travar o sistema esperando essa resposta, o FastAPI inicia a próxima requisição e retorna à anterior quando possível. Isso evita gargalos e melhora a escalabilidade, sendo ideal para sistemas com alto tráfego, como uma pizzaria delivery.

Iniciando o Projeto FastAPI do Zero

Vamos começar do básico. Abra o VS Code com um ambiente virtual configurado. Crie um arquivo chamado main.py, que será o ponto de entrada do projeto. A primeira etapa é instalar as bibliotecas necessárias:

• Instalação das Bibliotecas Necessárias: Instale o FastAPI, Uvicorn (servidor ASGI), SQLAlchemy (para integração com banco de dados), psycopg2-binary com o parâmetro bcrypt (para criptografia de senhas), python-jose[cryptography] (para tokens JWT), python-dotenv (para variáveis de ambiente) e python-multipart (dependência adicional). Um arquivo requirements.txt com as versões usadas estará disponível para download.

from fastapi import FastAPI

app = FastAPI()

Esse código cria uma instância básica do FastAPI. Para rodar o servidor, use o comando no terminal:

uvicorn main:app --reload

O parâmetro --reload reinicia automaticamente o servidor a cada alteração no código, facilitando o desenvolvimento.

Executando o Servidor FastAPI Localmente

Ao executar o comando uvicorn main:app --reload, o servidor será iniciado localmente (geralmente em http://127.0.0.1:8000). Acesse esse link no navegador para verificar que o servidor está ativo, mesmo sem rotas definidas (retornará "Not Found"). Acesse /docs para ver a documentação automática gerada pelo FastAPI, que será preenchida à medida que adicionarmos rotas.

Documentação Automática da API

O FastAPI gera automaticamente uma documentação interativa em /docs, baseada no padrão OpenAPI. Isso é útil para testar endpoints e compartilhar a API com outros desenvolvedores, sem necessidade de documentação manual.

Encerramento da Aula 1

Neste capítulo, configuramos o ambiente e iniciamos a API FastAPI. Criamos a base do projeto com uma instância do FastAPI e aprendemos a rodá-la localmente com Uvicorn.

Caso prefira esse conteúdo no formato de vídeo-aula, assista ao vídeo abaixo ou acesse o nosso canal do YouTube!

Aula 2: Criando e Testando Rotas para Pedidos e Autenticação

Agora que já sabemos configurar o ambiente e iniciar o servidor FastAPI, vamos focar na criação de rotas para autenticação e pedidos, aprender sobre requisições HTTP, organizar a estrutura de rotas e testar isso tudo usando a documentação automática do FastAPI.

Para receber por e-mail o(s) arquivo(s) utilizados na aula, preencha:

O que são Rotas em APIs REST e como Criá-las com FastAPI

Rotas, ou endpoints, definem o que acontece quando um usuário acessa um link específico da API, como dominio.com/pedidos.

No FastAPI, rotas são os caminhos que vêm após o domínio do site, e cada rota está associada a um tipo de requisição HTTP.

Por exemplo, ao acessar localhost:8000/pedidos, a API deve responder com uma ação predefinida, como listar pedidos.

Como estamos construindo uma API REST, essas rotas lidam com requisições e respostas no formato JSON, sem a camada de frontend.

Entendendo Requisições HTTP: GET, POST, PUT e DELETE

APIs REST geralmente seguem o padrão CRUD (Create, Read, Update, Delete), que se traduz em quatro tipos principais de requisições HTTP:

• GET: Lê informações (ex.: listar pedidos).
• POST: Cria ou envia dados (ex.: criar um pedido).
• PUT/PATCH: Edita dados existentes (ex.: atualizar ou modificar um pedido).
• DELETE: Remove dados (ex.: excluir um pedido).

Neste curso, vamos focar em requisições GET e POST, as mais comuns, mas você também aprenderá como implementar PUT e DELETE se necessário.

Por exemplo, uma rota GET em /pedidos pode listar todos os pedidos, enquanto uma POST em /pedidos/criar pode adicionar um novo pedido.

Criando Arquivos de Rotas para Autenticação e Pedidos

Para organizar o projeto, vamos imaginar arquivos de rotas separados: auth_routes.py (para autenticação) e order_routes.py (para pedidos). Esses arquivos serão explorado na Aula 3.

Essa separação evita conflitos e melhora a organização, especialmente em projetos com diferentes seções, como autenticação, pedidos e, futuramente, uma área administrativa.

Cada arquivo terá um prefixo único para suas rotas (ex.: /auth e /pedidos), garantindo que não haja sobreposição de caminhos.

Como Importar Rotas para o Arquivo Principal (main.py)

No arquivo main.py, importamos os roteadores de autenticação e pedidos após criar a instância do FastAPI (app = FastAPI()).

from auth_routes import auth_router
from order_routes import order_router

Criando o Roteador de Autenticação no FastAPI

No arquivo auth_routes.py, criamos um roteador com o prefixo /auth e a tag "auth" para organizar a documentação:

auth_router = APIRouter(prefix="/auth", tags=["auth"])

Criando o Roteador de Pedidos no FastAPI

No arquivo order_routes.py, criamos um roteador com o prefixo /pedidos e a tag "pedidos":

order_router = APIRouter(prefix="/pedidos", tags=["pedidos"])

Testando o Servidor Local com Uvicorn

Execute o servidor com

uvicorn main:app --reload.

Acesse http://localhost:8000/docs para ver a documentação automática do FastAPI.

Inicialmente, ela mostrará que não há rotas, mas à medida que as adicionarmos, elas aparecerão organizadas por tags.

Criando as Primeiras Rotas da Aplicação

Vamos criar rotas básicas para testar a estrutura.

No FastAPI, uma rota é definida usando um decorador (como @router.get) e uma função assíncrona que retorna a resposta.

Como Criar uma Rota Padrão de Pedidos na API

No arquivo order_routes.py, iremos adicionar uma rota GET padrão em /pedidos:

@order_router.get("/")
async def pedidos():
    """
    Essa é a rota padrão de pedidos do nosso sistema. Todas as rotas dos pedidos precisam de autenticação
    """
    return {"mensagem": "Você acessou a rota de pedidos"}

Criando uma Rota de Autenticação no FastAPI

No arquivo auth_routes.py, será adicionada uma rota GET padrão em /auth:

@auth_router.get("/")
async def home():
    """
    Essa é a rota padrão de autenticação do nosso sistema
    """
    return {"mensagem": "Você acessou a rota padrão de autenticação", "autenticado": False}

Como Definir Funções Assíncronas (async def) em Rotas FastAPI

As funções das rotas são assíncronas (async def) para aproveitar a capacidade do FastAPI de lidar com múltiplas requisições simultaneamente, como vimos na aula passada. Elas retornam dicionários que o FastAPI converte automaticamente para JSON.

O que é JSON e como o FastAPI Usa Esse Formato?

JSON é um formato leve de troca de dados, semelhante a um dicionário Python, amplamente usado em APIs REST.

No FastAPI, todas as respostas são enviadas como JSON, e o framework converte automaticamente os dicionários retornados pelas rotas nesse formato.

Visualizando e Testando Rotas na Documentação Automática do FastAPI

Acesse http://localhost:8000/docs novamente.

Agora, você verá duas seções: "pedidos" e "auth".

A rota /pedidos mostra a descrição da docstring e retorna {"mensagem": "Você acessou a rota de pedidos"} ao ser testada.

A rota /auth retorna {"mensagem": "Você acessou a rota padrão de autenticação", "autenticado": false}.

Use o botão "Try it out" para executar requisições diretamente na documentação.

Como Documentar Rotas Usando Docstrings

As docstrings nas funções das rotas (texto entre """) aparecem na documentação automática, ajudando a explicar o propósito da rota, parâmetros e possíveis erros.

Isso é essencial para APIs públicas, facilitando o uso por outros desenvolvedores.

Encerramento da Aula 2

Nesta aula, entendemos como funcionam as primeiras rotas para autenticação e pedidos, organizamos a estrutura do projeto com roteadores e aprendemos a usar a documentação automática do FastAPI.

Caso prefira esse conteúdo no formato de vídeo-aula, assista ao vídeo abaixo ou acesse o nosso canal do YouTube!

Aula 3: Bancos de Dados, Modelos e Migrações

Bem-vindo à terceira aula do nosso curso de FastAPI!

Nesta seção vamos entender estruturação do banco de dados para o nosso sistema de delivery de pizzaria. Aprenderemos a criar modelos, configurar o banco de dados, realizar migrações com segurança e garantir a integridade dos dados.

Vamos usar o SQLAlchemy para gerenciar o banco e o Alembic para o processo de migrações.

Para receber por e-mail o(s) arquivo(s) utilizados na aula, preencha:

Visão Geral do Banco de Dados no FastAPI

Para avançar no nosso projeto de pizzaria delivery, precisamos armazenar informações, tais como usuários, pedidos e itens de pedidos.

Isso exige um banco de dados funcional. O FastAPI permite integrar diversas ferramentas de banco de dados em Python, e uma das mais populares é o SQLAlchemy, que instalamos na Aula 1.

Com ele vamos criar e gerenciar nosso banco de dados localmente, utilizando o SQLite por simplicidade, mas você poderá adaptar também para outros bancos (como PostgreSQL) no futuro, apenas ajustando a conexão.

O que é SQLAlchemy e como funciona o ORM?

O SQLAlchemy é uma biblioteca Python que simplifica o trabalho com bancos de dados usando o conceito de ORM (Object-Relational Mapping).

Com o ORM, você cria classes em Python que são automaticamente traduzidas em tabelas no banco de dados.

Em vez de escrever comandos SQL manualmente, você manipula essas classes, e o SQLAlchemy converte suas ações em comandos SQL. Isso torna o processo mais simples e intuitivo, especialmente para quem está começando.

O SQLAlchemy suporta diversos bancos de dados SQL, como SQLite, PostgreSQL, MySQL, entre outros.

Criando o Banco de Dados com SQLAlchemy no FastAPI

Vamos estruturar o banco de dados criando um arquivo models.py, onde definiremos as classes que representam as tabelas. Primeiro, configuramos a conexão com o banco de dados usando o SQLAlchemy.

Passo 1: Configurando a Conexão com o Banco de Dados

Crie o arquivo models.py e adicione o código inicial para conectar ao banco SQLite:

from sqlalchemy import create_engine

from sqlalchemy.orm import declarative_base

# Conexão com o banco de dados SQLite
DATABASE_URL = "sqlite:///banco.db"
engine = create_engine(DATABASE_URL)

# Base para criar os modelos
Base = declarative_base()

Explicação do código:

• create_engine: Cria a conexão com o banco de dados. Aqui, usamos SQLite (sqlite:///banco.db), que criará um arquivo banco.db na pasta do projeto. Para outros bancos (como PostgreSQL), você substituirá o DATABASE_URL pelo link do servidor ao fazer o deploy.
• declarative_base: Cria a base que usaremos para definir as tabelas.

Passo 2: Definindo as Tabelas Necessárias

Nosso sistema de pizzaria delivery precisa de três tabelas principais:

• Usuários: Para armazenar informações de quem faz os pedidos.
• Pedidos: Para gerenciar os pedidos em si.
• Itens de Pedidos: Para os itens (como pizzas) dentro de cada pedido.

Como Modelar Tabelas com Classes Python

Vamos criar as classes que representam essas tabelas. Cada classe será uma subclasse de Base, e o SQLAlchemy a transformará em uma tabela no banco de dados.

Tabela de usuários

A tabela usuários armazenará os dados dos usuários. Vamos defini-la com os campos: id, nome, email, senha, ativo e admin.

from sqlalchemy import Column, Integer, String, Boolean

class Usuario(Base):

    __tablename__ = "usuarios"

    id = Column(Integer, primary_key=True, autoincrement=True)

    nome = Column(String)

    email = Column(String, nullable=False)

    senha = Column(String)

    ativo = Column(Boolean, default=True)

    admin = Column(Boolean, default=False)

    def __init__(self, nome: str, email: str, senha: str, ativo: bool = True, admin: bool = False):

        self.nome = nome

        self.email = email

        self.senha = senha

        self.ativo = ativo

        self.admin = admin

Explicação do código:

• __tablename__: Define o nome da tabela no banco de dados como usuarios.
• Campos:
  • id: Chave primária, auto-incrementada (começa em 1 e aumenta automaticamente).
  • nome, email, senha: Strings para armazenar o nome, email e senha do usuário.
  • ativo, admin: Campos booleanos com valores padrão (True para ativo, False para admin).
• nullable=False: Garante que o email nunca seja nulo (obrigatório).
• __init__: Define como criar um novo usuário no código Python, exigindo nome, email e senha, com ativo e admin opcionais (usam valores padrão).

Tabela de Pedidos

A tabela pedidos armazenará os pedidos, com os campos: id, usuario, status e preco.

class Pedido(Base):
    __tablename__ = "pedidos"

    id = Column(Integer, primary_key=True, autoincrement=True)
    usuario = Column(Integer, ForeignKey("usuarios.id"), nullable=False)
    status = Column(String)  # "PENDENTE", "CANCELADO", "FINALIZADO"
    preco = Column(Float, default=0.0)

    def __init__(self, usuario: int, status: str = "PENDENTE", preco: float = 0.0):
        self.usuario = usuario
        self.status = status
        self.preco = preco

Explicação do código:

• ForeignKey("usuarios.id"): Conecta a coluna usuario ao id da tabela usuarios, indicando que cada pedido está associado a um usuário específico.
• status: Inicialmente definido como string para simplificar (valores: "PENDENTE", "CANCELADO", "FINALIZADO"). Poderia usar ChoiceType para restringir os valores, mas optamos por restringir via código/esquemas (explicado mais adiante).
• preco: Um float com valor padrão 0.0, já que o preço será calculado com base nos itens.

Tabela de Itens de Pedidos

A tabela itens_pedidos armazenará os itens de cada pedido, com os campos: id, quantidade, sabor, tamanho, preco_unitario e pedido.

class ItemPedido(Base):
    __tablename__ = "itens_pedidos"

    id = Column(Integer, primary_key=True, autoincrement=True)
    quantidade = Column(Integer, nullable=False)
    sabor = Column(String, nullable=False)
    tamanho = Column(String, nullable=False)
    preco_unitario = Column(Float, nullable=False)
    pedido = Column(Integer, ForeignKey("pedidos.id"), nullable=False)

    def __init__(self, quantidade: int, sabor: str, tamanho: str, preco_unitario: float, pedido: int):
        self.quantidade = quantidade
        self.sabor = sabor
        self.tamanho = tamanho
        self.preco_unitario = preco_unitario
        self.pedido = pedido

Explicação do código:

• ForeignKey("pedidos.id"): Associa cada item a um pedido específico.
• sabor e tamanho: Strings para flexibilidade (novos sabores ou tamanhos podem ser adicionados sem alterar o banco).
• preco_unitario: Preço de um item (ex.: uma pizza de um sabor e tamanho específico).
• Todos os campos são obrigatórios (nullable=False), exceto id, que é auto-incrementado.

O que é Migração de Banco de Dados no FastAPI?

Quando você cria um banco de dados, ele começa com uma estrutura inicial. Mas, ao longo do desenvolvimento, você pode precisar alterá-lo (ex.: adicionar um campo telefone à tabela usuarios).

Essas alterações afetam a estrutura do banco e os dados existentes. Para fazer isso de forma segura, usamos migrações, que atualizam o banco de uma versão para outra sem perder dados.

No SQLAlchemy, a ferramenta mais usada para migrações é o Alembic. Vamos configurá-lo para gerenciar as migrações do nosso projeto.

Como Configurar o Alembic no Projeto FastAPI

Passo 1: Instalar o Alembic

No terminal, instale o Alembic:

pip install alembic

Passo 2: Inicializar o Alembic

Execute o comando para criar a estrutura do Alembic:

alembic init alembic

Isso cria:

• Um arquivo alembic.ini (configurações do Alembic).
• Uma pasta alembic com os arquivos env.py e uma subpasta versions (onde as migrações serão armazenadas).

Passo 3: Configurar o alembic.ini

Edite o arquivo alembic.ini para apontar para o nosso banco de dados. Localize a linha:

sqlalchemy.url

E substitua pelo nosso DATABASE_URL:

sqlalchemy.url = sqlite:///banco.db

Passo 4: Configurar o env.py

O arquivo alembic/env.py precisa importar o Base do nosso models.py para reconhecer as tabelas. Como env.py está dentro da pasta alembic, precisamos ajustar o caminho de importação.

Adicione no início de env.py:

import sys

import os

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from models import Base

• sys.path.append: Adiciona a pasta principal do projeto ao caminho de busca do Python, permitindo importar models.py.
• from models import Base: Importa a variável Base que contém os metadados das tabelas.

Em seguida, localize a linha no env.py que define target_metadata e ajuste para:

target_metadata = Base.metadata

Isso informa ao Alembic quais tabelas ele deve gerenciar.

Gerando Banco de Dados e Migrações com Alembic

Passo 1: Criar a Migração Inicial

Com o Alembic configurado, gere a primeira migração:

alembic revision --autogenerate -m "initial_migration"

• --autogenerate: O Alembic detecta automaticamente as tabelas definidas em models.py e gera os comandos de migração.
• -m "initial_migration": Nomeia a migração para fácil identificação.

Isso cria um arquivo de migração na pasta alembic/versions, detalhando a criação das tabelas usuarios, pedidos e itens_pedidos.

Passo 2: Executar a Migração

Aplique a migração para criar o banco de dados:

alembic upgrade head

• upgrade head: Executa a migração mais recente, criando o arquivo banco.db com todas as tabelas.

Visualizando o Banco de Dados

No VS Code, instale a extensão SQLite Viewer para visualizar o banco.db. Você verá as tabelas usuarios, pedidos e itens_pedidos com suas colunas conforme definidas.

Criando Novas Migrações Após Alterações

Se precisar alterar o banco de dados (ex.: remover o campo admin da tabela usuarios), o processo é semelhante.

Exemplo: Removendo o Campo admin

1. Edite models.py e remova a linha admin = Column(Boolean, default=False) da classe Usuario.
2. Gere uma nova migração:

alembic revision --autogenerate -m "remover_admin_usuario"

3. Aplique a migração:

alembic upgrade head

O Alembic atualizará o banco de dados, removendo a coluna admin. Você pode verificar no banco.db que a coluna não existe mais.

Exemplo: Adicionando o Campo admin Novamente

1. Adicione a linha admin = Column(Boolean, default=False) de volta à classe Usuario.
2. Gere outra migração:

alembic revision --autogenerate -m "adicionar_admin_usuario"

3. Aplique a migração:

alembic upgrade head

O campo admin será adicionado novamente.

Encerramento da Aula 3

Nesta aula, configuramos o banco de dados do nosso sistema de pizzaria delivery usando SQLAlchemy e Alembic.

Criamos as tabelas usuarios, pedidos e itens_pedidos, definimos relacionamentos com chaves estrangeiras e aprendemos a gerenciar migrações para atualizar o banco de dados com segurança.e

Caso prefira esse conteúdo no formato de vídeo-aula, assista ao vídeo abaixo ou acesse o nosso canal do YouTube!

Aula 4: Criar conta de usuário, Schemas e Criptografia

Bem-vindo à quarta aula do nosso curso. Nela, avançaremos no desenvolvimento do backend da nossa pizzaria delivery, implementando a funcionalidade de criação de usuários com autenticação segura.

Vamos criar uma rota para cadastrar usuários, gerenciar sessões de banco de dados de forma robusta, criptografar senhas com segurança e padronizar a entrada de dados usando esquemas Pydantic.

Para receber por e-mail o(s) arquivo(s) utilizados na aula, preencha:

O que vamos aprender nesta aula?

• Criar a rota /auth/criar_conta para cadastrar usuários.
• Implementar uma dependência para gerenciar sessões do banco de dados com segurança.
• Criptografar senhas usando bcrypt e armazenar a chave secreta em variáveis de ambiente.
• Usar esquemas Pydantic para padronizar e validar dados de entrada.
• Personalizar respostas HTTP com códigos de erro (ex.: erro 400 para e-mails duplicados).
• Testar a rota na documentação automática do FastAPI.

Contexto do Projeto

Na Aula 3, configuramos o banco de dados com SQLAlchemy e Alembic, criando as tabelas usuarios, pedidos, e itens_pedidos. Agora, vamos usar essas tabelas para cadastrar usuários, garantindo que:

• O e-mail seja único (não permita cadastros duplicados).
• A senha seja criptografada antes de ser armazenada.
• As sessões do banco sejam gerenciadas de forma segura e eficiente.
• Os dados de entrada sejam validados e padronizados.

Configurando o Ambiente para Autenticação Segura

Antes de criar a rota, precisamos configurar algumas ferramentas para garantir a segurança e a robustez do sistema:

1. Variáveis de Ambiente com python-dotenv
   • Para armazenar chaves secretas (como a usada na criptografia de senhas), usamos um arquivo .env. Este arquivo não deve ser compartilhado publicamente (ex.: no GitHub).
   • Crie um arquivo chamado .env na raiz do projeto com o seguinte conteúdo: SECRET_KEY=uma_chave_secreta_muito_forte_gerada_por_voce
   • Dica: Gere uma chave forte usando um gerador online ou o módulo secrets do Python. Exemplo: import secrets print(secrets.token_hex(32))
2. Criptografia de Senhas com bcrypt
   • Usaremos a biblioteca passlib[bcrypt] para criptografar senhas. No arquivo main.py, já configuramos o bcrypt_context para isso.
3. Dependências para Sessões do Banco
   • Criaremos uma função de dependência para gerenciar sessões do SQLAlchemy, garantindo que elas sejam abertas e fechadas corretamente, mesmo em caso de erros.

Criando a Dependência para Gerenciar Sessões

Para evitar repetir o código de conexão com o banco em todas as rotas e garantir que as sessões sejam fechadas corretamente, criaremos uma dependência no arquivo dependencies.py. Esta dependência usa yield e try/finally para gerenciar sessões de forma robusta.

Arquivo dependencies.py:

from models import db
from sqlalchemy.orm import sessionmaker

def pegar_sessao():
    try:
        Session = sessionmaker(bind=db)
        session = Session()
        yield session
    finally:
        session.close()

Explicação do código:

• sessionmaker: Cria uma fábrica de sessões para o banco SQLite (banco.db).
• pegar_sessao: Função geradora que:
  • Cria uma sessão com Session().
  • Usa yield para fornecer a sessão à rota.
  • Usa finally para garantir que a sessão seja fechada, mesmo se ocorrer um erro.
• Vantagem: Centraliza o gerenciamento de sessões, evita repetição de código e previne vazamentos de conexão.

Criando o Esquema Pydantic para Usuários

Para padronizar e validar os dados de entrada, usaremos o Pydantic para criar um esquema (UsuarioSchema) que define como um usuário deve ser enviado na requisição.

Arquivo schemas.py:

from pydantic import BaseModel
from typing import Optional

class UsuarioSchema(BaseModel):
    nome: str
    email: str
    senha: str
    ativo: Optional[bool] = True
    admin: Optional[bool] = False

    class Config:
        from_attributes = True

Explicação do código:

• BaseModel: Classe base do Pydantic para criar esquemas.
• Campos:
  • nome, email, senha: Obrigatórios, do tipo str.
  • ativo, admin: Opcionais, do tipo bool, com valores padrão True e False, respectivamente.
• from_attributes = True: Permite converter objetos SQLAlchemy (como Usuario) em objetos Pydantic para respostas da API.
• Vantagens:
  • Valida automaticamente os tipos de dados.
  • Gera documentação automática em /docs.
  • Padroniza a entrada, evitando que o usuário envie dados incorretos.

Configurando o Arquivo Principal (main.py)

O arquivo main.py já está configurado para carregar variáveis de ambiente, inicializar o FastAPI, configurar o bcrypt e importar os roteadores.

Arquivo main.py:

from fastapi import FastAPI
from passlib.context import CryptContext
from dotenv import load_dotenv
import os

load_dotenv()

SECRET_KEY = os.getenv("SECRET_KEY")

app = FastAPI()

bcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

from auth_routes import auth_router
from order_routes import order_router

app.include_router(auth_router)
app.include_router(order_router)

Explicação do código:

• load_dotenv: Carrega variáveis do arquivo .env.
• SECRET_KEY: Armazena a chave secreta para criptografia (usada pelo bcrypt).
• bcrypt_context: Configura o bcrypt para criptografar senhas, com deprecated="auto" para usar apenas esquemas seguros.
• app.include_router: Inclui os roteadores de autenticação e pedidos.

Implementando a Rota de Criação de Usuários

Agora, vamos implementar a rota /auth/criar_conta no arquivo auth_routes.py. Esta rota:

• Valida se o e-mail já existe no banco.
• Criptografa a senha antes de salvar.
• Usa a dependência pegar_sessao para gerenciar a sessão do banco.
• Retorna respostas HTTP personalizadas (200 para sucesso, 400 para erro).

Arquivo auth_routes.py:

from fastapi import APIRouter, Depends, HTTPException
from models import Usuario
from dependencies import pegar_sessao
from main import bcrypt_context
from schemas import UsuarioSchema
from sqlalchemy.orm import Session

auth_router = APIRouter(prefix="/auth", tags=["auth"])

@auth_router.get("/")
async def home():
    """
    Essa é a rota padrão de autenticação do nosso sistema
    """
    return {"mensagem": "Você acessou a rota padrão de autenticação", "autenticado": False}

@auth_router.post("/criar_conta")
async def criar_conta(usuario_schema: UsuarioSchema, session: Session = Depends(pegar_sessao)):
    """
    Cria um novo usuário no sistema, com senha criptografada.
    """
    usuario = session.query(Usuario).filter(Usuario.email == usuario_schema.email).first()
    if usuario:
        raise HTTPException(status_code=400, detail="E-mail do usuário já cadastrado")
    else:
        senha_criptografada = bcrypt_context.hash(usuario_schema.senha)
        novo_usuario = Usuario(
            nome=usuario_schema.nome,
            email=usuario_schema.email,
            senha=senha_criptografada,
            ativo=usuario_schema.ativo,
            admin=usuario_schema.admin
        )
        session.add(novo_usuario)
        session.commit()
        return {"mensagem": f"usuário cadastrado com sucesso {usuario_schema.email}"}

Explicação do código:

• UsuarioSchema: Valida os dados de entrada (nome, email, senha, ativo, admin).
• Depends(pegar_sessao): Injeta a sessão do banco, garantindo que ela seja fechada após o uso.
• session.query(Usuario).filter(...).first(): Verifica se já existe um usuário com o e-mail fornecido.
• bcrypt_context.hash: Criptografa a senha antes de salvar.
• session.add e session.commit: Adiciona o novo usuário ao banco e salva as alterações.
• HTTPException: Retorna erro 400 se o e-mail já estiver cadastrado.
• Resposta de sucesso: Retorna uma mensagem com o e-mail do usuário cadastrado.

Testando a Rota na Documentação Automática

1. Execute o servidor:uvicorn main:app --reload
2. Acesse http://localhost:8000/docs.
3. Na seção auth, localize a rota POST /auth/criar_conta.
4. Clique em "Try it out" e envie um JSON como:{ "nome": "João Silva", "email": "joao@example.com", "senha": "123456", "ativo": true, "admin": false }
5. Verifique a resposta:
   • Sucesso: {"mensagem": "usuário cadastrado com sucesso joao@example.com"} (código 200).
   • Erro (e-mail duplicado): {"detail": "E-mail do usuário já cadastrado"} (código 400).
6. Use a extensão SQLite Viewer no VS Code para verificar o banco banco.db. A senha do usuário estará criptografada.

Por que Usar Esquemas e Dependências?

• Esquemas Pydantic:
  • Garantem que os dados enviados sejam válidos (ex.: email é uma string válida, ativo é booleano).
  • Padronizam a entrada, facilitando a manutenção e a documentação.
  • Reduzem erros, pois o FastAPI valida automaticamente os dados antes de processá-los.
• Dependências:
  • Evitam repetição de código (ex.: criar/abrir/fechar sessões em cada rota).
  • Garantem que a sessão do banco seja fechada, mesmo em caso de erros, usando try/finally.
  • Tornam o código mais modular e reutilizável.

Cuidados Importantes

• Senhas Criptografadas: Nunca armazene senhas em texto puro. O bcrypt garante que as senhas sejam salvas de forma segura.
• Variáveis de Ambiente: A SECRET_KEY no arquivo .env não deve ser exposta. Adicione .env ao .gitignore para evitar commits acidentais.
• Códigos HTTP: Use códigos apropriados (200 para sucesso, 400 para erros do cliente) para melhorar a comunicação com os usuários da API.
• Gerenciamento de Sessões: A dependência pegar_sessao garante que o banco não fique com conexões abertas, evitando problemas de desempenho.

Encerramento da Aula 4

Nesta aula, implementamos a rota de criação de usuários com autenticação segura, usando:

• Esquemas Pydantic para validar e padronizar dados.
• Dependências para gerenciar sessões do banco de forma robusta.
• Criptografia de senhas com bcrypt e variáveis de ambiente.
• Respostas HTTP personalizadas com HTTPException.

Nosso backend agora pode cadastrar usuários de forma segura, com senhas criptografadas e validação de e-mails únicos.

Caso prefira esse conteúdo no formato de vídeo-aula, assista ao vídeo abaixo ou acesse o nosso canal do YouTube!

Python Impressionador

Você vai aprender a linguagem de Programação que mais cresce no Mundo para fazer automações incríveis, desenvolver sites, fazer análises de dados, trabalhar com Ciência de Dados, Inteligência Artificial, mesmo que você nunca tenha tido nenhum contato com Programação na vida.

Black Friday · Lote Especial · Lote 1 · Último Lote

De R$ 1.497 56% OFF54% OFF53% OFF

por 12x R$ 6467,5069

Preço do Lote Especial até 13/11, às 23h59Último dia do Lote Especial: o preço vale até hoje, às 23h59Preço do Lote 1 até 20/11, às 23h59Último dia do Lote 1: o preço vale até hoje, às 23h59Preço do Último Lote até 27/11, às 23h59Último dia do Último Lote: o preço vale até hoje, às 23h59

Começar agoraQuero a oferta de Black Friday

Aula 5: Implementando Login com Tokens JWT

Nesta quinta aula, avançaremos no desenvolvimento do backend da nossa pizzaria delivery, implementando a funcionalidade de login com a rota /auth/login.

Vamos autenticar usuários com base no e-mail, criar um esquema Pydantic para padronizar os dados de entrada e gerar um token de acesso simplificado, introduzindo o conceito de JSON Web Tokens (JWT).

Este é o primeiro passo para proteger rotas, como a de criação de pedidos, que atualmente aceita requisições sem autenticação.

Para receber por e-mail o(s) arquivo(s) utilizados na aula, preencha:

O que vamos aprender nesta aula?

• Criar a rota /auth/login para autenticar usuários com base no e-mail.
• Definir o esquema LoginSchema para padronizar os dados de entrada (e-mail e senha).
• Gerar um token de acesso simplificado com base no ID do usuário.
• Introduzir o conceito de JWT e sua importância na autenticação.
• Testar a rota de login na documentação automática do FastAPI.
• Preparar o terreno para validação de senha e geração de JWTs padrão nas próximas aulas.

Contexto do Projeto

Na Aula 4, implementamos a rota /auth/criar_conta para cadastrar usuários com senhas criptografadas e a rota /pedidos/pedido para criar pedidos associados a um usuário. No entanto, qualquer um pode criar pedidos sem autenticação, o que não é seguro.

Nesta aula, começaremos a construir o sistema de autenticação, criando a rota de login que retorna um token. Nas próximas aulas, usaremos esse token para proteger rotas e validaremos a senha do usuário.

O que é um JSON Web Token (JWT)?

Um JSON Web Token (JWT) é uma string codificada que armazena informações (como o ID ou e-mail do usuário) de forma segura. Ele é composto por três partes separadas por pontos (.):

• Header: Contém metadados, como o algoritmo de assinatura (ex.: HS256).
• Payload: Contém os dados do usuário (ex.: user_id, data de expiração).
• Signature: Assinatura que valida a integridade do token.

Exemplo de JWT: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjo2fQ.signature

O JWT é enviado no cabeçalho das requisições como Bearer <token>, permitindo que o servidor verifique a identidade do usuário sem armazenar estado.

Nesta aula, criaremos um token simplificado (não um JWT padrão) para entender o processo.

Atualizando o Esquema Pydantic

Para padronizar os dados de entrada da rota de login (e-mail e senha), criaremos o esquema LoginSchema no arquivo schemas.py.

Atualize schemas.py:

from pydantic import BaseModel
from typing import Optional

class UsuarioSchema(BaseModel):
    nome: str
    email: str
    senha: str
    ativo: Optional[bool]
    admin: Optional[bool]

    class Config:
        from_attributes = True

class PedidoSchema(BaseModel):
    id_usuario: int

    class Config:
        from_attributes = True

class LoginSchema(BaseModel):
    email: str
    senha: str

    class Config:
        from_attributes = True

Explicação:

• LoginSchema: Define os campos obrigatórios email e senha como strings.
• from_attributes = True: Permite compatibilidade com objetos SQLAlchemy, embora aqui seja usado apenas para entrada.

Nota: Esquemas Pydantic não correspondem diretamente a tabelas do banco, mas a padrões de entrada/saída de rotas. O LoginSchema padroniza os dados enviados para a rota de login, enquanto UsuarioSchema é usado para criação de usuários.

Implementando a Rota de Login

Vamos adicionar a rota /auth/login ao arquivo auth_routes.py. Esta rota verificará se o e-mail existe no banco e retornará um token simplificado baseado no ID do usuário. A validação da senha será abordada na próxima aula, conforme indicado na transcrição.

Atualize auth_routes.py:

from fastapi import APIRouter, Depends, HTTPException
from models import Usuario
from dependencies import pegar_sessao
from main import bcrypt_context
from schemas import UsuarioSchema, LoginSchema
from sqlalchemy.orm import Session

auth_router = APIRouter(prefix="/auth", tags=["auth"])

@auth_router.get("/")
async def home():
    """
    Essa é a rota padrão de autenticação do nosso sistema
    """
    return {"mensagem": "Você acessou a rota padrão de autenticação", "autenticado": False}

@auth_router.post("/criar_conta")
async def criar_conta(usuario_schema: UsuarioSchema, session: Session = Depends(pegar_sessao)):
    usuario = session.query(Usuario).filter(Usuario.email==usuario_schema.email).first()
    if usuario:
        # ja existe um usuario com esse email
        raise HTTPException(status_code=400, detail="E-mail do usuário já cadastrado")
    else:
        senha_criptografada = bcrypt_context.hash(usuario_schema.senha)
        novo_usuario = Usuario(usuario_schema.nome, usuario_schema.email, senha_criptografada, usuario_schema.ativo, usuario_schema.admin)
        session.add(novo_usuario)
        session.commit()
        return {"mensagem": f"usuário cadastrado com sucesso {usuario_schema.email}"}

def criar_token(user_id: int) -> str:
    """
    Gera um token simplificado com base no ID do usuário (para fins didáticos).
    """
    return f"token_{user_id}_random"

@auth_router.post("/login")
async def login(login_schema: LoginSchema, session: Session = Depends(pegar_sessao)):
    """
    Autentica um usuário com base no e-mail e retorna um token de acesso.
    """
    usuario = session.query(Usuario).filter(Usuario.email == login_schema.email).first()
    if not usuario:
        raise HTTPException(status_code=400, detail="Usuário não encontrado")
    
    access_token = criar_token(usuario.id)
    return {
        "access_token": access_token,
        "token_type": "bearer"
    }

Explicação:

• Importações: Adicionamos LoginSchema às importações de schemas.
• LoginSchema: Valida os dados de entrada (e-mail e senha).
• session.query: Busca o usuário pelo e-mail no banco.
• criar_token: Função auxiliar que gera um token simplificado no formato token_<user_id>_random (ex.: token_6_random para o usuário com ID 6). Esta é uma abordagem didática, não segura, para ilustrar o processo.
• Resposta: Retorna um dicionário com access_token (o token gerado) e token_type (bearer), seguindo o padrão de autenticação JWT.
• Erro: Lança HTTPException com código 400 se o e-mail não for encontrado.
• Nota: A validação da senha com bcrypt_context.verify será implementada na próxima aula, conforme a transcrição. O token atual é inseguro, pois não é assinado nem tem expiração.

Testando a Rota de Login

Para testar a rota, siga os passos abaixo:

1. Inicie o servidor:uvicorn main:app --reload
2. Acesse a documentação:Abra http://localhost:8000/docs no navegador.
3. Localize a rota POST /auth/login.
4. Envie uma requisição:Use um e-mail de um usuário existente no banco (banco.db). Exemplo:{ "email": "esquimadoidao@teste.com", "senha": "doido123456" }Nota: A senha não é validada nesta aula, mas é incluída no esquema para manter a estrutura esperada.
5. Resposta esperada (assumindo que o usuário com ID 6 existe):{ "access_token": "token_6_random", "token_type": "bearer" }
6. Erro (e-mail não encontrado):Se o e-mail não existir, a resposta será:{ "detail": "Usuário não encontrado" }
7. Verifique o banco:Use a extensão SQLite Viewer no VS Code para confirmar o e-mail e o ID do usuário no banco.db.

Por que Usar JWT?

Os JSON Web Tokens são amplamente utilizados em APIs por suas vantagens:

• Segurança: São assinados com uma chave secreta, garantindo que não foram alterados.
• Escalabilidade: Não requerem estado no servidor, ideal para sistemas distribuídos.
• Flexibilidade: Permitem armazenar informações como user_id ou data de expiração no payload.
• Padrão: Compatíveis com o formato Bearer nos cabeçalhos HTTP (Authorization: Bearer <token>).

No entanto, o token atual (token_<id>_random) é apenas um placeholder didático. Ele não é seguro porque:

• Não é criptografado nem assinado.
• Não inclui data de expiração.
• É previsível (baseado apenas no ID do usuário).

Limitações

A rota de login atual é um ponto de partida, mas precisa de melhorias:

• Validação de Senha: Devemos usar bcrypt_context.verify para comparar a senha fornecida com a senha criptografada no banco.
• JWT Padrão: Substituiremos criar_token por uma implementação com python-jose[cryptography], incluindo assinatura com SECRET_KEY e expiração (ex.: 30 minutos).
• Proteção de Rotas: Criaremos uma dependência para validar tokens e proteger rotas como /pedidos/pedido, garantindo que apenas usuários autenticados possam criar pedidos.

Cuidados Importantes

• Padronização com Esquemas: O LoginSchema garante que os dados de entrada sejam consistentes, facilitando manutenção e documentação.
• Gerenciamento de Sessões: A dependência pegar_sessao assegura que as conexões com o banco sejam fechadas corretamente.
• Segurança Futura: O token simplificado não deve ser usado em produção. A implementação de JWTs padrão é essencial.
• Testes: Sempre verifique o banco (banco.db) para garantir que os dados estão corretos e use a documentação do FastAPI para testar as rotas.

Encerramento da Aula 5

Nesta aula, implementamos a rota /auth/login, que autentica usuários com base no e-mail e retorna um token de acesso simplificado. Criamos o LoginSchema para padronizar os dados de entrada e introduzimos o conceito de JWT, preparando o sistema para autenticação segura.

Caso prefira esse conteúdo no formato de vídeo-aula, assista ao vídeo abaixo ou acesse o nosso canal do YouTube!

Aula 6: Autenticação Completa com Tokens JWT

Nesta aula, vamos implementar a autenticação de usuários com bcrypt para verificar senhas e gerar tokens JWT para controlar acessos. Também introduziremos refresh tokens, permitindo que usuários renovem seus acessos sem precisar fazer login repetidamente, tudo com segurança e escalabilidade.

Para receber por e-mail o(s) arquivo(s) utilizados na aula, preencha:

Como autenticar usuários com bcrypt?

Para autenticar usuários, usamos o bcrypt (definido em main.py com CryptContext) para comparar a senha fornecida no login com a senha criptografada no banco de dados. A função autenticar_usuario verifica o email e a senha, retornando o usuário autenticado ou False se falhar.

Exemplo de código:

from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
from passlib.context import CryptContext

bcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def autenticar_usuario(email: str, password: str, session: Session):
    usuario = session.query(Usuario).filter(Usuario.email == email).first()
    if not usuario:
        return False
    if not bcrypt_context.verify(password, usuario.senha):
        return False
    return usuario

Passos:

1. Busca o usuário pelo email no banco (usando SQLAlchemy).
2. Usa bcrypt_context.verify para comparar a senha fornecida com a hash armazenada.
3. Retorna o usuário se válido; caso contrário, retorna False.

Nota: Usuários antigos sem hash (criados antes da implementação do bcrypt) devem ser deletados do banco, pois não funcionarão.

Como configurar variáveis de ambiente para JWT?

As variáveis de ambiente, definidas no arquivo .env, são carregadas no main.py com python-dotenv. Elas incluem:

• ALGORITHM=HS256: Algoritmo de criptografia.
• SECRET_KEY=uma_chave_secreta: Chave para codificar/decodificar tokens.
• ACCESS_TOKEN_EXPIRE_MINUTES=30: Duração padrão do access token.

Exemplo de .env:

ALGORITHM=HS256
SECRET_KEY=uma_chave_secreta
ACCESS_TOKEN_EXPIRE_MINUTES=30

No main.py, essas variáveis são acessadas:

import os
from dotenv import load_dotenv

load_dotenv()
SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = os.getenv("ALGORITHM")
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES"))

Dica: Converta ACCESS_TOKEN_EXPIRE_MINUTES para int, pois variáveis de ambiente são strings por padrão.

Como criar um token JWT no FastAPI?

A função criar_token gera um token JWT com o python-jose, incluindo o ID do usuário e a data de expiração. A duração do token é configurável, permitindo diferenciar access tokens (30 minutos) e refresh tokens (7 dias).

Exemplo de código:

from jose import JWTError, jwt
from datetime import datetime, timedelta, timezone

def criar_token(user_id: int, duration: timedelta = timedelta(minutes=int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES")))):
    expire = datetime.now(timezone.utc) + duration
    to_encode = {"sub": str(user_id), "exp": expire}
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

Explicação:

1. Define a expiração: datetime.now(timezone.utc) + duration.
2. Cria um dicionário com sub (ID do usuário) e exp (data de expiração).
3. Codifica com jwt.encode, usando SECRET_KEY e ALGORITHM.

Tabela de exemplo:

Como implementar refresh tokens?

Refresh tokens permitem renovar access tokens sem novo login. A rota /auth/refresh recebe um refresh token, verifica sua validade, e gera um novo access token.

Exemplo de código (parcial):

from fastapi import APIRouter, Depends

auth_router = APIRouter()

@auth_router.get("/refresh")
async def use_refresh_token(token: str, session: Session = Depends(get_session)):
    usuario = verificar_token(token, session)
    if not usuario:
        raise HTTPException(status_code=401, detail="Token inválido")
    new_access_token = criar_token(usuario.id)
    return {"access_token": new_access_token, "token_type": "bearer"}

Observação: A função verificar_token (não implementada nesta aula) será ajustada na próxima aula para validar o token e extrair o user_id. Por enquanto, ela retorna um usuário fixo (ID=1) para testes.

Como testar a autenticação?

Para testar o login:

1. Acesse a documentação da API (/docs).
2. Use a rota /auth/login com um usuário válido (ex.: novo_usuario3@teste.com, senha 123456789).
3. Resposta esperada:
   • Sucesso (200): Retorna access_token e refresh_token.
   • Erro (401): “Usuário não encontrado ou credenciais inválidas”.

Exemplo de resposta:

{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "refresh_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "token_type": "bearer"
}

Verifique os tokens em jwt.io. O access token expira em 30 minutos, enquanto o refresh token dura 7 dias.

Encerramento da Aula 6

Implementamos autenticação segura no FastAPI com bcrypt para senhas e python-jose para tokens JWT. Com access tokens de 30 minutos e refresh tokens de 7 dias, criamos um sistema robusto para gerenciar sessões de usuários no nosso projeto de delivery.

Caso prefira esse conteúdo no formato de vídeo-aula, assista ao vídeo abaixo ou acesse o nosso canal do YouTube!

Aula 7: Bloqueio de Endpoints para Usuários Autenticados

Nesta aula vamos aprender a bloquear rotas específicas da API para permitir acesso apenas a usuários autenticados, usando tokens JWT e dependências no FastAPI.

Vamos transformar a verificação de tokens em uma dependência reutilizável, configurar autenticação OAuth2 com tokens Bearer e testar rotas protegidas diretamente na documentação do FastAPI.

Para receber por e-mail o(s) arquivo(s) utilizados na aula, preencha:

O que significa e por que transformar a verificação de token JWT em uma dependência reutilizável no FastAPI?

Para proteger rotas, precisamos verificar se o token JWT enviado pelo usuário é válido e corresponde a um usuário existente. Em vez de repetir esse processo em cada rota, criamos uma dependência reutilizável no arquivo dependencies.py. Essa dependência, chamada verificar_token, valida o token e retorna o usuário autenticado, simplificando o bloqueio de rotas.

Passos para criar a dependência

1. Mover a função verificar_token: Retiramos a função do arquivo auth_routes.py e a colocamos em dependencies.py.
2. Importar dependências necessárias: Incluímos Usuario, Session, Depends, jwt, JWTError, e HTTPException para manipular o banco de dados e tokens.
3. Configurar a dependência: A função usa Depends para receber a sessão do banco e o token como parâmetro.

Código da dependência em dependencies.py

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from sqlalchemy.orm import Session
from models import Usuario
from main import SECRET_KEY, ALGORITHM
from dependencies import pegar_sessao

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login_form")

def verificar_token(token: str = Depends(oauth2_scheme), session: Session = Depends(pegar_sessao)) -> Usuario:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Token inválido")
        usuario = session.query(Usuario).filter(Usuario.id == int(user_id)).first()
        if not usuario:
            raise HTTPException(status_code=401, detail="Usuário não encontrado")
        return usuario
    except JWTError:
        raise HTTPException(status_code=401, detail="Token inválido ou expirado")

Explicação

• oauth2_scheme: Define o esquema OAuth2 com tokens Bearer, apontando para a rota /auth/login_form (explicada adiante).
• jwt.decode: Decodifica o token usando SECRET_KEY e ALGORITHM (definidos em main.py).
• payload.get("sub"): Extrai o ID do usuário (armazenado como sub no token).
• Validação: Verifica se o usuário existe no banco; caso contrário, levanta um erro HTTP 401.
• Erro JWTError: Captura falhas na decodificação (token inválido ou expirado).

Vantagem: Essa dependência pode ser usada em qualquer rota para garantir que apenas usuários autenticados acessem.

Como configurar autenticação OAuth2 com tokens Bearer no FastAPI?

O FastAPI suporta autenticação OAuth2 com tokens Bearer, um padrão seguro para APIs. Tokens Bearer são enviados no cabeçalho HTTP (Authorization: Bearer <token>) em vez de parâmetros no corpo da requisição. Configuramos isso no arquivo main.py e ajustamos a dependência para ler o token do cabeçalho.

Configuração em main.py

from fastapi import FastAPI
from fastapi.security import OAuth2PasswordBearer
from passlib.context import CryptContext
from dotenv import load_dotenv
import os

load_dotenv()
SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = os.getenv("ALGORITHM")
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES"))

app = FastAPI()
bcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login_form")

from auth_routes import auth_router
from order_routes import order_router
app.include_router(auth_router)
app.include_router(order_router)

Explicação:

• OAuth2PasswordBearer: Define o esquema OAuth2, com tokenUrl apontando para a rota que processa o login (neste caso, /auth/login_form).
• Cabeçalho Authorization: O FastAPI extrai automaticamente o token do cabeçalho Authorization: Bearer <token>.

Atualização em auth_routes.py para suportar refresh token

from fastapi import APIRouter, Depends, HTTPException
from models import Usuario
from dependencies import pegar_sessao, verificar_token
from main import bcrypt_context, criar_token
from schemas import UsuarioSchema, LoginSchema
from sqlalchemy.orm import Session

auth_router = APIRouter(prefix="/auth", tags=["auth"])

@auth_router.get("/")
async def home():
    """Rota padrão de autenticação."""
    return {"mensagem": "Você acessou a rota padrão de autenticação", "autenticado": False}

@auth_router.post("/criar_conta")
async def criar_conta(usuario_schema: UsuarioSchema, session: Session = Depends(pegar_sessao)):
    """Cria um novo usuário com senha criptografada."""
    usuario = session.query(Usuario).filter(Usuario.email == usuario_schema.email).first()
    if usuario:
        raise HTTPException(status_code=400, detail="E-mail já cadastrado")
    senha_criptografada = bcrypt_context.hash(usuario_schema.senha)
    novo_usuario = Usuario(
        nome=usuario_schema.nome,
        email=usuario_schema.email,
        senha=senha_criptografada,
        ativo=usuario_schema.ativo,
        admin=usuario_schema.admin
    )
    session.add(novo_usuario)
    session.commit()
    return {"mensagem": f"Usuário cadastrado com sucesso: {usuario_schema.email}"}

@auth_router.post("/login")
async def login(login_schema: LoginSchema, session: Session = Depends(pegar_sessao)):
    """Autentica um usuário e retorna tokens."""
    usuario = session.query(Usuario).filter(Usuario.email == login_schema.email).first()
    if not usuario or not bcrypt_context.verify(login_schema.senha, usuario.senha):
        raise HTTPException(status_code=401, detail="Credenciais inválidas")
    access_token = criar_token(str(usuario.id))
    return {"access_token": access_token, "token_type": "bearer"}

@auth_router.get("/refresh")
async def use_refresh_token(usuario: Usuario = Depends(verificar_token)):
    """Gera um novo access token a partir de um refresh token."""
    new_access_token = criar_token(str(usuario.id))
    return {"access_token": new_access_token, "token_type": "bearer"}

Explicação:

• Rota /refresh: Agora usa a dependência verificar_token, que retorna o usuário autenticado diretamente, sem precisar de um parâmetro token.
• Cabeçalho Bearer: O token é enviado no cabeçalho, validado pela dependência verificar_token.

Como configurar o botão Authorize na documentação do FastAPI

A documentação automática do FastAPI (/docs) inclui um botão Authorize que permite testar rotas protegidas diretamente no navegador. Para isso, criamos uma nova rota /auth/login_form que processa um formulário OAuth2, compatível com o padrão da documentação.

Nova rota em auth_routes.py:

from fastapi.security import OAuth2PasswordRequestForm

@auth_router.post("/login_form")
async def login_form(form_data: OAuth2PasswordRequestForm = Depends(), session: Session = Depends(pegar_sessao)):
    """Processa login via formulário OAuth2 para a documentação."""
    usuario = session.query(Usuario).filter(Usuario.email == form_data.username).first()
    if not usuario or not bcrypt_context.verify(form_data.password, usuario.senha):
        raise HTTPException(status_code=401, detail="Credenciais inválidas")
    access_token = criar_token(str(usuario.id))
    return {"access_token": access_token, "token_type": "bearer"}

Explicação:

• OAuth2PasswordRequestForm: Extrai username (usado para o e-mail) e password do formulário na documentação.
• Validação: Verifica o e-mail e a senha com bcrypt_context.verify.
• Resposta: Retorna apenas o access_token, pois a documentação usa apenas ele.

Ajuste em main.py

Atualize o tokenUrl para apontar para a nova rota:

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login_form")

Testando o botão Authorize

1. Acesse http://localhost:8000/docs.
2. Clique no botão Authorize.
3. Insira o e-mail (ex.: novo_usuario3@test.com) no campo username e a senha (ex.: 123456789) no campo password.
4. Após login, as rotas protegidas (com cadeado fechado) podem ser testadas diretamente.

Diferenças entre /login e /login_form:

Como testar endpoints protegidos usando a biblioteca requests

Para simular requisições externas (ex.: de um frontend), criamos um script de teste usando a biblioteca requests. Isso valida se as rotas protegidas aceitam tokens Bearer corretamente.

Script de teste (test.py):

import requests

url = "http://127.0.0.1:8000/auth/refresh"
headers = {
    "Authorization": "Bearer <seu_refresh_token>"
}
response = requests.get(url, headers=headers)
print(response.json())

Passos

1. Instale a biblioteca: pip install requests.
2. Substitua <seu_refresh_token> por um token válido gerado via /auth/login.
3. Execute o script: python test.py.
4. Verifique a resposta: Deve retornar um novo access_token se o token for válido.

Resolução de erro comum

• Erro: "Subject must be a string" Ocorre porque o sub no token deve ser uma string. Corrigimos em auth_routes.py, na função criar_token:

def criar_token(user_id: int) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES")))
    to_encode = {"sub": str(user_id), "exp": expire}
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

• Correção: Convertemos user_id para string (str(user_id)).

Como resolver erros comuns de requisição e autenticação no FastAPI?

Erros de autenticação, como HTTP 401, são comuns ao configurar tokens. Aqui estão os principais problemas e soluções:

Dica de depuração:

Adicione um log na dependência verificar_token para capturar erros:

try:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
except JWTError as e:
    print(f"Erro JWT: {e}")
    raise HTTPException(status_code=401, detail="Token inválido ou expirado")

Vantagem: O erro aparece nos logs do servidor, facilitando a identificação.

Explicação detalhada do fluxo de autenticação na API

O fluxo de autenticação agora está completo e funciona assim:

1. Criação de conta: Qualquer um pode criar uma conta via /auth/criar_conta (sem autenticação, por design).
2. Login: O usuário faz login via /auth/login (JSON) ou /auth/login_form (formulário OAuth2), recebendo um access_token e, no caso do /login, um refresh_token.
3. Acesso a rotas protegidas: Rotas como /auth/refresh exigem um token Bearer no cabeçalho, validado pela dependência verificar_token.
4. Refresh token: A rota /auth/refresh gera um novo access_token para manter a sessão ativa.
5. Níveis de acesso: A dependência verificar_token retorna o Usuario, permitindo verificar se usuario.admin é True para restringir rotas a administradores.

Fluxo de autenticação:

Encerramento da Aula 7

Nesta aula, implementamos um sistema robusto de autenticação no FastAPI, bloqueando rotas com a dependência verificar_token e configurando OAuth2 com tokens Bearer. Criamos uma rota específica (/auth/login_form) para usar o botão Authorize na documentação, facilitando testes. Também aprendemos a testar rotas protegidas com a biblioteca requests e a resolver erros comuns, como tokens inválidos. Nosso backend agora está seguro e pronto para restringir acessos, inclusive por níveis (ex.: admin).

Caso prefira esse conteúdo no formato de vídeo-aula, assista ao vídeo abaixo ou acesse o nosso canal do YouTube!

Aula 8: Níveis de Acesso e Lazyloaded do Banco de Dados

Na oitava aula, vamos aprender como criar diferentes níveis de acesso para rotas específicas, permitindo que apenas usuários autenticados ou administradores realizem certas ações, como cancelar pedidos.

Além disso, implementaremos a rota de cancelamento de pedidos e resolveremos o problema de Lazy Loading do SQLAlchemy para exibir informações completas nas respostas da API.

Para receber por e-mail o(s) arquivo(s) utilizados na aula, preencha:

Como proteger rotas com dependências globais no APIRouter do FastAPI?

Para garantir que todas as rotas de pedidos sejam acessíveis apenas por usuários autenticados, usamos uma dependência global no APIRouter. No FastAPI, o parâmetro dependencies do APIRouter aplica uma dependência a todas as rotas definidas naquele roteador, sem precisar repeti-la em cada endpoint.

Configuração em order_routes.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from models import Pedido, Usuario
from schemas import PedidoSchema
from dependencies import pegar_sessao, verificar_token

order_router = APIRouter(prefix="/pedidos", tags=["pedidos"], dependencies=[Depends(verificar_token)])

@order_router.post("/")
async def criar_pedido(pedido_schema: PedidoSchema, session: Session = Depends(pegar_sessao)):
    """Cria um novo pedido."""
    novo_pedido = Pedido(
        usuario_id=pedido_schema.usuario_id,
        status="PENDENTE",
        preco=0.0
    )
    session.add(novo_pedido)
    session.commit()
    return {"mensagem": f"Pedido criado com sucesso: ID {novo_pedido.id}"}

@order_router.get("/")
async def listar_pedidos(session: Session = Depends(pegar_sessao)):
    """Lista todos os pedidos."""
    pedidos = session.query(Pedido).all()
    return pedidos

Explicação:

• dependencies=[Depends(verificar_token)]: Aplica a dependência verificar_token a todas as rotas do order_router, exigindo autenticação via token JWT.
• Impacto: Todas as rotas (/pedidos/, /pedidos/cancelar/{pedido_id}, etc.) agora possuem um "cadeado" na documentação (/docs), indicando que exigem autenticação.
• Diferenças com dependências locais:
  • Globais: Aplicadas a todas as rotas do roteador, mas a resposta da dependência (ex.: Usuario) não é acessível nas funções.
  • Locais: Definidas na rota específica, permitindo usar a resposta (ex.: usuario: Usuario = Depends(verificar_token)) como parâmetro.

Vantagem: Simplifica a proteção de múltiplas rotas, garantindo que apenas usuários autenticados acessem os endpoints de pedidos.

Implementando a rota de cancelamento de pedidos na prática

Criamos uma nova rota para cancelar pedidos, usando um parâmetro de caminho (pedido_id) para identificar o pedido a ser cancelado. Essa rota exige autenticação e manipula o status do pedido no banco de dados.

Código da rota em order_routes.py

@order_router.post("/cancelar/{pedido_id}")
async def cancelar_pedido(pedido_id: int, session: Session = Depends(pegar_sessao)):
    """Cancela um pedido específico."""
    pedido = session.query(Pedido).filter(Pedido.id == pedido_id).first()
    if not pedido:
        raise HTTPException(status_code=400, detail="Pedido não encontrado")
    pedido.status = "CANCELADO"
    session.commit()
    return {"mensagem": f"Pedido número {pedido_id} cancelado com sucesso"}

Explicação:

• Método POST: Usamos POST para indicar uma ação que modifica o estado do servidor.
• Parâmetro de caminho {pedido_id}: O ID do pedido é passado na URL (ex.: /pedidos/cancelar/1), como um parâmetro de path.
• Validação: Verifica se o pedido existe; se não, levanta um erro HTTP 400.
• Atualização: Altera o status para "CANCELADO" (em maiúsculas, conforme padrão definido no modelo).
• Commit: Salva a alteração no banco sem precisar de session.add(), pois o pedido já existe.

Teste inicial:

1. Acesse http://localhost:8000/docs.
2. Autentique-se via botão Authorize com um usuário válido (ex.: novo_usuario3@test.com).
3. Execute a rota /pedidos/cancelar/1 com pedido_id=1.
4. Resultado esperado: {"mensagem": "Pedido número 1 cancelado com sucesso"}.

Nota: Inicialmente, qualquer usuário autenticado pode cancelar qualquer pedido, o que será ajustado na próxima seção.

Lógica completa para cancelar pedidos com segurança

Para evitar que qualquer usuário autenticado cancele pedidos de outros, implementamos regras de autorização baseadas no nível de acesso (admin) ou na propriedade do pedido (usuario_id). Isso exige usar o usuário autenticado como parâmetro na rota.

Código atualizado em order_routes.py

@order_router.post("/cancelar/{pedido_id}")
async def cancelar_pedido(pedido_id: int, usuario: Usuario = Depends(verificar_token), session: Session = Depends(pegar_sessao)):
    """Cancela um pedido específico, apenas pelo dono ou admin."""
    pedido = session.query(Pedido).filter(Pedido.id == pedido_id).first()
    if not pedido:
        raise HTTPException(status_code=400, detail="Pedido não encontrado")
    if not usuario.admin and usuario.id != pedido.usuario_id:
        raise HTTPException(status_code=401, detail="Você não tem autorização para fazer essa modificação")
    pedido.status = "CANCELADO"
    session.commit()
    return {"mensagem": f"Pedido número {pedido_id} cancelado com sucesso"}

Explicação:

• Parâmetro usuario: Adicionamos usuario: Usuario = Depends(verificar_token) para receber o usuário autenticado.
• Regras de autorização:
  • Permite cancelar se usuario.admin == True (usuário é administrador).
  • Permite cancelar se usuario.id == pedido.usuario_id (usuário é o dono do pedido).
  • Caso contrário, levanta um erro HTTP 401.
• Lógica do if:
  • Condição: if not usuario.admin and usuario.id != pedido.usuario_id verifica se o usuário não é admin e não é o dono.
  • Se verdadeira, nega o acesso.

Fluxo de autorização

1. O token JWT é validado pela dependência verificar_token.
2. O usuário autenticado é extraído do token.
3. A função verifica o nível de acesso (admin) e a propriedade do pedido (usuario_id).
4. Se autorizado, o pedido é cancelado; caso contrário, retorna erro 401.

Vantagem: Garante que apenas o dono do pedido ou um admin possa cancelá-lo, aumentando a segurança.

Testes práticos de permissões: admin vs. usuário comum

Realizamos testes para validar as regras de autorização, comparando o comportamento de um usuário comum e um administrador.

Cenário 1: Usuário comum tenta cancelar pedido de outro

1. Usuário: novo_usuario3@test.com (ID 5, não admin).
2. Ação: Criar um pedido para o usuário ID 6 via /pedidos/ (JSON: {"usuario_id": 6}).
3. Resultado: Pedido ID 2 criado com sucesso.
4. Ação: Tentar cancelar o pedido ID 2 via /pedidos/cancelar/2.
5. Resultado: Erro 401: {"detail": "Você não tem autorização para fazer essa modificação"}.

Cenário 2: Admin cancela pedido de outro

1. Criar admin:
   • Rota: /auth/criar_conta
   • JSON: {"nome": "Admin", "email": "admin@test.com", "senha": "admin123", "ativo": true, "admin": true}
   • Resultado: Usuário ID 7 criado.
2. Login: Autenticar via /docs com admin@test.com e senha admin123.
3. Ação: Cancelar o pedido ID 2 via /pedidos/cancelar/2.
4. Resultado: Sucesso: {"mensagem": "Pedido número 2 cancelado com sucesso"}.

Verificação no banco

• Tabela pedidos: Pedido ID 2 agora tem status="CANCELADO".
• Tabela usuarios: Usuário ID 7 tem admin=True.

Tabela: Resultados dos testes

O que é Lazy Loading no SQLAlchemy e como afeta sua API

O Lazy Loading é um recurso do SQLAlchemy que carrega dados do banco apenas quando solicitados, otimizando o desempenho. Porém, isso pode causar problemas ao tentar acessar atributos de um objeto após o commit da sessão, como no caso da rota de cancelamento.

Problema identificado

Na rota inicial, tentamos retornar o objeto pedido completo após o session.commit():

return {"mensagem": f"Pedido número {pedido_id} cancelado com sucesso", "pedido": pedido}

• Erro: Após o commit, a sessão é encerrada, e o SQLAlchemy não carrega automaticamente os atributos do pedido, resultando em uma resposta vazia ou erro.

Solução temporária

Acessamos o atributo pedido.id antes do return para forçar o carregamento:

@order_router.post("/cancelar/{pedido_id}")
async def cancelar_pedido(pedido_id: int, usuario: Usuario = Depends(verificar_token), session: Session = Depends(pegar_sessao)):
    """Cancela um pedido específico, apenas pelo dono ou admin."""
    pedido = session.query(Pedido).filter(Pedido.id == pedido_id).first()
    if not pedido:
        raise HTTPException(status_code=400, detail="Pedido não encontrado")
    if not usuario.admin and usuario.id != pedido.usuario_id:
        raise HTTPException(status_code=401, detail="Você não tem autorização para fazer essa modificação")
    pedido.status = "CANCELADO"
    session.commit()
    pedido_id = pedido.id  # Força o carregamento do objeto
    return {"mensagem": f"Pedido número {pedido_id} cancelado com sucesso", "pedido": pedido}

Explicação:

• Forçar carregamento: Acessar pedido.id antes do return faz o SQLAlchemy carregar os atributos do objeto pedido.
• Resposta: O FastAPI serializa automaticamente o objeto pedido como um dicionário, incluindo id, usuario_id, status, e preco.

Resultado:

{
  "mensagem": "Pedido número 2 cancelado com sucesso",
  "pedido": {
    "id": 2,
    "usuario_id": 6,
    "status": "CANCELADO",
    "preco": 0.0
  }
}

Nota: Na próxima aula, implementaremos uma solução mais robusta usando esquemas Pydantic para evitar Lazy Loading e garantir respostas consistentes.

Como aplicar controle de acesso em outras rotas da API

O processo para restringir acesso em outras rotas (ex.: /auth/criar_conta) segue o mesmo padrão:

1. Adicionar dependência verificar_token: Inclua globalmente no APIRouter ou localmente na rota.
2. Receber usuário como parâmetro: Use usuario: Usuario = Depends(verificar_token) na função.
3. Criar regras de autorização: Use condições if para verificar usuario.admin ou outros atributos.

Exemplo: Restringir criação de admins em /auth/criar_conta

@auth_router.post("/criar_conta")
async def criar_conta(usuario_schema: UsuarioSchema, usuario: Usuario = Depends(verificar_token), session: Session = Depends(pegar_sessao)):
    """Cria um novo usuário, permitindo admin apenas por admins."""
    if usuario_schema.admin and not usuario.admin:
        raise HTTPException(status_code=401, detail="Apenas admins podem criar usuários admin")
    usuario_existente = session.query(Usuario).filter(Usuario.email == usuario_schema.email).first()
    if usuario_existente:
        raise HTTPException(status_code=400, detail="E-mail já cadastrado")
    senha_criptografada = bcrypt_context.hash(usuario_schema.senha)
    novo_usuario = Usuario(
        nome=usuario_schema.nome,
        email=usuario_schema.email,
        senha=senha_criptografada,
        ativo=usuario_schema.ativo,
        admin=usuario_schema.admin
    )
    session.add(novo_usuario)
    session.commit()
    return {"mensagem": f"Usuário cadastrado com sucesso: {usuario_schema.email}"}

Explicação:

• Regra: Se usuario_schema.admin == True e o usuário autenticado não é admin (not usuario.admin), levanta erro 401.
• Impacto: Apenas admins podem criar contas com admin=True.

Entendendo o impacto do Lazy Loading nas respostas da API

O Lazy Loading é uma otimização do SQLAlchemy que carrega dados apenas quando acessados, reduzindo a sobrecarga no banco. No entanto, ao retornar objetos diretamente após o commit, podemos enfrentar problemas:

• Por que ocorre? Após session.commit(), a sessão é encerrada, e atributos não carregados explicitamente tornam-se inacessíveis.
• Solução ideal: Usar esquemas Pydantic para serializar os dados antes do commit, garantindo que todas as informações necessárias sejam carregadas.

Exemplo com Pydantic:

from schemas import PedidoResponse

@order_router.post("/cancelar/{pedido_id}")
async def cancelar_pedido(pedido_id: int, usuario: Usuario = Depends(verificar_token), session: Session = Depends(pegar_sessao)):
    pedido = session.query(Pedido).filter(Pedido.id == pedido_id).first()
    if not pedido:
        raise HTTPException(status_code=400, detail="Pedido não encontrado")
    if not usuario.admin and usuario.id != pedido.usuario_id:
        raise HTTPException(status_code=401, detail="Você não tem autorização para fazer essa modificação")
    pedido.status = "CANCELADO"
    session.commit()
    return PedidoResponse.from_orm(pedido)

• Benefício: O esquema PedidoResponse define explicitamente quais campos retornar, evitando Lazy Loading.

Vantagem do Lazy Loading: Reduz consultas desnecessárias, melhorando a performance quando apenas alguns atributos são necessários.

Encerramento da Aula 8

Aprendemos a implementar a rota de cancelamento de pedidos com segurança, usando dependências globais para proteger todas as rotas de pedidos e regras de autorização para restringir ações a donos ou admins. Também vimos como resolver temporariamente o problema de Lazy Loading, garantindo que as informações do pedido sejam exibidas, e aprendemos como aplicar controles de acesso em outras rotas.

Caso prefira esse conteúdo no formato de vídeo-aula, assista ao vídeo abaixo ou acesse o nosso canal do YouTube!

Aula 9: Visualização de Pedidos e Gerenciamento de Itens

Nesta aula, finalizaremos nosso projeto de deliery com FastAPI, criando endpoints para visualizar um pedido específico (incluindo seus itens) e remover itens de um pedido.

Vamos garantir que apenas o dono do pedido ou administradores possam acessar essas funcionalidades, utilizando autenticação com tokens JWT e regras de autorização. Também abordaremos como exibir os itens de um pedido de forma eficiente, utilizando o relacionamento configurado com SQLAlchemy, e resolveremos problemas relacionados ao Lazy Loading com esquemas Pydantic. Por fim, implementaremos a remoção de itens e garantiremos que o preço do pedido seja atualizado automaticamente.

Para receber por e-mail o(s) arquivo(s) utilizados na aula, preencha:

Como visualizar um pedido específico e seus itens?

Para permitir que usuários (donos do pedido ou administradores) visualizem um pedido específico, incluindo todos os seus itens, criaremos um endpoint /pedidos/{pedido_id}.

Este endpoint usará o relacionamento itens definido na classe Pedido para carregar os itens associados e retornará os dados serializados com um esquema Pydantic para evitar problemas de Lazy Loading.

Configuração do endpoint em order_routes.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from models import Pedido, Usuario
from schemas import PedidoResponse
from dependencies import pegar_sessao, verificar_token

order_router = APIRouter(prefix="/pedidos", tags=["pedidos"], dependencies=[Depends(verificar_token)])

@order_router.get("/{pedido_id}")
async def visualizar_pedido(pedido_id: int, usuario: Usuario = Depends(verificar_token), session: Session = Depends(pegar_sessao)):
    """Visualiza um pedido específico, incluindo seus itens, apenas pelo dono ou admin."""
    pedido = session.query(Pedido).filter(Pedido.id == pedido_id).first()
    if not pedido:
        raise HTTPException(status_code=400, detail="Pedido não encontrado")
    if not usuario.admin and usuario.id != pedido.usuario_id:
        raise HTTPException(status_code=401, detail="Você não tem autorização para visualizar este pedido")
    return PedidoResponse.from_orm(pedido)

Explicação

• Rota GET /pedidos/{pedido_id}: Recebe o ID do pedido como parâmetro de caminho.
• Autenticação e autorização:
  • Usa a dependência verificar_token para garantir que o usuário está autenticado.
  • Verifica se o usuário é administrador (usuario.admin == True) ou o dono do pedido (usuario.id == pedido.usuario_id).
  • Caso contrário, levanta um erro HTTP 401.
• Consulta ao banco: Busca o pedido com session.query(Pedido).filter(Pedido.id == pedido_id).first().
• Resposta com Pydantic: Usa o esquema PedidoResponse para serializar o objeto Pedido, incluindo o relacionamento itens, evitando problemas de Lazy Loading.

Esquema Pydantic em schemas.py

Para garantir que os itens do pedido sejam incluídos na resposta, criamos um esquema ItemPedidoResponse e atualizamos PedidoResponse.

from pydantic import BaseModel
from typing import List

class ItemPedidoResponse(BaseModel):
    id: int
    quantidade: int
    sabor: str
    tamanho: str
    preco_unitario: float
    pedido_id: int

    class Config:
        from_attributes = True

class PedidoResponse(BaseModel):
    id: int
    usuario_id: int
    status: str
    preco: float
    itens: List[ItemPedidoResponse] = []

    class Config:
        from_attributes = True

Explicação do esquema

• ItemPedidoResponse: Define os campos de um item de pedido (id, quantidade, sabor, tamanho, preco_unitario, pedido_id).
• PedidoResponse: Inclui os campos do pedido (id, usuario_id, status, preco) e uma lista de ItemPedidoResponse para os itens, mapeada pelo relacionamento itens.
• from_attributes = True: Permite serializar objetos SQLAlchemy diretamente, carregando os dados do relacionamento itens.

Testando o endpoint

1. Autentique-se:
   • Acesse http://localhost:8000/docs.
   • Use o botão Authorize com um usuário admin (ex.: admin@test.com, senha admin123) ou o dono do pedido.
2. Execute a rota:
   • Acesse /pedidos/3 (supondo que o pedido ID 3 exista).
   • Resposta esperada:{ "id": 3, "usuario_id": 2, "status": "PENDENTE", "preco": 230.0, "itens": [ { "id": 1, "quantidade": 2, "sabor": "PEPERONI", "tamanho": "GIGA", "preco_unitario": 70.0, "pedido_id": 3 }, { "id": 2, "quantidade": 2, "sabor": "MARGUERITA", "tamanho": "PEQUENA", "preco_unitario": 45.0, "pedido_id": 3 } ] }
3. Teste de erro:
   • Tente acessar /pedidos/3 com um usuário não autorizado (ex.: novo_usuario3@test.com, ID 5, não admin).
   • Resposta: {"detail": "Você não tem autorização para visualizar este pedido"}.

Como remover um item de um pedido?

Para permitir a remoção de itens de um pedido, criaremos um endpoint /pedidos/remover_item/{pedido_id}/{item_id}. Este endpoint verificará se o usuário é o dono do pedido ou admin, removerá o item especificado e atualizará o preço do pedido automaticamente.

Configuração do endpoint em order_routes.py

@order_router.delete("/remover_item/{pedido_id}/{item_id}")
async def remover_item_pedido(pedido_id: int, item_id: int, usuario: Usuario = Depends(verificar_token), session: Session = Depends(pegar_sessao)):
    """Remove um item de um pedido específico, apenas pelo dono ou admin, e atualiza o preço."""
    pedido = session.query(Pedido).filter(Pedido.id == pedido_id).first()
    if not pedido:
        raise HTTPException(status_code=400, detail="Pedido não encontrado")
    if not usuario.admin and usuario.id != pedido.usuario_id:
        raise HTTPException(status_code=401, detail="Você não tem autorização para modificar este pedido")
    
    item = session.query(ItemPedido).filter(ItemPedido.id == item_id, ItemPedido.pedido_id == pedido_id).first()
    if not item:
        raise HTTPException(status_code=400, detail="Item não encontrado no pedido")
    
    session.delete(item)
    pedido.calcular_preco()
    session.commit()
    
    return {"mensagem": f"Item {item_id} removido do pedido {pedido_id} com sucesso", "preco_atualizado": pedido.preco}

Explicação

• Rota DELETE /pedidos/remover_item/{pedido_id}/{item_id}: Recebe o ID do pedido e o ID do item como parâmetros de caminho.
• Validações:
  • Verifica se o pedido existe (session.query(Pedido)).
  • Confirma se o usuário é admin ou o dono do pedido.
  • Busca o item específico com ItemPedido.id == item_id e ItemPedido.pedido_id == pedido_id para garantir que o item pertence ao pedido.
• Remoção e atualização:
  • Remove o item com session.delete(item).
  • Chama pedido.calcular_preco() para atualizar o preço do pedido com base nos itens restantes.
  • Comita as alterações com session.commit().
• Resposta: Retorna uma mensagem de sucesso e o preço atualizado do pedido.

Função calcular_preco no modelo Pedido

A função calcular_preco já foi definida na Aula 8, mas garantimos que ela está correta no modelo Pedido em models.py:

from sqlalchemy import Column, Integer, String, Float, ForeignKey
from sqlalchemy.orm import relationship
from database import Base

class Pedido(Base):
    __tablename__ = "pedidos"
    id = Column(Integer, primary_key=True, index=True)
    usuario_id = Column(Integer, ForeignKey("usuarios.id"))
    status = Column(String, index=True)
    preco = Column(Float, default=0.0)
    itens = relationship("ItemPedido", cascade="all,delete")

    def calcular_preco(self):
        self.preco = sum(item.preco_unitario * item.quantidade for item in self.itens)

Explicação

• Relacionamento itens: Usa relationship para acessar todos os itens associados ao pedido.
• Função calcular_preco:
  • Usa list comprehension para somar preco_unitario * quantidade de cada item em self.itens.
  • Atualiza o campo preco do pedido.
• Cascade all,delete: Garante que, ao deletar um pedido, todos os itens associados são removidos automaticamente.

Testando o endpoint de remoção

1. Autentique-se:
   • Use um usuário admin ou o dono do pedido ID 3 (ex.: admin@test.com).
2. Execute a rota:
   • Acesse /pedidos/remover_item/3/2 para remover o item ID 2 (pizza marguerita) do pedido ID 3.
   • Resposta esperada:{ "mensagem": "Item 2 removido do pedido 3 com sucesso", "preco_atualizado": 140.0 }
   • Explicação: O preço do pedido é atualizado para 140.0 (2 pizzas peperoni × 70.0).
3. Verifique no banco:
   • Tabela itens_pedidos: O item ID 2 foi removido.
   • Tabela pedidos: O pedido ID 3 tem preco = 140.0.
4. Teste de erro:
   • Tente remover com um usuário não autorizado: Erro 401.
   • Tente remover um item inexistente: Erro 400 ("Item não encontrado no pedido").

Como evitar problemas de Lazy Loading?

O uso do esquema PedidoResponse com ItemPedidoResponse garante que os dados dos itens sejam carregados e serializados corretamente antes do session.commit(), evitando erros de Lazy Loading. A configuração from_attributes = True permite que o Pydantic converta objetos SQLAlchemy em dicionários, incluindo relacionamentos.

Comparação: Com e sem Pydantic

Recomendação: Sempre use esquemas Pydantic para endpoints que retornam objetos complexos com relacionamentos.

Testes práticos: Admin vs. Usuário comum

Cenário 1: Usuário comum visualiza/remova item de outro usuário

• Usuário: novo_usuario3@test.com (ID 5, não admin).
• Ação: Tentar visualizar /pedidos/3 (pertence ao usuário ID 2).
• Resultado: Erro 401: {"detail": "Você não tem autorização para visualizar este pedido"}.
• Ação: Tentar remover item via /pedidos/remover_item/3/1.
• Resultado: Erro 401: {"detail": "Você não tem autorização para modificar este pedido"}.

Cenário 2: Admin visualiza/remove item

• Usuário: admin@test.com (ID 7, admin).
• Ação: Visualizar /pedidos/3.
• Resultado: Sucesso, retorna o pedido com seus itens.
• Ação: Remover item via /pedidos/remover_item/3/1.
• Resultado: Sucesso: {"mensagem": "Item 1 removido do pedido 3 com sucesso", "preco_atualizado": 0.0} (se for o último item).

Resolução de erros comuns

Dica de depuração: Adicione logs na função calcular_preco:

def calcular_preco(self):
    print(f"Calculando preço para pedido {self.id} com {len(self.itens)} itens")
    self.preco = sum(item.preco_unitario * item.quantidade for item in self.itens)
    print(f"Preço atualizado: {self.preco}")

Fluxo completo de gerenciamento de pedidos

Encerramento da Aula 9

Nesta aula, implementamos endpoints para visualizar pedidos específicos (com seus itens) e remover itens de pedidos, garantindo segurança com autenticação e autorização. Usamos esquemas Pydantic para evitar problemas de Lazy Loading e mantivemos o preço do pedido atualizado com a função calcular_preco. Nosso backend agora suporta um fluxo completo de gerenciamento de pedidos, com restrições de acesso para donos e administradores.

Na próxima aula: Criaremos um endpoint para finalizar pedidos, alterando o status para “FINALIZADO” e validando que todos os itens estão corretos. Também implementaremos paginação no endpoint /pedidos/listar para suportar sistemas com muitos pedidos.

Caso prefira esse conteúdo no formato de vídeo-aula, assista ao vídeo abaixo ou acesse o nosso canal do YouTube!

Aula 10: Finalização e Schemas para respostas de endpoints

Nesta aula, finalizamos o desenvolvimento de um backend completo para um sistema de delivery de pizzaria usando FastAPI. Vamos aprender a criar as últimas rotas da API, incluindo a remoção de itens de pedidos com controle de acesso, finalização de pedidos, visualização de pedidos específicos e listagem de pedidos por usuário. Além disso, abordaremos como padronizar respostas usando esquemas Pydantic e resolver problemas de Lazy Loading do SQLAlchemy. Tudo isso com exemplos práticos e linguagem acessível para iniciantes.

Para receber por e-mail o(s) arquivo(s) utilizados na aula, preencha:

Como remover itens de um pedido com controle de acesso?

A remoção de itens de pedidos exige verificar se o usuário é o dono do pedido ou um administrador, garantindo segurança. Usamos o ID do item do pedido para identificar o que será removido, atualizamos o preço do pedido e tratamos o Lazy Loading para exibir informações completas.

@order_router.post("/remover_item/{item_pedido_id}")
async def remover_item_pedido(item_pedido_id: int, usuario: Usuario = Depends(verificar_token), session: Session = Depends(pegar_sessao)):
    item_pedido = session.query(ItemPedido).filter(ItemPedido.id == item_pedido_id).first()
    if not item_pedido:
        raise HTTPException(status_code=400, detail="Item do pedido não encontrado")
    pedido = session.query(Pedido).filter(Pedido.id == item_pedido.pedido_id).first()
    if not usuario.admin and usuario.id != pedido.usuario_id:
        raise HTTPException(status_code=401, detail="Você não tem autorização para fazer essa modificação")
    session.delete(item_pedido)
    pedido.calcular_preco()
    session.commit()
    quantidade_itens = len(pedido.itens)
    return {
        "mensagem": "Item removido com sucesso",
        "pedido": {
            "id": pedido.id,
            "preco": pedido.preco,
            "quantidade_itens": quantidade_itens
        }
    }

Tabela: Resultados da remoção de itens

Como finalizar um pedido com segurança?

A rota para finalizar um pedido altera o status para "FINALIZADO", garantindo que apenas o dono ou um admin possa executar a ação.

@order_router.post("/finalizar/{pedido_id}")
async def finalizar_pedido(pedido_id: int, usuario: Usuario = Depends(verificar_token), session: Session = Depends(pegar_sessao)):
    pedido = session.query(Pedido).filter(Pedido.id == pedido_id).first()
    if not pedido:
        raise HTTPException(status_code=400, detail="Pedido não encontrado")
    if not usuario.admin and usuario.id != pedido.usuario_id:
        raise HTTPException(status_code=401, detail="Você não tem autorização para fazer essa modificação")
    pedido.status = "FINALIZADO"
    session.commit()
    return {"mensagem": f"Pedido número {pedido_id} finalizado com sucesso", "pedido": pedido}

Como visualizar um pedido específico com detalhes?

A rota para visualizar um pedido específico retorna todas as informações do pedido, incluindo itens, mas apenas para o dono ou admins.

@order_router.get("/{pedido_id}")
async def visualizar_pedido(pedido_id: int, usuario: Usuario = Depends(verificar_token), session: Session = Depends(pegar_sessao)):
    pedido = session.query(Pedido).filter(Pedido.id == pedido_id).first()
    if not pedido:
        raise HTTPException(status_code=400, detail="Pedido não encontrado")
    if not usuario.admin and usuario.id != pedido.usuario_id:
        raise HTTPException(status_code=401, detail="Você não tem autorização para visualizar este pedido")
    quantidade_itens = len(pedido.itens)
    return {"pedido": pedido, "quantidade_itens": quantidade_itens}

Como listar todos os pedidos de um usuário?

A rota para listar os pedidos de um usuário autenticado filtra os pedidos pelo usuario_id extraído do token JWT, usando esquemas Pydantic para padronizar a resposta.

from typing import List
from schemas import PedidoResponse

@order_router.get("/usuario", response_model=List[PedidoResponse])
async def listar_pedidos_usuario(usuario: Usuario = Depends(verificar_token), session: Session = Depends(pegar_sessao)):
    pedidos = session.query(Pedido).filter(Pedido.usuario_id == usuario.id).all()
    return pedidos

Código do esquema em schemas.py:

from pydantic import BaseModel
from typing import List, Optional

class ItemPedidoResponse(BaseModel):
    quantidade: int
    sabor: str
    tamanho: str
    preco_unitario: float

    class Config:
        from_attributes = True

class PedidoResponse(BaseModel):
    id: int
    status: str
    preco: float
    itens: Optional[List[ItemPedidoResponse]] = []

    class Config:
        from_attributes = True

Tabela: Exemplo de resposta da rota

Como padronizar respostas com esquemas Pydantic?

Os esquemas Pydantic permitem personalizar as respostas da API, definindo quais campos exibir e evitando informações desnecessárias, como o usuario_id em /pedidos/usuario.

• Controle de dados: Escolha quais campos exibir (ex.: omitir usuario_id).
• Evita Lazy Loading: O from_attributes=True mapeia objetos do SQLAlchemy para o esquema.
• Flexibilidade: Suporta listas e relações, como itens em PedidoResponse.

Dica: Use Optional[List[ItemPedidoResponse]] para campos que podem estar vazios, como itens.

Encerramento da aula 10

Completamos o backend de um sistema de delivery usando FastAPI, implementando rotas para remover itens, finalizar pedidos, visualizar pedidos específicos e listar pedidos por usuário. Aprendemos a usar esquemas Pydantic para padronizar respostas e contornar problemas de Lazy Loading. O próximo passo é conectar este backend a um frontend (ex.: com JavaScript e Fetch API) ou publicar a API com documentação clara. Experimente criar um frontend simples para testar essas rotas!

Caso prefira esse conteúdo no formato de vídeo-aula, assista ao vídeo abaixo ou acesse o nosso canal do YouTube!

Conclusão e resumo do curso

Ao longo de 10 episódios, desenvolvemos um backend robusto para um sistema de delivery de pizzaria usando FastAPI, abrangendo desde os fundamentos até recursos avançados, resultando em uma REST API escalável e pronta para integração ou publicação.

Veja um resumo de tudo o que aprendemos:

• Padronização de respostas: Usamos schemas Pydantic para personalizar e estruturar as respostas dos endpoints, garantindo consistência.
• Rotas para pedidos e autenticação: Criamos endpoints para gerenciar pedidos e autenticação de usuários, garantindo funcionalidades essenciais do sistema.
• Integração com banco de dados: Utilizamos SQLAlchemy para conectar e gerenciar o banco de dados, incluindo definição de modelos e migrações.
• Contas de usuário com segurança: Implementamos criação de contas com schemas Pydantic e criptografia para proteger dados sensíveis.
• Autenticação com JWT: Desenvolvemos login e autenticação completa usando tokens JWT, assegurando acesso seguro aos endpoints.
• Bloqueio de endpoints: Restringimos o acesso a rotas, permitindo apenas usuários autenticados, com controle de permissões.
• Níveis de acesso: Gerenciamos permissões para administradores e clientes, personalizando o acesso aos recursos da API.
• Otimização com Lazy Loading: Resolvemos problemas de Lazy Loading no SQLAlchemy para melhorar a eficiência das consultas ao banco.
• Gerenciamento de pedidos: Implementamos visualização, adição e remoção de itens de pedidos, com atualizações dinâmicas de preços.
• Finalização de pedidos: Criamos rotas para alterar o status de pedidos para "FINALIZADO", com validações de autorização.

Hashtag Treinamentos

Para acessar outras publicações de Python, clique aqui!

Posts mais recentes de Python

• 
  Agente de IA local com Python: crie o seu com Ollama11 de novembro de 2026
  Aprenda a criar um agente de IA local com Python, Ollama e LangChain: rode modelos no seu PC, sem pagar por API e sem expor os dados da empresa.
• 
  Como Criar CRM com Codex no VS Code usando Python e Flask6 de novembro de 2026
  Aprenda a criar CRM com Codex no VS Code usando Python, Flask e SQLite, com login, orçamentos e PDF, em fases com testes e commits no Git.
• 
  Curso Básico de Python: Saia do Zero em 10 Aulas Grátis18 de setembro de 2026
  Aprenda com o curso básico de Python gratuito da Hashtag: 10 aulas práticas, exercícios e projetos reais para sair do zero em programação.

Posts mais recentes da Hashtag Treinamentos

• 
  Como usar o Excel para saber se uma mudança melhorou de verdade13 de novembro de 2026
  Aprenda a montar no Excel um gráfico de tendência com a mediana e a regra dos 6 pontos para saber se uma mudança melhorou de verdade ou se é ruído.
• 
  Agente de IA local com Python: crie o seu com Ollama11 de novembro de 2026
  Aprenda a criar um agente de IA local com Python, Ollama e LangChain: rode modelos no seu PC, sem pagar por API e sem expor os dados da empresa.
• 
  Claude Dashboards: Como Usar Passo a Passo6 de novembro de 2026
  Aprenda a usar o Claude Dashboards: conecte o Google Sheets, crie um dashboard com IA, confira de onde vem cada número e atualize os dados em um clique.

Não vamos te encaminhar nenhum tipo de SPAM! A Hashtag Treinamentos é uma empresa preocupada com a proteção de seus dados e realiza o tratamento de acordo com a Lei Geral de Proteção de Dados (Lei n. 13.709/18). Qualquer dúvida, nos contate.