Machine Learning Usando FastAPI

Uma API assíncrona em FastAPI para disponibilizar um Modelo de Machine Learning

A ideia original desse post é criar um série de post com um material em português que ensine como criar uma API assíncrona, escalável e robusta utilizando o FastAPI, mas não é qualquer API, iriei demonstrar como utilizei recursos avançados do FastAPI e outras bibliotecas para tornar um projeto robusto com pinta profissional. Eu utilizei como exemplo uma API onde é possível consumir um modelo de Machine Learning que prevê se texto é spam ou não.

Nesse post não vou entrar em detalhes como criar, ativar um ambiente virtual em Python, ou mesmo como instalar as dependências necessárias para o projeto para deixá-lo um pouco mais curto. Caso tenham alguma dúvida em relação a esses pontos e boas práticas entrem em contato que posso pensar em algo diferente para os próximos posts.

O projeto pode ser clonado diretamente do meu repositório.

O que diferencia meu projeto

Algumas características dos projeto:

• rotas (endpoints) de acesso para cadastro e autenticação do usuário.
• somente com o usuário autenticado é possível utilizar a rota para acesso ao modelo de machine learning.
• rotas documentas automaticamente e interativas
• middleware que computa o tempo entre a entrada da request e saída da response.
• logs configurados para facilitar o debug da API mesmo em ambiente de produção
• estrutura reutilizável, utilizando orientação a objetos.
• arquivo de configuração de ambiente utilizando recursos do pydantic settings.
• a API e a conexão com o banco de dados possuem recursos assíncronos, utilizando a versão 1.4 do SQLAlchemy.
• as tabelas e suas respectivas atualizações podem ser criadas diretamente com o Alembic já utilizando configurações e updates assíncronos.
• utilizando o docker-compose é possível carregar a API e o banco de dados em containers locais.
• na API existem alguns endpoints para monitoramento, health-check, inclusive com validação da conexão com o banco de dados.
• cada usuário possui 10 utilizações disponível por mês.

Mãos a massa!!!!

O que é o FastAPI

FastAPI é um framework em Python que possui diversas ferramentas prontas para permitir desenvolvedores a utilizar uma interface REST para chamar funções utilizadas na criação de aplicativos. Neste exemplo, o autor usa FastAPI para criar contas, fazer login, autenticar e disponibilizar um modelo de machine learning.

Configurações iniciais

Após instalar as dependências será necessário criar um arquivo com as variáveis de ambiente. Um arquivo .env . Já deixei pronta a lista de variáveis necessárias, basta copiá-las e escrever os valores desejados na frente de cada uma. Mas nesse post, deixarei um pouco mais de detalhes.

APP_ENV=local # a ideia aqui é permitir que o código possa ser enviado para um ambiente de produção
PROJECT_NAME="Spam Predictor with FastAPI" # dê o nome para o seu projeto
POSTGRES_USER=postgres_user # exemplo usuário para conexão no banco de dados.
POSTGRES_PASSWORD=example # exemplo de senha para acesso ao banco de dados
POSTGRES_SERVER=localhost # exemplo de endereço para conexão ao banco dados
POSTGRES_PORT=5432 # porta padrão de acesso ao postgres
POSTGRES_DB=spam_ml # exemplo de database referente ao seu projeto
JWT_SECRET= # veja instruções como gerar seu secret
LOCAL_PORT=8000 # porta onde a API será disponibilizada
FIRST_SUPERUSER=first_user # exemplo de nome do primeiro usuário do banco de dados
FIRST_SUPERUSER_PASSWORD=123456 # exemplo de senha para acesso ao primeiro usuário
FIRST_SUPERUSER_EMAIL=first_user@yopmail.com # # exemplo de email para o primeiro usuário
LOG_LEVEL=DEBUG # nível de log para desenvolvimento e debug em ambiente local
WORKERS_PER_CORE=1 # configurações do uvicorn
MAX_WORKERS=1 # configurações do uvicorn

Para gerar um Token JWT out JWT_SECRET, execute na linha de comando:

openssl rand -hex 32

Será gerado algo como: b59cbee90cd294bf5e1b66fcd8a57fe8ce6999c2e0fa88304ff8c87766329937

Mais uma vez lembrando que esse é um arquivo com as variáveis de ambientes, isso permite configurações diferentes para diversos ambientes. Por exemplo, em um ambiente de produção é possível apontar para um banco de dados na nuvem ou mesmo utilizar uma porta local diferente do exemplo. Ajustar o nível dos logs e etc….

Abusando do PydanticSettings

