Criei um ORM em Python para minha empresa
Aqui na RPS Capital praticamente 100% do nosso dia a dia é lidando com dados no pandas, e muitos desses dados são implementados no SQL de uma maneira que cada projeto seja independente de models definidos via código (sei que é uma má prática, mas precisamos de muita agilidade e pouca complexidade).
Cada consulta que fazemos aqui dentro é única, mesmo puxando dados das mesmas tabelas, o que torna difícil construir funções padronizadas para consultas.
O Problema (ou meu toc)
Muitos aqui ainda fazem consultas da seguinte maneira no Python
import rps_databases # módulo que usamos internamente
db = rps_databases.connect()
SQL = """
SELECT column_a, column_b
FROM myschema.mytable
WHERE id in %s
"""
ids = [1,2,3]
df = db.fetch(SQL, (ids,))
É um exemplo simples, porém, como a query é um texto, eu poderia escrever ela em infinitas variações, que é o que acontece. Eu escrevi do jeito que eu acredito ser o mais legível, mas mesmo assim, não é um padrão.
A solução
df = db.myschema.mytable.get(columns=["column_a", "column_b"], id=ids)
Foi possível resumir em 1 linha o que foi feito em 6 linhas.
O que aconteceu aqui?
A classe Database tem acesso dinâmico às propriedades dela, se eu tento acessar uma propriedade que não existe, ela automaticamente é considerada como um "schema". Isso foi possível utilizando a função "__getattribute__":
class Database:
...
def __getattribute__(self, attr):
try:
return object.__getattribute__(self, attr)
except AttributeError:
return Schema(db=self, name=attr)
A mesma coisa acontece com a classe Schema, que retorna uma Table ao acessar qualquer propriedade:
class Schema:
def __init__(self, db, name: str):
self.db = db
self.name = name
def __getattribute__(self, attr):
try:
return object.__getattribute__(self, attr)
except AttributeError:
return Table(db=self.db, schema=self, name=attr)
Sério, o schema só serve pra isso.
O resto da mágica acontece na classe Table:
class Table:
def __init__(self, db: "Database", schema: "Schema", name: str):
self.schema = schema
self.name = name
self.db = db
def path(self):
return f"{self.schema.name}.{self.name}"
Como ela conhece o caminho do schema e da tabela, só precisamos de uma função para construir a query baseado em alguns requísitos.
def get(
self,
columns: Optional[list] = None,
distinct: Optional[list] = None,
min: Optional[list] = None,
max: Optional[list] = None,
sum: Optional[list] = None,
avg: Optional[list] = None,
**simple_where,
) -> pd.DataFrame:
...
Com apenas esses parâmetros, já dá pra fazer muita coisa... Aqui vai uns exemplos:
from rps_databases import operators as o
# Idade média por sobrenome onde os usuários são de curitiba
db.public.users.get(columns=["last_name"], avg=["age"], city="Curitiba")
# Idade média por sobrenome onde os usuários não são de curitiba
db.public.users.get(columns=["last_name"], avg=["age"], city=o.Not("Curitiba"))
# Idade média por sobrenome onde os usuários são de curitiba e são paulo
db.public.users.get(columns=["last_name"], avg=["age"], city=["Curitiba", "São Paulo"])
# Idade média por sobrenome onde os usuários não são de curitiba e são paulo
db.public.users.get(columns=["last_name"], avg=["age"], city=o.Not(["Curitiba", "São Paulo"]))
# Data mais recente em uma tabela
db.public.transactions.get(max=["transaction_date"])
# Data mais recente trocando de nome
db.public.transactions.get(max=[
("transaction_date", "most_recent_date") # pega o max de transaction_date e renomeia pra most_recent_date
])
# Consulta todas as colunas filtrando as linhas que o last_name contenha em algum lugar "A"
db.public.users.get(last_name=o.Ilike("A"))
Acho que deu pra pegar a ideia, o objetivo foi deixar consultas menos verbosas e menos complexas sem depender de models/schemas definidos.