DEV Community: Wesley de Morais

RAG na prática: transformando PDFs em respostas inteligentes com LLMs

Wesley de Morais — Fri, 13 Jun 2025 19:50:55 +0000

Sumário

Entendendo o que são Embeddings
Entendendo o que são Vector Store
Fluxo de uma aplicação normal com RAG
Fluxo da nossa aplicação usando RAG
Criando nossa aplicação

Entendendo o que são Embeddings

Modelos de inteligência artificial não compreendem diretamente a linguagem natural como os humanos. Em vez disso, eles trabalham melhor com representações matemáticas. É aí que entram os embeddings — vetores gerados por modelos de embedding que capturam a essência de um texto, imagem ou áudio.

Esses vetores tornam possível comparar conteúdos de forma rápida e precisa, permitindo que sistemas de IA reconheçam similaridades mesmo quando a linguagem usada não é exatamente a mesma.

Entendendo o que são Vector Store

Os vector stores são bancos de dados especializados em armazenar esses vetores (embedding vectors).Quando um novo vetor é consultado, o vector store retorna os vetores armazenados mais semelhantes a ele, com base em medidas matemáticas de similaridade.

Esses vetores recuperados podem então ser usados como contexto para um modelo de linguagem (LLM), que gera uma resposta em linguagem natural com base nas informações mais relevantes.

Fluxo de uma aplicação normal com RAG(Retrieval-augmented generation)

Aqui temos um fluxo de uma aplicação básica usando RAG

Um Vector Store é preenchido previamente com dados que foram convertidos em embedding vectors e armazenados para busca eficiente.
Quando um usuário faz uma pergunta, essa pergunta é convertida em um embedding vector correspondente.
Esse vetor da pergunta é enviado ao Vector Store, que utiliza algoritmos de similaridade para recuperar os embedding vectors mais similares.
Os vetores recuperados são mapeados de volta para seus respectivos Documents (textos originais).
Esses Documents relevantes são inseridos em um prompt estruturado, junto com a pergunta do usuário.
O prompt é enviado como contexto para uma LLM (Large Language Model), que utiliza as informações para gerar uma resposta.
A LLM gera o texto (recurso) como resposta ao usuário, combinando a informação do contexto com sua capacidade de linguagem natural.

Fluxo da nossa aplicação usando RAG

No nosso RAG, vamos seguir uma abordagem parecida, mas com uma diferença: em vez de a aplicação apenas ficar esperando perguntas baseadas em informações previamente carregadas, vamos permitir que o usuário envie um arquivo junto com a pergunta

Dessa forma, a aplicação poderá entender o contexto diretamente a partir do conteúdo enviado, respondendo com base naquele documento.

O arquivo em pdf vai ser dividido em partes usando a classe RecursiveCharacterTextSplitter e para cada parte vai ser criada um embedding vector relacionado.

Criando nossa aplicação

Antes de começar nossa aplicação é necessário que crie um arquivo .env contendo essas informações

OPENAI_API_KEY=<your_openai_api_key>
OPENAI_ORGANIZATION_KEY=<your_openai_api_key>

Você pode pegar essas informações acessando o link https://platform.openai.com/docs/overview

Dependências

Com o ambiente virtual criado instale as seguintes dependências

langchain==0.3.25
gradio==5.33.2
langchain-openai==0.3.22
langchain-community==0.3.25
pypdf==5.6.0
python-dotenv==1.1.0

Nossa aplicação vai seguir 2 etapas:

Criação da interface demo usando a biblioteca gradio
Criação da caso de uso para executar nossa aplicação

Criação da interface demo

Em um arquivo chamado app.py podemos adicionar seguinte código:

import gradio as gr

def handle_file(text_input, file):
    return text_input


with gr.Blocks() as demo:
    gr.Markdown("## Verificador de PDFs com RAG")
    text_input = gr.Textbox(label="Entre com sua questão aqui", placeholder="Faça uma pergunta sobre o PDF carregado")
    upload = gr.File(type="filepath")
    output = gr.Textbox(label="Output")
    button = gr.Button("Submit")

    button.click(
        fn=handle_file,
        inputs=[text_input, upload],
        outputs=output
    )
demo.launch(share=True)

O código acima é responsável por construir uma interface onde teremos :

Um campo para o usuário adicionar a sua questão
Um campo para adicionar o arquivo
Um campo que será adicionado a resposta da llm
Um botão de submissão, sendo responsável por quando for feito o evento de clique, então ser chamado a função handle_file.

No final ele expõe essa interface para ser acessado externamente.

Criação da caso de uso para executar nossa aplicação

No mesmo arquivo podemos criar um caso de uso onde inicialmente ele vai pegar o arquivo enviado e carregar na variável self.docs

class HandleLLMUseCase:

    def handle_pdf_load(self):
        """
        Carrega o PDF enviado e extrai os documentos.
        Se nenhum arquivo for enviado, levanta um erro.
        """
        if self.file is None:
            raise ValueError("Nenhum arquivo enviado.")

        loader = PyPDFLoader(self.file)
        self.docs = loader.load()

Com o arquivo carregado, podemos dividir seu conteúdo em pequenos trechos (chunks), o que facilita a criação dos embedding vectors posteriormente.

def handle_text_split(self):
        """
        Divide os documentos carregados em chunks menores de 
        500 caracteres com sobreposição de 50 caracteres.
        Se nenhum documento for carregado, levanta um erro.
        """
        if not hasattr(self, 'docs'):
            raise ValueError("Nenhum documento carregado.")

        text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
        self.chunks = text_splitter.split_documents(self.docs)

Com os chunks de texto já criados, podemos convertê-los em embedding vectors utilizando o modelo OpenAIEmbeddings. Esses vetores serão então armazenados em um Vector Store, que permite realizar buscas por similaridade de forma eficiente. Para fins de simplicidade e praticidade, vamos utilizar o InMemoryVectorStore, uma opção gratuita e fácil de usar.

No entanto, o Langchain oferece suporte a diversos outros vector stores, como FAISS, Chroma, Weaviate, Pinecone e outros — você pode conferir a lista completa na Documentação oficial.

def handle_embeddings(self):
        """
        Cria embeddings dos chunks de texto usando OpenAIEmbeddings .
        Se nenhum chunk de texto estiver disponível, levanta um erro.
        """
        if not hasattr(self, 'chunks'):
            raise ValueError("Nenhum chunk de texto disponível.")

        embeddings = OpenAIEmbeddings(openai_api_key=os.environ.get("OPENAI_API_KEY"))
        vectorstore = InMemoryVectorStore.from_documents(self.chunks, embeddings)
        self.retriever = vectorstore.as_retriever()

Com o Vector Store já criado, podemos agora construir uma cadeia de RAG que combina os documentos recuperados do Vector Store com uma LLM. Essa cadeia será responsável por buscar os trechos mais relevantes no conteúdo armazenado e usá-los como contexto para a LLM gerar uma resposta em linguagem natural.

def handle_chain_creation(self):
        """
        Cria uma cadeia de perguntas e respostas (RetrievalQA) usando o modelo LLM.
        """
        llm = ChatOpenAI(
            api_key=os.environ.get("OPENAI_API_KEY"),
            organization=os.environ.get("OPENAI_ORGANIZATION_KEY"),
            model="gpt-4o-mini",
        )

        self.qa_chain = RetrievalQA.from_chain_type(
            llm=llm,
            retriever=self.retriever
        )

def handle_query(self):
        """
        Executa a consulta na cadeia de perguntas e respostas e retorna o resultado.
        """
        response = self.qa_chain.invoke({"query": self.text_input})
        return response["result"]

