49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
from dataclasses import dataclass, field
|
|
from typing import List, Dict, Any, Optional
|
|
from sqlalchemy import asc, desc, select, func
|
|
from sqlalchemy.orm import selectinload, joinedload, Load
|
|
|
|
@dataclass
|
|
class QuerySpec:
|
|
filters: Dict[str, Any] = field(default_factory=dict)
|
|
order_by: List[str] = field(default_factory=list)
|
|
page: Optional[int] = None
|
|
per_page: Optional[int] = None
|
|
expand: List[str] = field(default_factory=list)
|
|
fields: Optional[List[str]] = None
|
|
|
|
FILTER_OPS = {
|
|
"__eq": lambda c, v: c == v,
|
|
"__ne": lambda c, v: c != v,
|
|
"__lt": lambda c, v: c < v,
|
|
"__lte": lambda c, v: c <= v,
|
|
"__gt": lambda c, v: c > v,
|
|
"__gte": lambda c, v: c >= v,
|
|
"__ilike": lambda c, v: c.ilike(v),
|
|
"__in": lambda c, v: c.in_(v),
|
|
"__isnull": lambda c, v: (c.is_(None) if v else c.is_not(None))
|
|
}
|
|
|
|
def build_query(Model, spec: QuerySpec, eager_policy=None):
|
|
stmt = select(Model).where(Model.deleted.is_(False))
|
|
# filters
|
|
for raw_key, val in spec.filters.items():
|
|
for op in FILTER_OPS:
|
|
if raw_key.endswith(op):
|
|
colname = raw_key[: -len(op)]
|
|
col = getattr(Model, colname)
|
|
stmt = stmt.where(FILTER_OPS[op](col, val))
|
|
break
|
|
else:
|
|
stmt = stmt.where(getattr(Model, raw_key) == val)
|
|
# order_by
|
|
for key in spec.order_by:
|
|
desc_ = key.startswith("-")
|
|
col = getattr(Model, key[1:] if desc_ else key)
|
|
stmt = stmt.order_by(desc(col) if desc_ else asc(col))
|
|
# eager loading
|
|
if eager_policy:
|
|
opts = eager_policy(Model, spec.expand)
|
|
if opts:
|
|
stmt = stmt.options(*opts)
|
|
return stmt
|