У меня давно была идея написать свою ORM для Python. Подтолкнуло на нее меня отсутствие асинхронных ORM. Это задача очень сложная и для ее реализации я ее решил разбить на подзадачи. Первая – это написать простенький SQL Builder, который станет основой для ORM. Я изучил исходники подобных продуктов на Ruby, Python, JavaScript, C# и Scala. На это у меня ушло пару дней. Сегодня я решил набросать основу:
from abc import abstractmethod
from functools import wraps
from typing import Any
__version__ = '0.1.0'
class Expr:
@abstractmethod
def to_sql(self):
raise NotImplementedError
class Table(Expr):
def __init__(
self,
name: str,
*,
schema: str = None,
alias: str = None
) -> None:
self.name = name
self.schema = schema
self.alias = alias
def __getitem__(self, name) -> Expr:
if name == '*':
return Star(self)
return Field(self, name)
__getattr__ = __getitem__
class Star(Expr):
def __init__(self, table: Table) -> None:
self.table = table
# TODO: нужно придумать другое имя
def to_expr(f):
@wraps(f)
def wrapper(self, other):
if isinstance(other, Expr):
other = Value(other)
return f(self, other)
return wrapper
class Field(Expr):
def __init__(self, table: Table, name: str) -> None:
self.table = table
self.name = name
@to_expr
def __eq__(self, other: Any) -> 'EQ':
return EQ(self, other)
@to_expr
def __ne__(self, other: Any) -> 'NE':
return NE(self, other)
@to_expr
def __lte__(self, other: Any) -> 'LTE':
return LTE(self, other)
@to_expr
def __gte__(self, other: Any) -> 'GTE':
return GTE(self, other)
@to_expr
def __lt__(self, other: Any) -> 'LT':
return LT(self, other)
@to_expr
def __gt__(self, other: Any) -> 'GT':
return GT(self, other)
@to_expr
def __or__(self, other: Any) -> 'OR':
return OR(self, other)
@to_expr
def __and__(self, other: Any) -> 'AND':
return AND(self, other)
class Value(Expr):
def __init__(self, value: Any) -> None:
self.value = value
class BinOp(Expr):
def __init__(self, left: Expr, right: Expr) -> None:
self.left = left
self.right = right
class EQ(BinOp):
...
class NE(BinOp):
...
class LTE(BinOp):
...
class GTE(BinOp):
...
class LT(BinOp):
...
class GT(BinOp):
...
class OR(BinOp):
...
class AND(BinOp):
...
class Query:
def __init__(self, dsn: str) -> None:
self.dsn = dsn
def select(self, *args, **kw) -> 'Select':
return Select(self, *args, **kw)
class Statement(Expr):
...
class Select(Statement):
def __init__(self, db, *args, **kw) -> None:
self.db = db
def from_(self, *args, **kw):
return self
def where(self, *args, **kw):
return self
def offset(self, *args, **kw):
return self
def limit(self, *args, **kw):
return self
def paginate(self, page: int = 1, per_page: int = 10):
return self.offset((page - 1) * per_page).limit(per_page)
def join(self, *args, **kw):
return self
def order_by(self, *args, **kw):
return self
def fetch(self):
sql = self.to_sql()
...
def fetchall(self):
...
def single(self):
...
class Order(Expr):
def __init__(self, field: Field) -> None:
self.field = Field
class ASC(Order):
...
class DESC(Order):
...
q = Query('postgresql:///test')
p = Table('posts')
u = Table('users')
posts = q.select(p['*'], author=u.username) \
.from_(p) \
.join(u, p.author_id == u.id) \
.where(p.deleted == False) \
.order_by(DESC(p.published_at)) \
.paginate() \
.fetchall()
Примерно такая архитектура классов должна быть в библиотеке. Мне интересны ваши мнения. Может я что-то упускаю из виду. Я смутно представляю как все должно работать. Макет я набросал за 2 часа. Я не ставлю целью создать универсальный SQL Builder. Мне нужны только select с подзапросами, insert, update и delete.