Já que estamos falando das variáveis de ambiente vale comentar sobre o arquivo app/settings.py esse arquivo herda do pydantic a classe BaseSettings. Esse herança permite especificar o arquivo de onde virá os valores para cada configuração necessária. Dentro da classe Settings observe que há outra classe Config com env_file = '.env' . Caso vc deseje alterar o nome do seu arquivo de de variáveis para, por exemplo, local.env basta alterar o valor que a variável env_file recebe para 'local.env'. Dentro desse mesmo arquivo ainda é gerado o link completo para conexão com o banco de dados. A variável SQLALCHEMY_DATABASE_URI recebe um valor composto pela função PostgresDsn.build(), onde podemos definir diferentes parâmetros como esquema de conexão com o banco da dados, configurações de acesso e tamanho do cache.

O arquivo principal app.py

Uma vez configurado o ambiente é válido olhar para o arquivo app/app.py onde a mágica acontece, pois é o arquivo principal da nossa API.

Inicialmente as variáveis do arquivo settings.py são carregadas. Observe que nesse caso fiz questão de colocar o decorator @lru_cache(), pois como tratam de variáveis de ambiente, dificilmente serão modificadas e dessa forma o python se encarrega de buscar os valores diretamente do cache / memória e não mais do arquivo de texto criado anteriormente.

@lru_cache()
def get_settings():
return Settings()

settings = get_settings()

O próximo passo é instanciar a classe FastAPI e definir a rota de acesso a documentação. aqui também vale alterar da forma como for necessária.

app = FastAPI(
title=settings.PROJECT_NAME,
docs_url=f"/docs/",
version="1.0.0",
)

No próximo trecho de código são carregados os middlewares e as rotas de acesso. Aqui uma preferência do autor, por carregá-las nos respectivos arquivos __init__.py e passá-las como uma lista ao arquivo principal, deixando o código mais limpo.

if len(middlewares_list) > 0:
for middleware in middlewares_list:
app.add_middleware(middleware)

if len(endpoints_list) > 0:
for endpoint in endpoints_list:
app.include_router(endpoint)

O penúltimo trecho desse arquivos é onde definimos como a API irá lidar com os erros durante uma requisição, deixei genérico, mas também poderia ser customizado de acordo com a necessidade de cada negócio.

@app.exception_handler(HTTPError)
async def http_error_handler(request: Request, exception: HTTPError):
if exception.msg:
    return JSONResponse(
        status_code=exception.code,
        content={exception.**repr**},
    )
else:
    str_tb = format_exc()
    msg = ErrorMessage(
        traceback=str_tb,
        title="Some error occur",
        code=exception.code
        )
    return JSONResponse(status_code=exception.code, content=msg.dict())

Note que o decorator herda a classe HTTPError que poderia ser diretamente do FastAPI / Pydantic ou como no meu caso de urllib.error

O último trecho é um artifício para “startar” a API diretamente pelo arquivo, mas conforme foram sendo adicionados recursos acabou não sendo conveniente dessa forma.

if **name** == '**main**':
from uvicorn import run

    run(app,  # type: ignore
        port=settings.PORT,
        host=settings.HOST,
        use_colors=True
        )

Esses parâmetros podem ser passados diretamente para a IDE em que você está desenvolvendo para permitir uma rápida iniciação do projeto.

O Banco de Dados

Conforme destacado anteriormente fiz questão de utilizar o SQLAlchemy na versão 1.4 para permitir o acesso ao banco de dados de forma assíncrona. Para essas configurações vamos olhar dentro da pasta app/database.

O arquivo app/database/session.py é responsável pela conexão e disponibilizar a sessão assíncrona com o Banco de Dados.

A engine do banco é criada na função:

engine = create_async_engine(
SQLALCHEMY_DATABASE_URI,
pool_pre_ping=True,
connect_args={"server_settings": {"jit": "off"}}
)

Vale destacar que tive alguns problemas de conexão com o banco de dados Postgres de forma assíncrona e ao buscar uma solução na internet me deparei com esses connect_args={"server_settings": {"jit": "off"}}) . Foi a forma que eu encontrei para contornar a issue de conexão do esquema Asyncpg e uma coluna do tipo Enum.

Depois criado a engine do banco dados, vêm os comandos para disponibilizar a sessão:

session*local = sessionmaker(bind=engine, class*=AsyncSession, expire_on_commit=False, autoflush=True)
ScopedSession = scoped_session(session_local)

async def get_session() -> AsyncGenerator:
async with ScopedSession() as session:
yield session

Ainda dentro da pasta app/database vale uma olhada rápida no arquivo base_class.py. Ele será responsável por auxiliar o SQLAlchemy e o Alembic na criação das tabelas no nosso banco de dados.

@as_declarative()
class Base:
id: int = Column(Integer, primary_key=True, autoincrement=True)
**name**: str

    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls) -> str:
        return cls.__name__.lower()

A representação em python dos modelos das tabelas do banco dados estão na pasta app/models. Essa representação se faz necessário para que o SQLAlchemy possa “traduzir e conversar” com o “dialeto” do Postgres.

