Aplicativos de alto desempenho com Python – um tutorial FastAPI

Publicados: 2022-03-11

Boas estruturas de linguagem de programação facilitam a produção de produtos de qualidade mais rapidamente. Ótimos frameworks tornam toda a experiência de desenvolvimento agradável. FastAPI é um novo framework web Python que é poderoso e agradável de usar. Os seguintes recursos fazem com que o FastAPI valha a pena tentar:

  • Velocidade: FastAPI é um dos frameworks web Python mais rápidos. Na verdade, sua velocidade está no mesmo nível do Node.js e do Go. Verifique estes testes de desempenho.
  • Documentos de desenvolvedor detalhados e fáceis de usar
  • Digite dica seu código e obtenha validação e conversão de dados gratuita.
  • Crie plugins facilmente usando injeção de dependência.

Construindo um aplicativo TODO

Para explorar as grandes ideias por trás do FastAPI, vamos construir um aplicativo TODO, que configura listas de tarefas para seus usuários. Nosso pequeno aplicativo fornecerá os seguintes recursos:

  • Cadastro e Login
  • Adicionar novo item TODO
  • Obter uma lista de todos os TODOs
  • Excluir/Atualizar um item TODO

SQLAlchemy para modelos de dados

Nosso aplicativo possui apenas dois modelos: User e TODO. Com a ajuda do SQLAlchemy, o kit de ferramentas de banco de dados para Python, podemos expressar nossos modelos assim:

 class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) lname = Column(String) fname = Column(String) email = Column(String, unique=True, index=True) todos = relationship("TODO", back_populates="owner", cascade="all, delete-orphan") class TODO(Base): __tablename__ = "todos" id = Column(Integer, primary_key=True, index=True) text = Column(String, index=True) completed = Column(Boolean, default=False) owner_id = Column(Integer, ForeignKey("users.id")) owner = relationship("User", back_populates="todos")