No final, o código fica estruturado de forma que é possível fazer perguntas e receber respostas contextualizadas com base no conteúdo do arquivo fornecido.

import gradio as gr
import os
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.vectorstores import InMemoryVectorStore
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()  

class HandleLLMUseCase:

    def __init__(self, text_input, file):
        self.text_input = text_input
        self.file = file
        self.docs = None
        self.chunks = None
        self.retriever = None
        self.qa_chain = None

    def handle_pdf_load(self):
        """
        Carrega o PDF enviado e extrai os documentos.
        Se nenhum arquivo for enviado, levanta um erro.
        """
        if self.file is None:
            raise ValueError("Nenhum arquivo enviado.")

        loader = PyPDFLoader(self.file)
        self.docs = loader.load()

    def handle_text_split(self):
        """
        Divide os documentos carregados em chunks menores de 
        500 caracteres com sobreposição de 50 caracteres.
        Se nenhum documento for carregado, levanta um erro.
        """
        if not hasattr(self, 'docs'):
            raise ValueError("Nenhum documento carregado.")

        text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
        self.chunks = text_splitter.split_documents(self.docs)

    def handle_embeddings(self):
        """
        Cria embeddings dos chunks de texto usando OpenAIEmbeddings .
        Se nenhum chunk de texto estiver disponível, levanta um erro.
        """
        if not hasattr(self, 'chunks'):
            raise ValueError("Nenhum chunk de texto disponível.")

        embeddings = OpenAIEmbeddings(openai_api_key=os.environ.get("OPENAI_API_KEY"))
        vectorstore = InMemoryVectorStore.from_documents(self.chunks, embeddings)
        self.retriever = vectorstore.as_retriever()

    def handle_chain_creation(self):
        """
        Cria uma cadeia de perguntas e respostas (RetrievalQA) usando o modelo LLM.
        """
        llm = ChatOpenAI(
            api_key=os.environ.get("OPENAI_API_KEY"),
            organization=os.environ.get("OPENAI_ORGANIZATION_KEY"),
            model="gpt-4o-mini",
        )

        self.qa_chain = RetrievalQA.from_chain_type(
            llm=llm,
            retriever=self.retriever
        )

    def handle_query(self):
        """
        Executa a consulta na cadeia de perguntas e respostas e retorna o resultado.
        """
        response = self.qa_chain.invoke({"query": self.text_input})
        return response["result"]

    def execute(self):
        self.handle_pdf_load()
        self.handle_text_split()
        self.handle_embeddings()
        self.handle_chain_creation()
        response = self.handle_query()
        return response



def handle_file(text_input, file):
    instance = HandleLLMUseCase(text_input, file)
    return instance.execute()


with gr.Blocks() as demo:
    gr.Markdown("## Verificador de PDFs com RAG")
    text_input = gr.Textbox(label="Entre com sua questão aqui", placeholder="Faça uma pergunta sobre o PDF carregado")
    upload = gr.File(type="filepath")
    output = gr.Textbox(label="Output")
    button = gr.Button("Submit")

    button.click(
        fn=handle_file,
        inputs=[text_input, upload],
        outputs=output
    )
demo.launch(share=True)

Referencias

https://python.langchain.com/docs/introduction/

https://datawaybr.medium.com/guia-definitivo-para-vector-databases-bbeeb8f0d802

https://www.gradio.app

Link do repositório

https://github.com/WesleyVitor/rag_pdf_openai_app

Entendendo Relações Genéricas no framework Django

Wesley de Morais — Sun, 16 Apr 2023 20:47:23 +0000

Sumário

Introdução a relações genéricas
Estudo de caso com django
Criando aplicação sem Relação genérica
Refatorando para ter relação genérica
Referências

Introdução a relações genéricas

Para entender o que são relações genéricas devemos antes entender algumas associações de banco de dados. Quando temos modelos como Post e Curtida podemos ter uma relação muitos para um, pois um Post pode ter muitas Curtidas, e uma Curtida é de um Post.

Caso tivermos um modelo Comentário que também pode ter curtida, então não seria de bacana utilizar o modelo Curtida anterior, assim criaríamos um modelo CurtidaComentário, mas podemos entender que ambos os modelos, Curtida e CurtidaComentario, tem o mesmo objetivo, porém para modelos diferentes.

Assim, podemos usar relação genérica, pois podemos fazer uso do famoso polimorfismo para o modelo Curtida ser tanto usado pelo Post quanto pelo Comentário.

Estudo de caso com Django

Vamos utilizar o nosso problema da sessão anterior para entender na prática, então podemos fazer uso de um diagrama entidade relacionamento para clarear a mente.

Na imagem acima temos, 1 Post tem muitos comentários, 1 comentário pode ter muitas curtidas e 1 Post também pode ter muitas curtidas, porém quando se curte um comentário, este tem relação com um post, só que a instância de curtida de um comentário deve se relacionar com o comentário, mas a curtida de um post não deve se relacionar com um comentário, pois a curtida foi do post, e não do comentário, só que vai ser utilizado o mesmo modelo.

Criando aplicação sem Relação genérica

Dependências

Primeiramente, crie uma ambiente virtual para instalar as dependências da aplicação. Dentro do ambiente virtual instale as seguintes dependências:

mkdir understand_generic_relation 

python -m venv venv

source venv/bin/activate

pip install django

Instalamos o django e django rest framework para fazer a criação da nossa aplicação e dentro da pasta do ambiente virtual digite o comando para criar um projeto e a nossa aplicação:

django-admin startproject understand_generic_relation .

django-admin startapp core

Settings

INSTALLED_APPS = [
    ...
    'django.contrib.staticfiles',
    'core.apps.CoreConfig',
]

Models

from django.db import models
from django.utils import timezone
# Create your models here.

class Post(models.Model):
    text = models.TextField(max_length=400)

class Comment(models.Model):
    post = models.ForeignKey(Post, on_delete=models.DO_NOTHING)
    text = models.TextField(max_length=130)

    def __str__(self) -> str:
        return f"{(self.post.id)}-comment#{self.id}"

class Like(models.Model):
    post = models.ForeignKey(Post, null=True, blank=True, on_delete=models.DO_NOTHING, related_name="likes_post")
    comment = models.ForeignKey(Comment,null=True, blank=True, on_delete=models.DO_NOTHING, related_name="likes_comment")

    created_at = models.DateTimeField(default=timezone.now)

    class Meta:
        # The instance of like is to Post exclusive or Comment
        constraints = [
            models.UniqueConstraint(
                fields=['post'],
                name="unique_like_to_post",
                condition=models.Q(post__isnull=False)
            ),
            models.UniqueConstraint(
                fields=['comment'],
                name="unique_like_to_comment",
                condition=models.Q(comment__isnull=False)
            )
        ]

Vamos criar 3 modelos, sendo um Post, um Comment que tem relação com Post e um Like que tem relação com ou post ou comentário.

Crie a migração e execute-as com o seguinte comando no terminal

python manage.py makemigrations
python manage.py migrate

Shell

Entrando no shell podemos testar a nossa modelagem, entre nele com seguinte comando:

python manage.py shell

Criando uma postagem e seus comentários:

>>> from core.models import *
>>> post = Post(text="Atualização do Etherium vai ser boa?")
>>> post.save()
>>> comment1 = Comment(text="Claro que vai!", post=post)
>>> comment1.save()
>>> comment2 = Comment(text="Já deu bom!", post=post)
>>> comment2.save()

Podemos agora criar instâncias de Like para uma postagem ou um comentário

>>> like_to_post1 = Like(post=post)
>>> like_to_post1.save()
>>> like_to_comment1 = Like(comment=comment1)
>>> like_to_comment1.save()

