
Eu sei, eu sei, você provavelmente está pensando "o que, de novo?!"
Sim, no Habré eles têm já escrito sobre os FastAPI quadro muitas vezes . Mas proponho considerar esta ferramenta um pouco mais detalhadamente e escrever a API do seu próprio mini Habr sem carma e avaliações, mas
Esquema de banco de dados e migrações
Em primeiro lugar, usando SQLAlchemy Expression Language , descreveremos o esquema do banco de dados. Vamos criar um arquivo models / users.py :
import sqlalchemy
from sqlalchemy.dialects.postgresql import UUID
metadata = sqlalchemy.MetaData()
users_table = sqlalchemy.Table(
"users",
metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
sqlalchemy.Column("email", sqlalchemy.String(40), unique=True, index=True),
sqlalchemy.Column("name", sqlalchemy.String(100)),
sqlalchemy.Column("hashed_password", sqlalchemy.String()),
sqlalchemy.Column(
"is_active",
sqlalchemy.Boolean(),
server_default=sqlalchemy.sql.expression.true(),
nullable=False,
),
)
tokens_table = sqlalchemy.Table(
"tokens",
metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
sqlalchemy.Column(
"token",
UUID(as_uuid=False),
server_default=sqlalchemy.text("uuid_generate_v4()"),
unique=True,
nullable=False,
index=True,
),
sqlalchemy.Column("expires", sqlalchemy.DateTime()),
sqlalchemy.Column("user_id", sqlalchemy.ForeignKey("users.id")),
)
E o arquivo models / posts.py :
import sqlalchemy
from .users import users_table
metadata = sqlalchemy.MetaData()
posts_table = sqlalchemy.Table(
"posts",
metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
sqlalchemy.Column("user_id", sqlalchemy.ForeignKey(users_table.c.id)),
sqlalchemy.Column("created_at", sqlalchemy.DateTime()),
sqlalchemy.Column("title", sqlalchemy.String(100)),
sqlalchemy.Column("content", sqlalchemy.Text()),
)
Para automatizar as migrações do banco de dados, instale o alambique :
$ pip install alembic
Para inicializar o Alembic, execute:
$ alembic init migrations
Este comando criará no diretório atual o arquivo alembic.ini e um diretório de migração contendo:
- diretório de versões onde os arquivos de migração serão armazenados
- script env.py executado quando o alambique é chamado
- um arquivo script.py.mako contendo um modelo para novas migrações.
Indicaremos a url de nosso banco de dados, para isso, no arquivo alembic.ini, adicione a linha:
sqlalchemy.url = postgresql://%(DB_USER)s:%(DB_PASS)s@%(DB_HOST)s:5432/%(DB_NAME)s
O formato % (variable_name) s nos permite definir diferentes valores de variáveis dependendo do ambiente, substituindo-as no arquivo env.py desta forma:
from os import environ
from alembic import context
from app.models import posts, users
# Alembic Config
# alembic.ini
config = context.config
section = config.config_ini_section
config.set_section_option(section, "DB_USER", environ.get("DB_USER"))
config.set_section_option(section, "DB_PASS", environ.get("DB_PASS"))
config.set_section_option(section, "DB_NAME", environ.get("DB_NAME"))
config.set_section_option(section, "DB_HOST", environ.get("DB_HOST"))
fileConfig(config.config_file_name)
target_metadata = [users.metadata, posts.metadata]
Aqui pegamos os valores de DB_USER, DB_PASS, DB_NAME e DB_HOST das variáveis de ambiente. Além disso, o arquivo env.py especifica os metadados de nosso banco de dados no atributo target_metadata , sem o qual o Alembic não será capaz de determinar quais alterações precisam ser feitas no banco de dados.
Está tudo pronto e podemos gerar migrações e atualizar o banco de dados:
$ alembic revision --autogenerate -m "Added required tables"
$ alembic upgrade head
Lançamos o aplicativo e conectamos o banco de dados
Vamos criar um arquivo main.py :
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
E inicie o aplicativo executando o comando:
$ uvicorn main:app --reload
Vamos garantir que tudo funcione como deveria. Abra http://127.0.0.1:8000/ no navegador e veja
{"Hello": "World"}
Para se conectar ao banco de dados, usaremos o módulo de bancos de dados , que nos permite executar consultas de forma assíncrona.
Vamos configurar os eventos de inicialização e desligamento do nosso serviço, nos quais ocorrerá a conexão e desconexão do banco de dados. Vamos editar o arquivo main.py :
from os import environ
import databases
#
DB_USER = environ.get("DB_USER", "user")
DB_PASSWORD = environ.get("DB_PASSWORD", "password")
DB_HOST = environ.get("DB_HOST", "localhost")
DB_NAME = "async-blogs"
SQLALCHEMY_DATABASE_URL = (
f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:5432/{DB_NAME}"
)
# database,
database = databases.Database(SQLALCHEMY_DATABASE_URL)
app = FastAPI()
@app.on_event("startup")
async def startup():
#
await database.connect()
@app.on_event("shutdown")
async def shutdown():
#
await database.disconnect()
@app.get("/")
async def read_root():
# ,
query = (
select(
[
posts_table.c.id,
posts_table.c.created_at,
posts_table.c.title,
posts_table.c.content,
posts_table.c.user_id,
users_table.c.name.label("user_name"),
]
)
.select_from(posts_table.join(users_table))
.order_by(desc(posts_table.c.created_at))
)
return await database.fetch_all(query)
Abrimos http://127.0.0.1:8000/ e se virmos uma lista vazia [] na resposta , tudo correu bem e podemos seguir em frente.
Validação de solicitação e resposta
Implementaremos a possibilidade de cadastro do usuário. Para fazer isso, precisamos validar as solicitações e respostas HTTP. Para resolver este problema, usaremos a biblioteca pydantic :
pip install pydantic
Crie um arquivo schemas / users.py e adicione um modelo que é responsável por validar o corpo da solicitação:
from pydantic import BaseModel, EmailStr
class UserCreate(BaseModel):
""" sign-up """
email: EmailStr
name: str
password: str
Observe que os tipos de campo são definidos usando a anotação de tipo. Além dos tipos de dados integrados, como int e str , o pydantic oferece um grande número de tipos que fornecem validação adicional. Por exemplo, o tipo EmailStr verifica se o valor recebido é um email válido. Para usar o tipo EmailStr , você precisa instalar o módulo validador de e-mail :
pip install email-validator
O corpo da resposta deve conter seus próprios campos específicos, por exemplo id e access_token , então vamos adicionar os modelos responsáveis por gerar a resposta ao arquivo schemas / users.py :
from typing import Optional
from pydantic import UUID4, BaseModel, EmailStr, Field, validator
class UserCreate(BaseModel):
""" sign-up """
email: EmailStr
name: str
password: str
class UserBase(BaseModel):
""" """
id: int
email: EmailStr
name: str
class TokenBase(BaseModel):
token: UUID4 = Field(..., alias="access_token")
expires: datetime
token_type: Optional[str] = "bearer"
class Config:
allow_population_by_field_name = True
@validator("token")
def hexlify_token(cls, value):
""" UUID hex """
return value.hex
class User(UserBase):
""" """
token: TokenBase = {}
Para cada campo do modelo, você pode escrever um validador personalizado . Por exemplo, hexlify_token converte o valor UUID em uma string hexadecimal. É importante notar que você pode usar a classe Field quando precisar substituir o comportamento padrão de um campo de modelo. Por exemplo, token: UUID4 = Field (..., alias = "access_token") define o alias access_token para o campo de token . Para indicar que o campo é obrigatório, um valor especial - ... ( reticências ) é passado como o primeiro parâmetro .
Adicione o arquivo utils / users.py , no qual criaremos os métodos necessários para gravar um usuário no banco de dados:
import hashlib
import random
import string
from datetime import datetime, timedelta
from sqlalchemy import and_
from app.models.database import database
from app.models.users import tokens_table, users_table
from app.schemas import users as user_schema
def get_random_string(length=12):
""" , """
return "".join(random.choice(string.ascii_letters) for _ in range(length))
def hash_password(password: str, salt: str = None):
""" """
if salt is None:
salt = get_random_string()
enc = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 100_000)
return enc.hex()
def validate_password(password: str, hashed_password: str):
""" , """
salt, hashed = hashed_password.split("$")
return hash_password(password, salt) == hashed
async def get_user_by_email(email: str):
""" """
query = users_table.select().where(users_table.c.email == email)
return await database.fetch_one(query)
async def get_user_by_token(token: str):
""" """
query = tokens_table.join(users_table).select().where(
and_(
tokens_table.c.token == token,
tokens_table.c.expires > datetime.now()
)
)
return await database.fetch_one(query)
async def create_user_token(user_id: int):
""" user_id """
query = (
tokens_table.insert()
.values(expires=datetime.now() + timedelta(weeks=2), user_id=user_id)
.returning(tokens_table.c.token, tokens_table.c.expires)
)
return await database.fetch_one(query)
async def create_user(user: user_schema.UserCreate):
""" """
salt = get_random_string()
hashed_password = hash_password(user.password, salt)
query = users_table.insert().values(
email=user.email, name=user.name, hashed_password=f"{salt}${hashed_password}"
)
user_id = await database.execute(query)
token = await create_user_token(user_id)
token_dict = {"token": token["token"], "expires": token["expires"]}
return {**user.dict(), "id": user_id, "is_active": True, "token": token_dict}
Vamos criar um arquivo routers / users.py e adicionar uma rota de inscrição , indicando que espera um modelo CreateUser na solicitação e retorna um modelo de usuário :
from fastapi import APIRouter
from app.schemas import users
from app.utils import users as users_utils
router = APIRouter()
@router.post("/sign-up", response_model=users.User)
async def create_user(user: users.UserCreate):
db_user = await users_utils.get_user_by_email(email=user.email)
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
return await users_utils.create_user(user=user)
Resta apenas conectar as rotas do arquivo routers / users.py . Para fazer isso, adicione as seguintes linhas a main.py :
from app.routers import users
app.include_router(users.router)
Autenticação e controle de acesso
Agora que temos usuários em nosso banco de dados, estamos prontos para configurar a autenticação do aplicativo. Vamos adicionar um endpoint que pega um nome de usuário e senha e retorna um token. Atualize o arquivo routers / users.py para adicionar :
from fastapi import Depends
from fastapi.security import OAuth2PasswordRequestForm
@router.post("/auth", response_model=users.TokenBase)
async def auth(form_data: OAuth2PasswordRequestForm = Depends()):
user = await users_utils.get_user_by_email(email=form_data.username)
if not user:
raise HTTPException(status_code=400, detail="Incorrect email or password")
if not users_utils.validate_password(
password=form_data.password, hashed_password=user["hashed_password"]
):
raise HTTPException(status_code=400, detail="Incorrect email or password")
return await users_utils.create_user_token(user_id=user["id"])
Ao mesmo tempo, não precisamos descrever o modelo de solicitação nós mesmos, Fastapi fornece uma classe de dependência especial OAuth2PasswordRequestForm , que faz com que a rota espere dois campos de nome de usuário e senha.
Para restringir o acesso a certas rotas para usuários não autenticados, iremos escrever um método de dependência. Ele verificará se o token fornecido pertence ao usuário ativo e retornará os detalhes do usuário. Isso nos permitirá usar as informações do usuário em todas as rotas que exigem autenticação. Vamos criar um arquivo utils / dependecies.py :
from app.utils import users as users_utils
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth")
async def get_current_user(token: str = Depends(oauth2_scheme)):
user = await users_utils.get_user_by_token(token)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
)
if not user["is_active"]:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user"
)
return user
Observe que uma dependência pode, por sua vez, depender de outra dependência. Por exemplo, OAuth2PasswordBearer é uma dependência que deixa claro para FastAPI que a rota atual requer autenticação.
Para verificar se tudo funciona conforme o esperado, adicione a rota / users / me , que retorna os detalhes do usuário atual. Adicione as linhas a roteadores / users.py :
from app.utils.dependencies import get_current_user
@router.get("/users/me", response_model=users.UserBase)
async def read_users_me(current_user: users.User = Depends(get_current_user)):
return current_user
Agora temos a rota / users / me , à qual apenas usuários autenticados têm acesso.
Tudo está pronto para finalmente adicionar a capacidade dos usuários de criar e editar publicações:
utils / posts.py
from datetime import datetime
from app.models.database import database
from app.models.posts import posts_table
from app.models.users import users_table
from app.schemas import posts as post_schema
from sqlalchemy import desc, func, select
async def create_post(post: post_schema.PostModel, user):
query = (
posts_table.insert()
.values(
title=post.title,
content=post.content,
created_at=datetime.now(),
user_id=user["id"],
)
.returning(
posts_table.c.id,
posts_table.c.title,
posts_table.c.content,
posts_table.c.created_at,
)
)
post = await database.fetch_one(query)
# Convert to dict and add user_name key to it
post = dict(zip(post, post.values()))
post["user_name"] = user["name"]
return post
async def get_post(post_id: int):
query = (
select(
[
posts_table.c.id,
posts_table.c.created_at,
posts_table.c.title,
posts_table.c.content,
posts_table.c.user_id,
users_table.c.name.label("user_name"),
]
)
.select_from(posts_table.join(users_table))
.where(posts_table.c.id == post_id)
)
return await database.fetch_one(query)
async def get_posts(page: int):
max_per_page = 10
offset1 = (page - 1) * max_per_page
query = (
select(
[
posts_table.c.id,
posts_table.c.created_at,
posts_table.c.title,
posts_table.c.content,
posts_table.c.user_id,
users_table.c.name.label("user_name"),
]
)
.select_from(posts_table.join(users_table))
.order_by(desc(posts_table.c.created_at))
.limit(max_per_page)
.offset(offset1)
)
return await database.fetch_all(query)
async def get_posts_count():
query = select([func.count()]).select_from(posts_table)
return await database.fetch_val(query)
async def update_post(post_id: int, post: post_schema.PostModel):
query = (
posts_table.update()
.where(posts_table.c.id == post_id)
.values(title=post.title, content=post.content)
)
return await database.execute(query)
routers / posts.py
from app.schemas.posts import PostDetailsModel, PostModel
from app.schemas.users import User
from app.utils import posts as post_utils
from app.utils.dependencies import get_current_user
from fastapi import APIRouter, Depends, HTTPException, status
router = APIRouter()
@router.post("/posts", response_model=PostDetailsModel, status_code=201)
async def create_post(post: PostModel, current_user: User = Depends(get_current_user)):
post = await post_utils.create_post(post, current_user)
return post
@router.get("/posts")
async def get_posts(page: int = 1):
total_cout = await post_utils.get_posts_count()
posts = await post_utils.get_posts(page)
return {"total_count": total_cout, "results": posts}
@router.get("/posts/{post_id}", response_model=PostDetailsModel)
async def get_post(post_id: int):
return await post_utils.get_post(post_id)
@router.put("/posts/{post_id}", response_model=PostDetailsModel)
async def update_post(
post_id: int, post_data: PostModel, current_user=Depends(get_current_user)
):
post = await post_utils.get_post(post_id)
if post["user_id"] != current_user["id"]:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to modify this post",
)
await post_utils.update_post(post_id=post_id, post=post_data)
return await post_utils.get_post(post_id)
Vamos conectar novas rotas adicionando a main.py
from app.routers import posts
app.include_router(posts.router)
Testando
Vamos escrever testes em pytest :
$ pip install pytest
Para testar terminais, FastAPI fornece uma ferramenta especial TestClient .
Vamos escrever um teste de endpoint que não exija uma conexão de banco de dados:
from app.main import app
from fastapi.testclient import TestClient
client = TestClient(app)
def test_health_check():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"Hello": "World"}
Como você pode ver, tudo é muito simples. Você precisa inicializar o TestClient e usá-lo para testar as solicitações HTTP.
Para testar o restante dos terminais, você precisa criar um banco de dados de teste. Vamos editar o arquivo main.py , adicionando a configuração da base de teste a ele:
from os import environ
import databases
DB_USER = environ.get("DB_USER", "user")
DB_PASSWORD = environ.get("DB_PASSWORD", "password")
DB_HOST = environ.get("DB_HOST", "localhost")
TESTING = environ.get("TESTING")
if TESTING:
#
DB_NAME = "async-blogs-temp-for-test"
TEST_SQLALCHEMY_DATABASE_URL = (
f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:5432/{DB_NAME}"
)
database = databases.Database(TEST_SQLALCHEMY_DATABASE_URL)
else:
DB_NAME = "async-blogs"
SQLALCHEMY_DATABASE_URL = (
f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:5432/{DB_NAME}"
)
database = databases.Database(SQLALCHEMY_DATABASE_URL)
Ainda estamos usando o banco de dados "blogs assíncronos" para nosso aplicativo. Mas se o valor da variável de ambiente TESTING for definido, o banco de dados "async-blogs-temp-for-test" será usado .
Para criar o banco de dados "async-blogs-temp-for-test" automaticamente ao executar os testes e excluí-los após executá-los, crie um fixture no arquivo tests / conftest.py :
import os
import pytest
# `os.environ`,
os.environ['TESTING'] = 'True'
from alembic import command
from alembic.config import Config
from app.models import database
from sqlalchemy_utils import create_database, drop_database
@pytest.fixture(scope="module")
def temp_db():
create_database(database.TEST_SQLALCHEMY_DATABASE_URL) #
base_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
alembic_cfg = Config(os.path.join(base_dir, "alembic.ini")) # alembic
command.upgrade(alembic_cfg, "head") #
try:
yield database.TEST_SQLALCHEMY_DATABASE_URL
finally:
drop_database(database.TEST_SQLALCHEMY_DATABASE_URL) #
Para criar e deletar o banco de dados, usaremos a biblioteca sqlalchemy_utils .
Usando a fixação temp_db em nossos testes, podemos testar todos os endpoints de nosso aplicativo:
def test_sign_up(temp_db):
request_data = {
"email": "vader@deathstar.com",
"name": "Darth Vader",
"password": "rainbow"
}
with TestClient(app) as client:
response = client.post("/sign-up", json=request_data)
assert response.status_code == 200
assert response.json()["id"] == 1
assert response.json()["email"] == "vader@deathstar.com"
assert response.json()["name"] == "Darth"
assert response.json()["token"]["expires"] is not None
assert response.json()["token"]["access_token"] is not None
tests / test_posts.py
import asyncio
from app.main import app
from app.schemas.users import UserCreate
from app.utils.users import create_user, create_user_token
from fastapi.testclient import TestClient
def test_create_post(temp_db):
user = UserCreate(
email="vader@deathstar.com",
name="Darth",
password="rainbow"
)
request_data = {
"title": "42",
"content": "Don't panic!"
}
with TestClient(app) as client:
# Create user and use his token to add new post
loop = asyncio.get_event_loop()
user_db = loop.run_until_complete(create_user(user))
response = client.post(
"/posts",
json=request_data,
headers={"Authorization": f"Bearer {user_db['token']['token']}"}
)
assert response.status_code == 201
assert response.json()["id"] == 1
assert response.json()["title"] == "42"
assert response.json()["content"] == "Don't panic!"
def test_create_post_forbidden_without_token(temp_db):
request_data = {
"title": "42",
"content": "Don't panic!"
}
with TestClient(app) as client:
response = client.post("/posts", json=request_data)
assert response.status_code == 401
def test_posts_list(temp_db):
with TestClient(app) as client:
response = client.get("/posts")
assert response.status_code == 200
assert response.json()["total_count"] == 1
assert response.json()["results"][0]["id"] == 1
assert response.json()["results"][0]["title"] == "42"
assert response.json()["results"][0]["content"] == "Don't panic!"
def test_post_detail(temp_db):
post_id = 1
with TestClient(app) as client:
response = client.get(f"/posts/{post_id}")
assert response.status_code == 200
assert response.json()["id"] == 1
assert response.json()["title"] == "42"
assert response.json()["content"] == "Don't panic!"
def test_update_post(temp_db):
post_id = 1
request_data = {
"title": "42",
"content": "Life? Don't talk to me about life."
}
with TestClient(app) as client:
# Create user token to add new post
loop = asyncio.get_event_loop()
token = loop.run_until_complete(create_user_token(user_id=1))
response = client.put(
f"/posts/{post_id}",
json=request_data,
headers={"Authorization": f"Bearer {token['token']}"}
)
assert response.status_code == 200
assert response.json()["id"] == 1
assert response.json()["title"] == "42"
assert response.json()["content"] == "Life? Don't talk to me about life."
def test_update_post_forbidden_without_token(temp_db):
post_id = 1
request_data = {
"title": "42",
"content": "Life? Don't talk to me about life."
}
with TestClient(app) as client:
response = client.put(f"/posts/{post_id}", json=request_data)
assert response.status_code == 401
tests / test_users.py
import asyncio
import pytest
from app.main import app
from app.schemas.users import UserCreate
from app.utils.users import create_user, create_user_token
from fastapi.testclient import TestClient
def test_sign_up(temp_db):
request_data = {
"email": "vader@deathstar.com",
"name": "Darth",
"password": "rainbow"
}
with TestClient(app) as client:
response = client.post("/sign-up", json=request_data)
assert response.status_code == 200
assert response.json()["id"] == 1
assert response.json()["email"] == "vader@deathstar.com"
assert response.json()["name"] == "Darth"
assert response.json()["token"]["expires"] is not None
assert response.json()["token"]["token"] is not None
def test_login(temp_db):
request_data = {"username": "vader@deathstar.com", "password": "rainbow"}
with TestClient(app) as client:
response = client.post("/auth", data=request_data)
assert response.status_code == 200
assert response.json()["token_type"] == "bearer"
assert response.json()["expires"] is not None
assert response.json()["access_token"] is not None
def test_login_with_invalid_password(temp_db):
request_data = {"username": "vader@deathstar.com", "password": "unicorn"}
with TestClient(app) as client:
response = client.post("/auth", data=request_data)
assert response.status_code == 400
assert response.json()["detail"] == "Incorrect email or password"
def test_user_detail(temp_db):
with TestClient(app) as client:
# Create user token to see user info
loop = asyncio.get_event_loop()
token = loop.run_until_complete(create_user_token(user_id=1))
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token['token']}"}
)
assert response.status_code == 200
assert response.json()["id"] == 1
assert response.json()["email"] == "vader@deathstar.com"
assert response.json()["name"] == "Darth"
def test_user_detail_forbidden_without_token(temp_db):
with TestClient(app) as client:
response = client.get("/users/me")
assert response.status_code == 401
@pytest.mark.freeze_time("2015-10-21")
def test_user_detail_forbidden_with_expired_token(temp_db, freezer):
user = UserCreate(
email="sidious@deathstar.com",
name="Palpatine",
password="unicorn"
)
with TestClient(app) as client:
# Create user and use expired token
loop = asyncio.get_event_loop()
user_db = loop.run_until_complete(create_user(user))
freezer.move_to("'2015-11-10'")
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {user_db['token']['token']}"}
)
assert response.status_code == 401
Fontes PS
Isso é tudo, o repositório de origem da postagem pode ser visualizado no GitHub .