Assim que nossos modelos estiverem prontos, vamos escrever o arquivo de configuração do SQLAlchemy para que ele saiba como estabelecer uma conexão com o banco de dados.

 import os from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL = os.environ['SQLALCHEMY_DATABASE_URL'] engine = create_engine( SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

Liberte o poder das dicas de tipo

Uma parte considerável de qualquer projeto de API diz respeito às coisas de rotina, como validação e conversão de dados. Vamos resolver isso de antemão antes de escrevermos os manipuladores de solicitação. Com FastAPI, expressamos o esquema de nossos dados de entrada/saída usando modelos pydantic e, em seguida, usamos esses modelos pydantic para digitar dica e aproveitar a validação e conversão de dados gratuitas. Observe que esses modelos não estão relacionados ao nosso fluxo de trabalho de banco de dados e especificam apenas a forma dos dados que estão entrando e saindo de nossa interface REST. Para escrever modelos pydantic, pense em todas as maneiras pelas quais as informações do usuário e do TODO fluirão para dentro e para fora.

Tradicionalmente, um novo usuário se inscreve em nosso serviço TODO e um usuário existente faz login. Ambas as interações lidam com informações do usuário, mas a forma dos dados será diferente. Precisamos de mais informações dos usuários durante a inscrição e mínimas (somente e-mail e senha) ao efetuar login. Isso significa que precisamos de dois modelos pydantic para expressar essas duas formas diferentes de informações do usuário.

Em nosso aplicativo TODO, no entanto, aproveitaremos o suporte OAuth2 integrado no FastAPI para um fluxo de login baseado em JSON Web Tokens (JWT). Só precisamos definir um esquema UserCreate aqui para especificar os dados que fluirão para nosso endpoint de inscrição e um esquema UserBase para retornar como resposta caso o processo de inscrição seja bem-sucedido.

 from pydantic import BaseModel from pydantic import EmailStr class UserBase(BaseModel): email: EmailStr class UserCreate(UserBase): lname: str fname: str password: str

Aqui, marcamos o sobrenome, o primeiro nome e a senha como uma string, mas isso pode ser ainda mais restrito usando strings restritas pydantic que permitem verificações como comprimento mínimo, comprimento máximo e regexes.

Para dar suporte à criação e listagem de itens TODO, definimos o seguinte esquema:

 class TODOCreate(BaseModel): text: str completed: bool

Para suportar a atualização de um item TODO existente, definimos outro esquema:

 class TODOUpdate(TODOCreate): id: int

Com isso, terminamos com a definição de esquemas para todas as trocas de dados. Agora voltamos nossa atenção para os manipuladores de solicitações, onde esses esquemas serão usados ​​para fazer todo o trabalho pesado de conversão e validação de dados gratuitamente.

Permitir que os usuários se inscrevam

Primeiro, vamos permitir que os usuários se cadastrem, pois todos os nossos serviços precisam ser acessados ​​por um usuário autenticado. Escrevemos nosso primeiro manipulador de solicitação usando o esquema UserCreate e UserBase definido acima.

 @app.post("/api/users", response_model=schemas.User) def signup(user_data: schemas.UserCreate, db: Session = Depends(get_db)): """add new user""" user = crud.get_user_by_email(db, user_data.email) if user: raise HTTPException(status_code=409, detail="Email already registered.") signedup_user = crud.create_user(db, user_data) return signedup_user

Há muita coisa acontecendo neste pequeno pedaço de código. Usamos um decorador para especificar o verbo HTTP, o URI e o esquema de respostas bem-sucedidas. Para garantir que o usuário enviou os dados corretos, digitamos dica no corpo da solicitação com um esquema UserCreate definido anteriormente. O método define outro parâmetro para obter um identificador no banco de dados — isso é injeção de dependência em ação e será discutido posteriormente neste tutorial.

Protegendo nossa API

Queremos os seguintes recursos de segurança em nosso aplicativo:

  • Hash de senha
  • Autenticação baseada em JWT

Para hash de senha, podemos usar Passlib. Vamos definir funções que tratam do hash de senha e verificam se uma senha está correta.

 from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def authenticate_user(db, email: str, password: str): user = crud.get_user_by_email(db, email) if not user: return False if not verify_password(password, user.hashed_password): return False return user

Para habilitar a autenticação baseada em JWT, precisamos gerar JWTs e decodificá-los para obter as credenciais do usuário. Definimos as seguintes funções para fornecer essa funcionalidade.

 # install PyJWT import jwt from fastapi.security import OAuth2PasswordBearer SECRET_KEY = os.environ['SECRET_KEY'] ALGORITHM = os.environ['ALGORITHM'] def create_access_token(*, data: dict, expires_delta: timedelta = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def decode_access_token(db, token): credentials_exception = HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) email: str = payload.get("sub") if email is None: raise credentials_exception token_data = schemas.TokenData(email=email) except PyJWTError: raise credentials_exception user = crud.get_user_by_email(db, email=token_data.email) if user is None: raise credentials_exception return user

Tokens de emissão no login bem-sucedido

Agora, definiremos um endpoint de login e implementaremos o fluxo de senha OAuth2. Este endpoint receberá um e-mail e uma senha. Verificaremos as credenciais no banco de dados e, em caso de sucesso, emitiremos um token da Web JSON para o usuário.

Para receber as credenciais, faremos uso do OAuth2PasswordRequestForm , que faz parte dos utilitários de segurança do FastAPI.

 @app.post("/api/token", response_model=schemas.Token) def login_for_access_token(db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()): """generate access token for valid credentials""" user = authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Incorrect email or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token(data={"sub": user.email}, expires_delta=access_token_expires) return {"access_token": access_token, "token_type": "bearer"}

Usando injeção de dependência para acessar o banco de dados e proteger endpoints

Configuramos o endpoint de login que fornece um JWT a um usuário após o login bem-sucedido. O usuário pode salvar esse token no armazenamento local e mostrá-lo ao nosso back-end como um cabeçalho de autorização. Os endpoints que esperam acesso apenas de usuários logados podem decodificar o token e descobrir quem é o solicitante. Esse tipo de trabalho não está vinculado a um endpoint específico, mas à lógica compartilhada utilizada em todos os endpoints protegidos. É melhor configurar a lógica de decodificação de token como uma dependência que pode ser usada em qualquer manipulador de solicitação.

Na linguagem FastAPI, nossas funções de operação de caminho (manipuladores de solicitação) dependeriam de get_current_user . A dependência get_current_user precisa ter uma conexão com o banco de dados e se conectar à lógica OAuth2PasswordBearer da OAuth2PasswordBearer para obter um token. Resolveremos esse problema fazendo com que get_current_user dependa de outras funções. Dessa forma, podemos definir cadeias de dependência, que é um conceito muito poderoso.

 def get_db(): """provide db session to path operation functions""" try: db = SessionLocal() yield db finally: db.close() def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)): return decode_access_token(db, token) @app.get("/api/me", response_model=schemas.User) def read_logged_in_user(current_user: models.User = Depends(get_current_user)): """return user settings for current user""" return current_user