Quando tentarmos criar algo como “Like(post=post, comment=comment1).save()” receberemos a seguinte mensagem de erro:

django.db.utils.IntegrityError: UNIQUE constraint failed: core_like.post_id

Assim nosso objetivo foi concluído, mas temos o seguinte o problema. Caso necessitarmos no nosso sistema de mais elementos que podem serem curtidos, o nosso modelo teria muitas chaves estrangeiras nulas para cada instância criada, e também aumentaríamos o tamanho dele.

Refatorando para ter relação genérica

Antes de ir para as relações genéricas, vamos entender o modelo ContentType, toda vez que um modelo é criado, é criado uma instância de contenttype para aquele modelo, assim este guarda o identificador do modelo criado. Esse modelo vai nos ajudar a saber se uma curtida é de um modelo Comment ou Post.

Flush

Vamos apagar os dados no banco com o seguinte comando:

python manage.py flush

Models

from django.db import models
from django.utils import timezone
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
from django.contrib.contenttypes.models import ContentType

# Create your models here.

class Post(models.Model):
    text = models.TextField(max_length=400)
    likes = GenericRelation("Like")

class Comment(models.Model):
    post = models.ForeignKey(Post, on_delete=models.DO_NOTHING)
    text = models.TextField(max_length=130)
    likes = GenericRelation("Like")

    def __str__(self) -> str:
        return f"{(self.post.id)}-comment#{self.id}"

class Like(models.Model):
    content_type = models.ForeignKey(ContentType, on_delete=models.DO_NOTHING, default=None, null=True)
    object_id = models.PositiveIntegerField(default=None, null=True)
    content_object = GenericForeignKey('content_type', 'object_id')

    created_at = models.DateTimeField(default=timezone.now)

Criação do atributo content_type vai fazer a relação entre os modelos, pois é uma chave estrangeira de ContentType
Criação do atributo object_id vai guardar o valor da instância criada seja de um comentário ou de uma postagem
Criação do atributo content_object vai guardar a instância em si, dessa forma quando passarmos a instância para este atributo, então ele vai pegar o id do content_type relacionado e atribuir ao nosso atributo, e também vai pegar o id da instância e atribuir ao nosso atributo.
Afim de conseguir pegar todas as instâncias de um post, criamos o atributo likes do tipo GenericRelation no modelo de Post
Afim de conseguir pegar todas as instâncias de um comentário, criamos o atributo likes do tipo GenericRelation no modelo de Comment ### Shell

Criando uma postagem e seus comentários:

>>> from core.models import *
>>> post = Post(text="Atualização do Etherium vai ser boa?")
>>> post.save()
>>> comment1 = Comment(text="Claro que vai!", post=post)
>>> comment1.save()
>>> comment2 = Comment(text="Já deu bom!", post=post)
>>> comment2.save()

Podemos agora criar instâncias de Like para uma postagem ou um comentário

>>> like_to_post1 = Like(content_object=post)
>>> like_to_post1.save()
>>> like_to_comment1 = Like(content_object=comment1)
>>> like_to_comment1.save()

Repositório no Github

Referências

https://leportella.com/pt-br/relacoes-genericas-django/

https://www.linkedin.com/pulse/relações-genéricas-generic-relation-com-django-fabio-carvalho/?originalSubdomain=pt

https://www.djangoproject.com

Entendendo Task Queue com Django, Celery e RabbitMQ

Wesley de Morais — Tue, 21 Feb 2023 13:35:56 +0000

Sumário

Introdução sobre task queue
Entendendo o que é celery e rabbitmq
Estudo de caso com django
Criando a Api com DRF sem Task Queue
Usando task queue na API
Referências

Introdução sobre Task Queue

Quando estamos criando os nossos projetos existem determinadas tarefas que demoram um pouco(Ou muito) para serem processadas, exemplo disso é a geração de relatórios e acesso a determinadas APis que podem estar fora do ar por vários motivos, como acesso a api do governo para checar o CPF de alguém, afim de verificar se o portador tem algum problema legal. Desta forma é imprescindível que o usuário não fique esperando determinadas tarefas serem concluídas para ter um feedback do que aconteceu ou o que poderá acontecer. Portanto, a task queue, ou também chamada de fila de tarefas, é um modo de postergar determinadas tarefas, dessa maneira essas tarefas vão ser feitas de forma assíncronas em relação ao fluxo da aplicação.

Na imagem acima vemos como funciona esse conceito, assim existe um produtor(Producers) que cria as Tasks para serem processadas, essas vão ser armazenadas dentro de uma fila, dentro do message broker, posteriormente essas tasks vão ser consumidas pelos consumidores(Consumers) para serem processadas, esses podem ou não fazer chamadas ao banco de dados, dependendo da tarefa.

Entendendo o que é o Celery e RabbitMQ

O Celery é uma aplicação python que nos possibilita ter uma abstração para construção de uma Task Queue, este tem uma dependência da existência de um message broker para funcionar. Os consumidores do celery são chamados de Workers que são processos no qual sempre ficam fazendo solicitações para o broker por novas tasks para serem processadas. O RabbitMQ é um dos brokers suportados, no qual tem o objetivo de guardar determinadas tasks que serão consumidas pelos workers. Junto com o celery é possível criar múltiplas filas de prioridades, afim de determinadas tarefas serem processadas antes de outras. O protocolo de comunicação entre o produtor e o broker é chamado de AMQP.

Estudo de caso com Django

Para entender melhor como funciona a task queue com o celery vamos para uma situação problema, imagine que temos uma aplicação que faz o gerenciamento de uma loja de roupas, no qual podemos gerenciar nossas vendas, um dos endpoints que temos é para gerar relatórios no formato xlsx(Excel) das nossas vendas. Assim podemos ter o seguinte diagrama entidade relacionamento:

Em resumo no nosso diagrama entidade relacionamento acima temos que um vendedor fazendo muitas vendas e uma venda tendo muitas roupas.

Criando a Api com DRF sem Task Queue

Dependências

Primeiramente, crie uma ambiente virtual para instalar as dependências da aplicação. Dentro do ambiente virtual instale as seguintes dependências:

pip install django djangorestframework model_bakery openpyxl

Instalamos o django e django rest framework para fazer a criação da nossa aplicação; também instalamos o model bakery e openpyxml, a primeira para fazer determinados fakes para a gente fazer o nosso teste, e a segunda para criar o nosso relatório.

Dentro da pasta do ambiente virtual digite o comando para criar um projeto e a nossa aplicação:

django-admin startproject store_app .
django-admin startapp core

Models

Vamos mapear o nosso diagrama em modelos

from django.db import models
from django.utils import timezone
from django.core.exceptions import ValidationError
# Create your models here.

class Seller(models.Model):
    name = models.CharField(max_length=30)

    def __str__(self) -> str:
        return self.name

class Sale(models.Model):
    total = models.FloatField(default=0)
    created_at = models.DateTimeField(default=timezone.now)

    saller = models.ForeignKey(Seller, on_delete=models.DO_NOTHING)

SIZE_CLOTHES = (
    ('p','P'),
    ('m','M'),
    ('g','G')
)

class Clothe(models.Model):
    name = models.CharField(max_length=50)
    size = models.CharField(max_length=1, choices=SIZE_CLOTHES)
    amount = models.PositiveIntegerField()

    def __str__(self) -> str:
        return f"{self.name}({self.size})"