Aqui também houve uma preferência do autor em mesclar o modelo do banco de dados com os schemas de cada endpoint. Em diversas literaturas, inclusive na documentação oficial você irá ver o schemas em uma pasta a parte no projeto. Mas preferi deixá-los em um mesmo arquivo, pois esse projeto não possui um alto nível de complexidade. Mais uma vez preferência minha.

No arquivo app/models/usermodel.py é possível ver onde eu fiz a presentação da tabela usermodel do banco de dados. Note que a classe UserModel herda da classe Base, lá da pasta app/database. Aqui me fiz valer de mais algumas facilidades do Pydantic como validação dos campos e deixar como exemplo para implementações futuras. Por exemplo, a validação do atributo fullname ou nome completo deve ter pelo menos Nome e Sobrenome. A ideia aqui é que seja passada uma string com pelo menos um espaço no meio. E a validação do documento informado no momento do cadastro, que em ambientes locais está desativado, mas alterado o valor da variável APP_ENV para “dev” por exemplo passa a ser necessário informar um CPF ou CNPJ que seja válido.

Um pouco mais abaixo é possível observar os schemas utilizados para validação das requisições e das respostas. Aqui é onde o Pydantic gera magicamente a documentação do OpenAPI.

python code:

class BaseUser(BaseModel):
email: EmailStr = Field(..., description="User email for contact")
full_name: FullNameField = Field(min_length=3)
document_number: str = Field(..., description="User document number")

    @validator("full_name")
    def validate_full_name(cls, v):
        try:
            first_name, last_name = v.split()
            return v
        except Exception:
            raise HTTPException(status.HTTP_404_NOT_FOUND,
                                detail="You should provide at least 2 names")

    @validator("document_number")
    def validate_document_number(cls, v):
        try:
            logger.info(f"Validating document number {v}")
            return parse_doc_number(v)
        except Exception:
            raise HTTPException(status.HTTP_404_NOT_FOUND,
                                detail="You should provide a valid document number")

class UserSignIn(BaseUser):
password: str = Field(min_length=6, description="User Password")
username: str = Field(..., min_length=3, description="Username to login") # role: Optional[str]
phone: Optional[str]

    class Config:
        orm_mode = True

class UserSignOut(BaseUser):
id: int
created_at: datetime # role: UserRole
phone: Optional[str]
username: str

Aqui deixo registrado meus agradecimentos Pydantic!

Repare que os endpoints e os métodos possíveis de utilizam já foram carregados, caos abra cada um deles, a forma de utilizá-los estará preenchida. Note também que os endpoints /users/whoami e /predict_sentiment possuem um cadeado do lado direto, indicando que é necessária a autenticação. A autenticação deve ser feita no endpoint /login

Já parou para pensar como a documentação de uma API é importante? Podemos imaginar uma API como uma tomada. Não precisamos nos preocupar como a energia elétrica foi gerada ou como ele chegou até a tomada, mas temos certeza que ao ligar um dispositivo na tomada ele irá funcionar. Por isso é importante termos muito bem registrado como utilizar cada endpoint.

Alembic - Uma ferramenta extremamente versátil para migração do banco de dados.

Como já falamos sobre o endereço de conexão com o banco dados obtido do arquivo settings, já falamos sobre a sessão assíncrona com o banco e comentamos sobre o modelos, estamos aptos a ver um pouco sobre o Alembic.

execute na linha de comando:

alembic init -t async app/database/migrations

Esse comando é responsável por gerar os scripts assíncronos do alembic. Será criado um arquivo na raiz do projeto alembic.ini e um pasta migrations dentro do caminho informado app/database/migrations

Dentro dessa pasta será necessário editar o arquivo env.py para:

# this is the Alembic Config object, which provides

# access to the values within the .ini file in use.

from app.database.base_class import Base
from app.database.session import SQLALCHEMY_DATABASE_URI

config = context.config
config.set_main_option("sqlalchemy.url", SQLALCHEMY_DATABASE_URI)

# Interpret the config file for Python logging.

# This line sets up loggers basically.

if config.config_file_name is not None:
fileConfig(config.config_file_name)

# add your model's MetaData object here

# for 'autogenerate' support

# from myapp import mymodel

# target_metadata = mymodel.Base.metadata

# target_metadata = mymodel.Base.metadata

from app.models.usermodel import UserModel # noqa
from app.models.predictions import PredictionModel # noqa
target_metadata = [Base.metadata]

Feitas as alterações necessárias para que o Alembic consiga identificar como ele fará a conexão com o banco de dados e quais são os modelos que ele deverá criar deve-se executar o comando novamente na linha do terminal:

alembic revision --autogenerate -m "create first migrations"

e

alembic upgrade head