Usuários logados podem CRUD TODOs

Antes de escrevermos as funções de operação de caminho para TODO Create, Read, Update, Delete (CRUD), definimos as seguintes funções auxiliares para executar o CRUD real no banco de dados.

 def create_todo(db: Session, current_user: models.User, todo_data: schemas.TODOCreate): todo = models.TODO(text=todo_data.text, completed=todo_data.completed) todo.owner = current_user db.add(todo) db.commit() db.refresh(todo) return todo def update_todo(db: Session, todo_data: schemas.TODOUpdate): todo = db.query(models.TODO).filter(models.TODO.id == id).first() todo.text = todo_data.text todo.completed = todo.completed db.commit() db.refresh(todo) return todo def delete_todo(db: Session, id: int): todo = db.query(models.TODO).filter(models.TODO.id == id).first() db.delete(todo) db.commit() def get_user_todos(db: Session, userid: int): return db.query(models.TODO).filter(models.TODO.owner_id == userid).all()

Essas funções de nível de banco de dados serão usadas nos seguintes endpoints REST:

 @app.get("/api/mytodos", response_model=List[schemas.TODO]) def get_own_todos(current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """return a list of TODOs owned by current user""" todos = crud.get_user_todos(db, current_user.id) return todos @app.post("/api/todos", response_model=schemas.TODO) def add_a_todo(todo_data: schemas.TODOCreate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """add a TODO""" todo = crud.create_meal(db, current_user, meal_data) return todo @app.put("/api/todos/{todo_id}", response_model=schemas.TODO) def update_a_todo(todo_id: int, todo_data: schemas.TODOUpdate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """update and return TODO for given id""" todo = crud.get_todo(db, todo_id) updated_todo = crud.update_todo(db, todo_id, todo_data) return updated_todo @app.delete("/api/todos/{todo_id}") def delete_a_meal(todo_id: int, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """delete TODO of given id""" crud.delete_meal(db, todo_id) return {"detail": "TODO Deleted"}

Escrever testes

Vamos escrever alguns testes para nossa API TODO. FastAPI fornece uma classe TestClient baseada na popular biblioteca Requests, e podemos executar os testes com Pytest.

Para garantir que apenas usuários logados possam criar um TODO, podemos escrever algo assim:

 from starlette.testclient import TestClient from .main import app client = TestClient(app) def test_unauthenticated_user_cant_create_todos(): todo=dict(text="run a mile", completed=False) response = client.post("/api/todos", data=todo) assert response.status_code == 401

O teste a seguir verifica nosso endpoint de login e gera um JWT se apresentado com credenciais de login válidas.

 def test_user_can_obtain_auth_token(): response = client.post("/api/token", data=good_credentials) assert response.status_code == 200 assert 'access_token' in response.json() assert 'token_type' in response.json()

Resumindo

Concluímos a implementação de um aplicativo TODO muito simples usando FastAPI. Até agora, você viu o poder das dicas de tipo ser bem usado na definição da forma dos dados de entrada e saída por meio de nossa interface REST. Definimos os esquemas em um só lugar e deixamos para o FastAPI aplicar a validação e conversão de dados. O outro recurso digno de nota é a injeção de dependência. Usamos esse conceito para empacotar a lógica compartilhada de obter uma conexão de banco de dados, decodificar o JWT para obter o usuário conectado no momento e implementar OAuth2 simples com senha e portador. Também vimos como as dependências podem ser encadeadas.

Podemos aplicar facilmente esse conceito para adicionar recursos como acesso baseado em função. Além disso, estamos escrevendo um código conciso e poderoso sem aprender as peculiaridades de um framework. Em palavras simples, FastAPI é uma coleção de ferramentas poderosas que você não precisa aprender porque são apenas Python moderno. Divirta-se.