class ClotheItem(models.Model):
    price_unit = models.FloatField()
    amount_items = models.PositiveIntegerField()

    sale = models.ForeignKey(Sale, on_delete=models.CASCADE, related_name="items", null=False)
    clothe = models.ForeignKey(Clothe, on_delete=models.CASCADE, null=False)
    def clean(self) -> None:
        if self.amount_items < self.clothe.amount:
            raise ValidationError("Amount Item more than Amount Clothes ")
        return super(ClotheItem, self).clean()

    def __str__(self) -> str:
        return f"{self.clothe}"

Criando as migrações e o banco:

python manage.py makemigrations
python manage.py migrate

Tasks

O nosso objetivo é criar uma rota que gere um relatório sobre as vendas da nossa loja que pode ser lida pelo Excel, então vamos criar um arquivo na aplicação chamada tasks.py contendo uma função chamada create_report_of_sale

from openpyxl import Workbook
from core.models import Sale

def create_report_of_sale():
    wb = Workbook()
    ws = wb.active
    ws.append(["ID", "Criado Em", "Total","Vendedor","Roupas", "Quantidade de Roupas"])

    sales = Sale.objects.prefetch_related("items")

    for sale in sales:
        date = sale.created_at.strftime('%Y-%m-%d')
        items = []
        for item in sale.items.all():
            items.append(item.clothe.name)

        ws.append([sale.pk,date,sale.total, sale.saller.name, str(items), len(items)])

    wb.save("store_sales.xlsx")

O código acima apenas cria um arquivo store_sales.xlsx contendo uma tabela com alguns dados das vendas.

Urls

from django.contrib import admin
from django.urls import path
from core.views import CreateReportToSale

urlpatterns = [
    path('admin/', admin.site.urls),
    path('report/sales/', CreateReportToSale.as_view()),
]

Views

Vamos criar a nossa view para gerenciar a nossa rota, adicione o seguinte conteúdo:

from rest_framework.views import APIView
from core.tasks import create_report_of_sale
from rest_framework.response import Response
import time
# Create your views here.
class CreateReportToSale(APIView):

    def get(self, request):
        tb = time.time()
        create_report_of_sale()
        ta = time.time()
        print(f"Tempo Processando:{ta-tb}s")
        return Response(data={"msg":f"Relatório criado com sucesso !"},status=200)

O código acima cria uma view para gerenciar a rota com o método HTTP GET e criar o nosso relatório. Adicionamos um print para saber o tempo que demora isso.

Fazendo acontecer

Para a gente gerar o nosso relatório vamos necessitar de dados, então vamos usar a nossa biblioteca model_bakery. Abra o shell do django com os seguinte comando:

python manage.py shell

Dentro dele, execute os seguintes comandos:

>>> from model_bakery import baker
>>> baker.make("core.ClotheItem", _quantity=3000)

O código acima vai criar 3000 Itens de roupas, 3000 roupas e 3000 vendas, pois o nosso modelo ClotheItem está relacionado com esses modelos e temos o argumento null=False. O código serve para simular a criação de 3000 vendas com 1 item de roupa cada.

Depois entre na rota da criação do relatório, com o projeto rodando, e vamos ver que existe uma demora para ser processada, e também podemos ver o tempo que demorou quando vamos no terminal.

Usando Task Queue na API

Dependências

Esse é um exemplo de código que demora para ser processado, também temos outros exemplos como acesso a API que esteja fora do ar, envio de email e etc. Assim para resolver o nosso problema vamos utilizar Task Queue. Primeiramente vamos instalar a nossa biblioteca celery.

pip install celery

Posteriormente temos que subir o nosso message broker(RabbitMQ) com o seguinte comando docker:

docker run -d -p 5672:5672 rabbitmq

Celery

Com essas duas dependências podemos dá continuidade no nosso projeto. Comece criando um arquivo chamado celery.py com o conteúdo:

from celery import Celery
import django
import os

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'store_app.settings')
django.setup()
app = Celery("store_app", broker='pyamqp://guest@localhost//')

app.autodiscover_tasks()

O código acima está usando as configurações do arquivo settings.py e também está criando uma nova instância do celery, dando o nome dela de store_app e fazendo conexão ao nosso broker pelo protocolo AMQP.

A linha que se segue é para que todos os apps que foram colocados no settings.py possam fazer uso dessa instância.

Uma outra configuração que devemos fazer é dentro do arquivo “init.py” do projeto, afim de inicializar a instância do celery quando o django inicializar o projeto, e assim as nossas aplicações poderão fazer acesso a partir da função shared_task.

from celery import app

__all__ = ('app',)

Tasks

No arquivo tasks.py podemos modificar o código para o seguinte:

from openpyxl import Workbook
from core.models import Sale
from celery import shared_task #<-new
@shared_task(
        bind=True,
        max_retries=5,
        default_retry_delay=30) #<-new
def create_report_of_sale(self): #<-new
    try: #<-new
        wb = Workbook()
        ws = wb.active
        ws.append(["ID", "Criado Em", "Total","Vendedor","Roupas", "Quantidade de Roupas"])

        sales = Sale.objects.prefetch_related("items")

        for sale in sales:
            date = sale.created_at.strftime('%Y-%m-%d')
            items = []
            for item in sale.items.all():
                items.append(item.clothe.name)

            ws.append([sale.pk,date,sale.total, sale.saller.name, str(items), len(items)])

        wb.save("sample.xlsx")
    except: #<-new
        raise self.retry() #<-new

No conteúdo acima estamos decorando a nossa função usando o decorator shared_task, usando este decorator estamos criando uma nova task que poderá ser chamada por qualquer código, e ainda estamos passando alguns argumentos para ele, como:

bind - Nos possibilita acessar determinadas propriedades da task específica, como o objeto de requisição ou a função retry, que nos possibilita executar novamente a task caso algo tenha dado errado;
max_retries - É o contador que adiciona um número máximo de tentativas de chamada da função retry;
default_retry_delay - É o tempo entre uma chamada ao retry e outra. ### View

Na nossa view podemos modificar o código para o seguinte:

from rest_framework.views import APIView
from core.tasks import create_report_of_sale
from rest_framework.response import Response
import time
# Create your views here.
class CreateReportToSale(APIView):

    def get(self, request):
        tb = time.time()
        create_report_of_sale.delay() #<-new
        ta = time.time()
        print(f"Tempo de Processamento:{ta-tb}")
        return Response(data={"msg":f"Relatório criado com sucesso !"},status=200)

Com essa simples chamada a “mágica” pode acontecer, mas para isso é importante criarmos nossos consumidores de tasks, então crie um outro terminal e adicione o seguinte código:

celery -A store_app worker --loglevel=INFO

Obs.: Lembre que o nosso process worker não tem hot reloading, ou seja, todas as vezes que modificar sua task rode o código acima.

Entre no seu projeto e acesse a rota de criação do nosso relatório. Como podemos ver as coisas ocorrem bem mais rápido, pois quando acessamos a nossa rota criamos uma task que foi adicionada ao broker e consumida por um worker disponível, e o tempo de adicionar um task no broker é bem mais rápida do que processa-la.

Repositório:
Código Desenvolvido

Referências

https://docs.celeryq.dev/en/stable/index.html

https://denibertovic.com/posts/celery-best-practices/

https://www.django-rest-framework.org

https://docs.djangoproject.com/en/4.1/

Entendendo o problema de N+1 query usando o Django

Wesley de Morais — Mon, 13 Feb 2023 01:44:54 +0000

Sumário

Introdução do que é o problema n+1 query
Entendendo a situação problema em Django
Criação da API com DRF
Resolvendo usando o que o Django dispõe por padrão
Entendendo o Django Virtual Model
Referências

Introdução do que é o problema n+1 query