Assim automaticamente as tabelas serão criadas no banco dados. Com os tabelas criadas agora podemos executar pela primeira vez nossa API.

Executando a API

Ao executarmos na linha de comando:

uvicorn --host 127.0.0.1 --port 3333 app.app:app --reload

Com a API funcionando, podemos acessar o endereço no nosso navegador e utilizar a documentação iterativa:

para a acessar a API insira no navegador ou mesmo no terminal:

curl -X 'GET' \
  'http://127.0.0.1:3333/'

observe o número logo após o símbolo de : 3333 deve ser o mesmo inserido no comando acima logo após o --port, pois a API estará servindo nessa porta, caso altere o valor do comando será necessário alterar também a porta no navegador.

a resposta será:

Caso aponte o navegador para http://127.0.0.1:3333/docs para ter acesso a documentação dos endpoints.

clicando no endpoint /health_check é possível utilizar o método GET e realizar uma chamada contra API.

spam_health_check.png

Ao clicar no botão Try it out, outro campo irá se abrir, usando o botão execute irá realizar a solicitação.

No terminal onde a API está sendo executada é possível observar os logs.

Registrando um novo usuário e autenticando

Para registrar um novo usuário utilize o endpoint /users/register/ e preencha com as informações que desejar. Lembre-se dos campos de validação que comentei anteriormente.

Agora para autenticar o usuário é necessário utilizar o endpoint /login/. Insira os dados utilizados no passo anterior: username e password

Se os dados estiverem corretos o retorno será algo como:

O modelo de Machine Learning

Como o foco desse projeto não é o modelo de machine learning em si, mas como disponibilizá-lo para utilização e produção. Eu tomei a liberdade para reaproveitar a ideia desse post e recriar o modelo.

Como já comentando em um post anterior sobre o assunto praticamente 90% dos modelos de machine learning não são disponibilizados.

O modelo foi treinando utilizando os recursos da biblioteca Scikit Learn, pois o modelo classificador multinomial Naive Bayes é adequado para classificação com características discretas (por exemplo, contagem de palavras para classificação de texto). A distribuição multinomial normalmente requer contagens de recursos inteiros. No entanto, na prática, contagens fracionárias como tf-idf também podem funcionar.

def run_model_training(X_train, X_test, y_train, y_test):
    clf = MultinomialNB()
    clf.fit(X_train,y_train)
    clf.score(X_test,y_test)
    y_pred = clf.predict(X_test)
    print(classification_report(y_test, y_pred))

    return clf

Mais detalhes sobre a implementação do modelo pode ser vista diretamente no repositório do projeto, no diretório app -> ml_models.

Após a etapa de treino o modelo é disponibilizado para uso utilizando-se a biblioteca do Python joblib, que disponibiliza um pipeline rápido, com a possibilidade de utilização de cache e computação paralela.

Utilizando o endpoint de classificação de Spam.

Na rota de utilização do modelo de machine learning, "/predict_sentiment" foram implementadas diversas funcionalidades. Inclusive a verificação se o modelo existe e foi treinado. Caso não seja encontrado o modelo, ao realizar o deploy da API o modelo será treinado utilizando como base o arquivo spam.csv localizado no diretório app -> data.

Após o usuário realizar o cadastro e login nos endpoints corretos o endpoint estará liberado para requisição.

Do campo response podemos extrair a mensagem:

{
  "text_message": "Yes... I trust u to buy new stuff ASAP.....",
  "spam_polarity": "Ham",
  "month_used_amount": 1
}

No campo text_message frase que foi enviada para teste. No campo spam_polarity a saída Ham indica que o texto não foi classificado como spam e o campo month_used_amount indica quantas vezes o usuário logado utilizou o serviço.

Conclusões

Parabéns, se você leu o post até esse ponto devem ter surgido algumas dúvidas, interrogações e outros comentários. Mas ao longo desse post foi possível ver sobre a utilização de recursos modernos de python para uma solução elegante de API.

Espero que tenha atendido as expectativas em relação a proposta dos tópicos que seriam abordados e como foram discutidos aqui, caso ainda tenha dúvidas ou comentários utilize os canais de contato para discutirmos esses pontos. Questionamentos geram insights para novos posts.

Mais uma vez, esse e outros projetos podem ser acessados diretamente no meu repositório.

Recapitulando

• Utilizamos recursos do Pydantic para configurar a aplicação e gerar a documentação de cada endpoint;
• Alembic de forma assíncrona;
• Endpoint para criação e autenticação de usuários;
• Refresh tokens;
• JWT payload e JSON Web Encryption;

Próximos Post já previstos.

• Solução utilizada do docker-compose para subir a API e o banco de dados;
• Teste automatizados;
• middleware que computa o tempo entre a entrada da request e saída da response;
• logs configurados para facilitar o debug da API mesmo em ambiente de produção