O problema N+1 query é um problema que ocorre quando existem determinadas associações entre entidades na aplicação, desta forma quando é necessário acessar informações de A em relação a B ocorre muitas requisições ao banco de dados para cada entidade de A, tornando a aplicação lenta e possivelmente gastando mais do que o necessário quando a aplicação está em produção.

Entendendo a situação problema em Django

Para podermos entender o problema de n+1 query, vamos a um exemplo, imagine uma aplicação na qual usuários de uma plataforma postam vídeos, e esses vídeos tem determinadas categorias que o próprio usuário pode adicionar para que o vídeo chegue em pessoas que gostam dessas categorias. Desta forma podemos ter o seguinte diagrama entidade relacionamento:

Podemos pensar um pouco em relação ao primeiro relacionamento, um usuário publica muitos vídeos, e um vídeo é de um usuário, temos assim um relacionamento 1 para muitos. Imagine que temos 3 usuários na plataforma inteira(Começou agora), e cada um postou 1 vídeo(Uma máquina de postagem), e temos uma rota na nossa API que nos mostra informações dos usuários e o título do vídeo que ele postou, então vai ser necessário 1 query para pegar todos os Users e N queries para pegar as informações dos vídeos relacionados a cada User. Assim temos:

SELECT * FROM User ...
SELECT title FROM Video WHERE userID = 1
SELECT title FROM Video WHERE userID = 2
SELECT title FROM Video WHERE userID = 3

O grande problema ocorre quando aumentamos o número de usuários. Em relação a segunda associação as ideias são as mesmas. Então já podemos imaginar qual a solução para problema, eu vou precisar fazer uma query para pegar todos os usuários e uma outra para pegar os vídeos de todos os usuários

SELECT * FROM User ...
SELECT title FROM Video WHERE userID in (1,2,3)

Mas surge a pergunta, como faço isso usando o ORM do django?

Criação da API com DRF

Pré-requisitos

Entendimento básico sobre o funcionamento do Django
Entendimento básico sobre o funcionamento do Django Rest Framework

Mão na massa

Crie uma pasta chamada video_plataform e instale um ambiente virtual para não poluir seu ambiente.

python -m venv venv

Instalando dependências:

pip install django djangorestframework

Criar o projeto e a aplicação:

django-admin startproject video_plataform .
django-admin startapp core

Settings

Posteriormente, vamos adicionar no arquivo settings.py as duas aplicações

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',

    "core.apps.CoreConfig", #<- new
    "rest_framework" #<- new
]

Models

Para fazer o mapeamento podemos adicionar o seguinte código no arquivo models.py e fazer a migração ao banco de dados:

from django.db import models


class User(models.Model):
    name = models.CharField(max_length=30)

    def __str__(self) -> str:
        return self.name

class Video(models.Model):
    title = models.CharField(max_length=30)
    url = models.CharField(max_length=255)
    visualizations = models.IntegerField(default=0)
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="videos")

    def __str__(self) -> str:
        return self.title

class Category(models.Model):
    title = models.CharField(max_length=30)
    videos = models.ManyToManyField(Video, related_name="categories")

    def __str__(self) -> str:
        return self.title

python manage.py makemigrations
python manage.py migrate

Povoando o banco de dados

Crie um diretório chamado fixtures e dentro dele crie um arquivo chamado data.json com o seguinte conteúdo:

[   {
    "model": "core.user",
    "pk": 1,
    "fields": {
        "name": "Paul"
        }
    },{
        "model": "core.video",
        "pk": 1,
        "fields": {
            "title": "Dancinha do loirinho",
            "url":"www.google.com.br",
            "visualizations": 0,
            "user": 1
        }
    },{
    "model": "core.user",
    "pk": 2,
    "fields": {
        "name": "Marie"
        }
    },{
        "model": "core.video",
        "pk": 2,
        "fields": {
            "title": "Dancinha do forró",
            "url":"www.google.com.br",
            "visualizations": 0,
            "user": 2
        }
    },{
    "model": "core.user",
    "pk": 3,
    "fields": {
        "name": "Mary"
        }
    },{
        "model": "core.video",
        "pk": 3,
        "fields": {
            "title": "The office",
            "url":"www.google.com.br",
            "visualizations": 0,
            "user": 3
        }
    },{
    "model": "core.category",
    "pk": 1,
    "fields": {
        "title": "Entretenimento",
        "videos":[1,2,3]
        }
    },{
    "model": "core.category",
    "pk": 2,
    "fields": {
        "title": "Forró",
        "videos":[2]
        }
    },{
    "model": "core.category",
    "pk": 3,
    "fields": {
        "title": "Funk",
        "videos":[1]
        }}]

No terminal execute o comando:

python manage.py loaddata data.json

Serializer

Crie o arquivo serializers.py e coloque o conteúdo:

from rest_framework import serializers
from core.models import User, Video, Category

class VideoSerializer(serializers.ModelSerializer):

    class Meta:
        model = Video
        fields = ("title","url","visualizations")
class UserSerializer(serializers.ModelSerializer):
    videos = VideoSerializer(many=True)
    class Meta:
        model = User
        fields = ("name", "videos")

class CategorySerializer(serializers.ModelSerializer):
    videos = VideoSerializer(many=True)

    class Meta:
        model = Category
        fields = ("title")

URL

No arquivo urls.py na pasta do projeto adicione o seguinte comando:

from django.contrib import admin
from django.urls import path
from core.views import UserListAPIView #<- new

urlpatterns = [
    path('admin/', admin.site.urls),
    path('users/', UserListAPIView.as_view()) #<- new
]

View

Afim de podermos fazer a listagem de usuários da nossa aplicação vamos criar a nossa view UserListAPIView

from rest_framework.generics import ListAPIView
from core.serializers import UserSerializer
from core.models import User
from django.db import connection, reset_queries
class UserListAPIView(ListAPIView):
    serializer_class = UserSerializer
    def get_queryset(self):
        users = User.objects.all()

        for query in connection.queries:
            print("Query:", query)

        reset_queries()

        return users

O código é um pouco curioso, temos uma classe que herda de ListAPIView, pois eu quero que ela faça a funcionalidade de listagem de todos as instâncias de um determinado recurso de uma forma serializada, assim eu faço o comando User.objects.all(), mas o django nos possibilita saber quais foram as queries feitas ao banco usando o atributo connection.queries, assim caso você acesse a rota pelo browser e for ao seu terminal vai encontra algo semelhante a isso:

Query: {'sql': 'SELECT "core_user"."id", "core_user"."name" FROM "core_user"', 'time': '0.004'}
Query: {'sql': 'SELECT "core_video"."id", "core_video"."title", "core_video"."url", "core_video"."visualizations", "core_video"."user_id" FROM "core_video" WHERE "core_video"."user_id" = 1', 'time': '0.003'}
Query: {'sql': 'SELECT "core_video"."id", "core_video"."title", "core_video"."url", "core_video"."visualizations", "core_video"."user_id" FROM "core_video" WHERE "core_video"."user_id" = 2', 'time': '0.000'}
Query: {'sql': 'SELECT "core_video"."id", "core_video"."title", "core_video"."url", "core_video"."visualizations", "core_video"."user_id" FROM "core_video" WHERE "core_video"."user_id" = 3', 'time': '0.000'}

Se você lembrar são, praticamente, as mesmas queries que mostrei anteriormente, ou seja, podemos ver o problema aqui, caso o número de usuários aumente, então maior vai ser o número de queries feitas para pegar os vídeos deles para essa view.

Resolvendo usando o que o Django dispõe por padrão

Para resolver esse problema vamos modificar o modo como usamos o ORM, assim modifique:

#users = User.objects.all()
#=>
users = User.objects.prefetch_related("videos")

Decorrente que estamos fazer um acesso inverso, pois quem tem a chave estrangeira na relação é o modelo Video e não User, temos que usar o a função prefetch_related() que faz 2 queries e é feito um join por meio do python, também temos o select_related() que faz a mesma coisa, mas deve ser usado em modelos que tenha a chave estrangeira e também ocorre o join por meio do banco de dados, assim fazendo apenas 1 query. Portanto, se acessarmos a rota e formos no terminal vamos ver algo parecido com:

Query: {'sql': 'SELECT "core_user"."id", "core_user"."name" FROM "core_user"', 'time': '0.004'}
Query: {'sql': 'SELECT "core_video"."id", "core_video"."title", "core_video"."url", "core_video"."visualizations", "core_video"."user_id" FROM "core_video" WHERE "core_video"."user_id" IN (1, 2, 3)', 'time': '0.003'}

Como podemos ver na query, mesmo que o sistema tenha 1 milhão de usuários, apenas será feita 2 requisições ao banco de dados, e não 1 milhão de requisições mais 1.

Indo Além

Vamos adicionar um dado novo aos dados do usuário que são mostrados, podemos querer mostrar a quantidade de Vídeos que cada usuário tem, podemos adicionar o seguinte código no arquivo serializer.py:

class UserSerializer(serializers.ModelSerializer):
    videos = VideoSerializer(many=True)
    amount_videos = serializers.SerializerMethodField() #<-new
    class Meta:
        model = User
        fields = ("name", "videos", "amount_videos")   #<-new

    def get_amount_videos(self, obj):                  #<-new
        return Video.objects.filter(user=obj).count()  #<-new

O código acima vai criar uma nova chave nos dados que são mandados para o cliente com a quantidade de vídeos de cada usuário, porém podemos visualizar as queries feitas pelo terminal:

Query: {'sql': 'SELECT "core_user"."id", "core_user"."name" FROM "core_user"', 'time': '0.002'}
Query: {'sql': 'SELECT "core_video"."id", "core_video"."title", "core_video"."url", "core_video"."visualizations", "core_video"."user_id" FROM "core_video" WHERE "core_video"."user_id" IN (1, 2, 3)', 'time': '0.001'}

Query: {'sql': 'SELECT COUNT(*) AS "__count" FROM "core_video" WHERE "core_video"."user_id" = 1', 'time': '0.001'}
Query: {'sql': 'SELECT COUNT(*) AS "__count" FROM "core_video" WHERE "core_video"."user_id" = 2', 'time': '0.000'}
Query: {'sql': 'SELECT COUNT(*) AS "__count" FROM "core_video" WHERE "core_video"."user_id" = 3', 'time': '0.000'}

Veja, as nossas duas queries ainda estão lá(TUDO OK!), evitamos o problema de n+1 query, mas para saber a quantidade de vídeos de cada usuário o problema ainda persiste.

Entendendo o Django Virtual Model

Para podermos resolver o problema proposto, uma das formar é usar uma biblioteca chamada django virtual model, esta biblioteca nos oferece um meio de resolver o problema de N+1 query, mas ainda criar um código com uma alta mantenabilidade.

Instale a biblioteca:

pip install django-virtual-model

Crie um arquivo chamado virtual_models.py na aplicação com o seguinte código:

import django_virtual_models as v
from core.models import User, Video
from django.db.models import Count
class VirtualVideo(v.VirtualModel):
    class Meta:
        model = Video
class VirtualUser(v.VirtualModel):
    videos = VirtualVideo(manager=Video.objects)

    amount_videos = v.Annotation(
        lambda qs, **kwargs: qs.annotate(
            amount_videos=Count("videos")
        ).distinct()
    )
    class Meta:
        model = User

O código acima vai criar modelos virtuais que nos auxiliaram na processo de pré-processamento tanto dos vídeos quanto da quantidade deles, podemos ver que estamos usando uma anotação afim de adicionar no atributo amount_videos a quantidade de vídeos de um usuário.

No nosso arquivo serializers.py podemos modificar o serializer de usuário para o seguinte:

import django_virtual_models as dvm
class UserSerializer(dvm.VirtualModelSerializer):
    videos = VideoSerializer(many=True)
    amount_videos = serializers.IntegerField(read_only=True)

    class Meta:
        model = User
        virtual_model = VirtualUser
        fields = ("name", "videos","amount_videos")

E a nossa view ficará a seguinte:

class UserListAPIView(dvm.VirtualModelListAPIView):
    serializer_class = UserSerializer
    queryset = User.objects.all()

    def get_queryset(self):

        queryset = super().get_queryset()
        for query in connection.queries:
            print("Query:", query)
        reset_queries()
        return queryset

No código acima estamos já usando um queryset geral, pois a nossa biblioteca já cuida de resolver o nosso problema de vídeos e também da quantidade de vídeos. Assim, quando visitarmos a rota e visualizar o terminal podemos ver algo como:

Query: {'sql': 'SELECT DISTINCT "core_user"."id", "core_user"."name", COUNT("core_video"."id") AS "amount_videos" FROM "core_user" LEFT OUTER JOIN "core_video" ON ("core_user"."id" = "core_video"."user_id") GROUP BY "core_user"."id", "core_user"."name"', 'time': '0.001'}
Query: {'sql': 'SELECT "core_video"."id", "core_video"."title", "core_video"."url", "core_video"."visualizations", "core_video"."user_id" FROM "core_video" WHERE "core_video"."user_id" IN (1, 2, 3)', 'time': '0.000'}

Podemos ver que voltamos para a quantidade de 2 queries, pois na query de pegar os dados já é feita a contagem dos vídeos. Legal em!?

Código desenvolvido: Repositório

Referências

Django Virtual Model
Django Rest Framework
Django official

Elixir - Como fazer determinadas operações com a biblioteca Ecto (Parte 2)

Wesley de Morais — Fri, 27 Jan 2023 13:54:33 +0000

Vamos para segunda parte do nosso artigo, agora vamos as operações propriamente ditas.

Fazendo Operações no banco de dados

Todas as operações que iremos fazer será usando o shell que o elixir nos oferece, então digite no terminal:

iex -S mix

Operações sem usar Ecto.Schema

Algumas vezes será necessário fazer operações de banco de dados sem a utilização do Ecto.Schema, e apenas usar as tabelas associadas, porém mesmo sem a utilização do Ecto.Schema não vamos necessitar criar o SQL bruto.

Busca

A primeira operação que vamos fazer será a de busca, com o seguinte comando podemos criar uma query para verificar todos os clientes inseridos no nosso banco de dados:

import Ecto.Query
alias PizzariaEcto.Repo
query = from("clients", select: [:name, :phone])
Repo.all(query)

O código acima vai criar o seguinte sql e executa-lo:

SELECT name, phone FROM clients

Como eu falei a um tempo quando criamos a query ela não será executada, pois o Ecto usa o padrão repositório, ou seja, apenas funções do repositório poderá operar em cima do banco de dados. Assim, essa função Repo.all() irá receber uma query(Uma struct Ecto.Query) e retornar uma lista de mapas contendo as informações no banco, mas como não temos nada no banco será retornado uma lista vazia, então vamos inserir algo no banco de dados.

Inserção

data = [
    %{name: "Bruno Santos", phone: "999878767"},
  %{name: "Maria Pereira", phone: "994657876"},
  %{name: "Fábio Bio", phone: "999870900"}
]
Repo.insert_all("clients", data)
#=> {3, nil}

Com esse código inserirmos 3 clientes na nossa base de dados, então podemos novamente executar o comando de busca para ver esses clientes.

Busca específica

Afim de poder executar uma operação de busca específica podemos executar:

import Ecto.Query
alias PizzariaEcto.Repo
query = from(c in "clients", where: c.name == "Bruno Santos", select: [:name, :phone])
Repo.all(query)

Remoção

Afim de poder executar uma operação de remoção de dados podemos executar:

import Ecto.Query
alias PizzariaEcto.Repo
from(c in "clients", where: c.name == "Bruno Santos") |> Repo.delete_all

Atualização

Podemos fazer atualizações também ainda sem usar schemas:

import Ecto.Query
alias PizzariaEcto.Repo
from(c in "clients", where: c.name == "Fábio Bio") |> Repo.update_all

Operações usando o Ecto.Schema

Com os schemas em mãos podemos fazer a mesma coisa que fazermos sem os schemas, porém o retorno dessas operações vão está em estruturas elixir.

Busca

As buscas usando Ecto.Schema é a mesma coisa de sem usa essa estrutura, mas o retorno da função será uma lista de estruturas elixir populadas.

import Ecto.Query
alias PizzariaEcto.{Repo, Client}
query = from(Client)
Repo.all(query)
#=> [%PizzariaEcto.Client{...},...]

Buscas específicas

Podemos fazer retornos em estruturas elixir usando a função Repo.get_by().

import Ecto.Query
alias PizzariaEcto.{Repo, Client}

Repo.get_by(Client, name: "Maria Pereira")
#=> %PizzariaEcto.Client{...}

Ecto.Changeset

Antes de fazermos operações com inserção e atualização de dados usando as estruturas elixir, podemos falar sobre a estrutura Ecto.Changeset. Esta estrutura nos permite criar casts e validações para dados que são inseridos pelo usuário.

alias PizzariaEcto.Client
import Ecto.Changeset
params = %{name: "Mariana Bom", phone: "999808909", email: "mariana@gmail.com"}
changeset = cast(Client, params, [:name, :phone])
Repo.insert(changeset)

A função cast permiti filtrar quais chaves vão poder serem usadas para criação da estrutura PizzariaEcto.Client, e assim usando a função insert inserir no banco de dados. O ecto tem funções de validação como o validate_length que recebe um changeset e retorna um novo changeset com essa validação aplicada, então podemos melhorar o nosso código:

alias PizzariaEcto.Client
import Ecto.Changeset
params = %{name: "Mariana Bom", phone: "999808909", email: "mariana@gmail.com"}
changeset = cast(%Client{}, params, [:name, :phone]) |> validate_length(:phone, min: 9)
Repo.insert(changeset)

O código acima permite inserir apenas os dados de name e phone que tenha no mínimo 9 dígitos, caso alguma validação não ocorra a inserção não funcionará.

Podemos fazer atualizações usando changeset, o seguinte código vai procurar um cliente chamado “Maria Bom”, criar um changeset com base nesse retorno e mandar para a função update para atualizar o banco.

alias PizzariaEcto.{Repo,Client}
import Ecto.Changeset
client = Repo.get_by(Client, name: "Mariana Bom")
changeset = change(client, name: "Mariana Ruim")
Repo.update(changeset)
#=> {:ok, struct_com_cliente_atualizado}

Conclusão

No presente artigo vimos como a ferramenta Ecto nos possibilita fazer operações no banco de dados de nossa preferência. O uso do Ecto também se dá principalmente junto ao framework Phoenix que nos permite criar aplicações web poderosas, afim de resolver alguns problemas existentes.

Referências

Ecto Site Oficial

Programming Ecto - Build Database Apps in Elixir for Scalability and Performance

Elixir - Como fazer determinadas operações com a biblioteca Ecto (Parte 1)

Wesley de Morais — Fri, 27 Jan 2023 13:54:11 +0000

Requisitos

Elixir 1.14.2
Erlang/otp 25

Introdução

O Ecto é uma biblioteca feita em elixir que nos possibilita ter uma Api, assim como uma abstração para fazer operações em diferentes tipos de bancos de dados. Por meio dele podemos fazer uso de drives de banco de dados como Mysql(Myxql) e Postgresql(Postgrex).

Esta biblioteca se utiliza de um padrão chamado Repository, no qual todo e qualquer acesso ao banco de dados se dará por meio de funções de um determinado repositório criado. O Ecto é dividido em 4 componentes principais:

Ecto.Repo
- Responsável por ser o Repositório de acesso ao banco, ou seja, operações de busca(Query), Inserção, atualização e remoção de dados do banco de dados será por meio de funções dele.
Ecto.Query
- Responsável por ter funções de criação de Query, ou seja, ler do banco de dados alguma informação.
Ecto.Schema
- Responsável por mapear determinados dados externos para estruturas elixir.
Ecto.Changeset
- Responsável por validar e filtrar determinados dados recebidos, afim de acessar o banco de dados.

Situação problema

Nada melhor do que entender uma nova tecnologia do que em uma situação problema que você vai aplica-la.

Imagine que um determinado proprietário de uma pizzaria tenha apenas o interesse de manter os dados dos clientes, assim como as pizzas que ele poderá fazer e os pedidos desses clientes para determinadas pizzas existentes.

Podemos mapear um modelo relacional dessa maneira:

Mão na massa

Criando o projeto

Primeiramente vamos criar o nosso projeto usando a ferramenta Mix, então digite no terminal:

mix new pizzaria_ecto --sup

Caso você nunca criou um projeto com o mix, aqui estamos apenas gerando uma estrutura inicial e adicionando um supervisor na nosso projeto, precisamos desse supervisor, pois a nossa aplicação com ecto será um processo e caso ocorra algum erro nele, por alguma razão, então o nosso supervisor irá reiniciar o nosso processo do inicio(let it crash 🥳).

Instalando as dependências

Antes de a gente trabalharmos com o ecto temos que ter essa biblioteca no nosso projeto, então vamos adicionar essas duas dependências na função deps dentro do arquivo mix.exs:

defp deps do
    [
      {:postgrex, "~> 0.16.5"},
      {:ecto_sql, "~> 3.9"}
    ]
  end

Aqui estamos usando o drive do postgresql e estamos fazendo uso de uma biblioteca ecto_sql que tem como dependência o ecto, mas adiciona outras funcionalidades para trabalharmos com banco de dados relacionais. Para instalar e compilar as aplicações devemos rodar o seguinte comando no terminal:

mix do deps.get, compile

Criando o nosso próprio repositório

Dentro do diretório pizzaria_ecto que foi criado poderemos criar um arquivo chamado repo.ex, com seguinte código:

defmodule PizzariaEcto.Repo do
  use Ecto.Repo, otp_app: :pizzaria_ecto, adapter: Ecto.Adapters.Postgres
end

Aqui criamos um módulo que chamamos de PizzariaEcto.Repo no qual ele faz uso do Ecto.Repo, eu gosto de pensar que quando usamos o “use” fazemos com que o nosso módulo seja um módulo Ecto.Repo . Além disso temos a keyword :otp_app que nos diz quais configurações vamos usar dentro de um arquivo chamado config.exs(Vamos criar ainda), e por último termos a keyword :adapter que dizemos para o ecto que vamos usar o banco de dados Postgresql. Agora vamos criar um diretório chamado config(Na raiz do projeto) e dentro dele vamos criar um arquivo config.exs com seguinte código:

import Mix.Config


config :pizzaria_ecto, ecto_repos: [PizzariaEcto.Repo]

config :pizzaria_ecto, PizzariaEcto.Repo,
  database: "pizzaria_ecto",
  username: "postgres", # Username "postgres" is stardard user in pgdb
  password: "postgres", # Password "postgres" is our password when created pg container
  hostname: "localhost" # We expose our port when created our pg container

Na código acima estamos fazendo algumas configurações para o nosso querido ecto, no qual temos o uso da macro config que setamos qual o nome da configuração, veja que usamos o mesmo atom usado na keyword :opt_app no módulo PizzariaEcto.Repo, além disso temos uma keyword dizendo quais serão os repositórios no nosso projeto, poderiamos ter mais de um acessando diferentes banco de dados.

Na segunda chamada a macro config temos as informações necessárias para o adapter acessar o banco de dados rodando. Opps, ainda não subimos um container postgresql, então vamos fazer isso.

Subindo container postgresql

No diretório home do projeto vamos criar um arquivo docker-compose.yml com seguinte código:

version: '3.9'

services:
  db:
    container_name: "pg_pizzaria_ecto"
    image: postgres:latest
    environment:
      POSTGRES_PASSWORD: postgres
    volumes:
      - postgres-db:/var/lib/postgresql/data
    ports:
      - '5432:5432'

volumes:
  postgres-db:

Posteriormente vamos subir o container com seguinte comando no terminal:

docker-compose up -d

Criando Schemas

Uma das formas de mapear determinados dados externos para estruturas elixir é usando o Ecto.Schema, com elas é possível fazer operações no banco de dados com retornos em estruturas elixir específicas. Dentro do diretório pizzaria_ecto crie um arquivo chamado client.ex com o seguinte conteúdo:

defmodule PizzariaEcto.Client do
  use Ecto.Schema
  alias PizzariaEcto.{Order}
  schema "clients" do
    field :name, :string
    field :phone, :string
      timestamps() # field :updated_at, :naive_datetime
                                 # field :inserted_at, :naive_datetime 

    has_many :order, Order
  end
end

Aqui temos um módulo que é um Ecto.Schema, assim algumas funções e macros podem ser chamadas dentro do módulo que ele é especificado, um destes macros é a schema que recebe como argumento a tabela no banco de dados, assim ele vai mapear determinados campos para uma tabela no banco de dados, porém ainda não será criada as tabelas no banco de dados.

Associação Um para Muitos

Algo curioso que podemos ver no schema de PizzariaEcto.Client é a macro has_many, aqui estamos tentando mapear Um cliente podendo fazer vários pedidos, porém para ter o “casamento” deste relacionamento é necessário criar o segundo schema, então crie um arquivo chamado order.ex com o conteúdo:

defmodule PizzariaEcto.Order do
  use Ecto.Schema
  alias PizzariaEcto.{Client, Pizza, OrderPizza}

  schema "orders" do
    field :price, :float
    timestamps()

    belongs_to  :client, Client
    many_to_many :pizza, Pizza, join_through: OrderPizza
  end
end

No código acima fizemos a mesma coisa do código anterior, mapeamos determinados campos com uma tabela no banco de dados, entretanto temos a outra parte do primeiro relacionamento entre Cliente e Order, em que usamos a macro belongs_to, com esse relacionamento estabelecido é esperado que tenha uma coluna na tabela orders chamada client_id.

Associação Muitos para Muitos

Um outro relacionamento que precisamos mapear é entre Order e Pizza, como é uma associação muito para muitos é necessário a criação de uma terceira tabela que vai fazer o papel de guardar order_id e pizza_id, assim é necessário usar a macro many_to_many em ambos os schemas e usar a keyword join_through mostrando qual o schema será usado. Vamos criar o schema de pizza, então crie um arquivo chamado pizza.ex, com o conteúdo:

defmodule PizzariaEcto.Pizza do
  use Ecto.Schema
  alias PizzariaEcto.{Order,OrderPizza}

  schema "pizzas" do
    field :name, :string
    timestamps()

    many_to_many :order, Order, join_through: OrderPizza
  end

end

Vamos também criar o schema OrderPizza, então crie o arquivo order_pizza.ex.

defmodule PizzariaEcto.OrderPizza do
  use Ecto.Schema
  alias PizzariaEcto.{Pizza, Order}

  schema "orders_pizzars" do
    timestamps()
    belongs_to  :order, Order
    belongs_to  :pizza, Pizza

  end
end

Ótimo, com esses schemas criados poderemos utiliza-los para fazer determinadas operações no banco de dados, mas com a criação desses ainda não estão criadas as tabelas no banco de dados, para isso é necessário criar as migrações, então vamos lá.

Criando migrations

Tabela de Cliente

O Ecto se utiliza de um componente chamado Ecto.Migrations para fazer as migrações necessárias, então para cria-la podemos usar um linha de comando no terminal, então digite:

mix ecto.gen.migrations add_clients_table

O código acima vai criar um arquivo dentro do seguinte path priv/SEU_REPO/migrations/ID_ALEATÓRIO_add_clients_table , como eu já mencionei podemos ter vários repositórios, então este SEU_REPO significa qual repositório será criado a migração, porém como existe apenas um, então será ele. O conteúdo do arquivo criado, será o seguinte:

defmodule PizzariaEcto.Repo.Migrations.AddClientsTable do
  use Ecto.Migration

  def change do
  end
end

Esta função change será a chamada pelo Ecto para fazer a migração, então vamos adicionar a nossa tabela:

defmodule PizzariaEcto.Repo.Migrations.AddClientsTable do
  use Ecto.Migration

  def change do
    create table("clients") do
      add :name, :string, null: false
      add :phone, :string, null: false
      timestamps()
    end

    create index("clients", :name)
  end
end

Temos aqui a macro create que recebe o que será criado no banco e um bloco de código, então estamos criando a tabela clients com algumas colunas, podemos perceber que o nome aqui é o mesmo que colocamos no schema PizzariaEcto.Client. Além da tabela poderemos criar index para maximizar as buscas por uma coluna no banco. Podemos criar agora a tabela de Order.

Tabela de Order

mix ecto.gen.migrations add_orders_table

Criado o arquivo priv/SEU_REPO/migrations/ID_ALEATÓRIO_add_orders_table

defmodule PizzariaEcto.Repo.Migrations.AddOrdersTable do
  use Ecto.Migration

  def change do
    create table("orders") do
      add :price, :float, null: false
      add :client_id, references("clients",  on_delete: :nothing)
      timestamps null: true
    end
  end
end

Veja que pelo fato de termos uma associação de 1 para muitos entre Cliente e Order , então é necessário que orders tenha a foreign_key, então podemos adicionar usando o tipo references.

Tabela de Pizza

mix ecto.gen.migrations add_pizzas_table

defmodule PizzariaEcto.Repo.Migrations.AddPizzasTable do
  use Ecto.Migration

  def change do
    create table("pizzas") do
      add :name, :string, null: false

      timestamps null: true
    end
  end
end

No código acima criamos a migration de pizza, mas podemos voltar no schema de Pizza e Order, e ver que eles tem uma relação de muitos para muitos entre eles, e criamos um schema OrderPizza para receber as foreign_keys, então vamos criar a tabela orders_pizzas que mapeamos no schema OrderPizza.

mix ecto.gen.migrations add_orders_pizzas_table

defmodule PizzariaEcto.Repo.Migrations.AddOrdersPizzasTable do
  use Ecto.Migration

  def change do
    create table("orders_pizzas") do
      add :order_id, references("orders")
      add :pizza_id, references("pizzas")
      timestamps null: true
    end

    create index(:orders_pizzas, :order_id)
    create index(:orders_pizzas, :pizza_id)
  end
end

OK, todas as migrations foram criadas até agora, mas é necessário executar a migração, então vamos digitar no terminal:

mix ecto.migrate

E done!

Vamos para parte 2:
Segunda Parte do Artigo

Referências

Ecto Site Oficial

Programming Ecto - Build Database Apps in Elixir for Scalability and Performance