205 lines
7.7 KiB
Python
205 lines
7.7 KiB
Python
from typing import Any, Dict, Iterable, List, Tuple
|
|
from sqlalchemy import Column, Integer, DateTime, Boolean, String, JSON, func, inspect
|
|
from sqlalchemy.orm import declarative_mixin, declarative_base, NO_VALUE
|
|
|
|
Base = declarative_base()
|
|
|
|
def _safe_get_loaded_attr(obj, name):
|
|
try:
|
|
st = inspect(obj)
|
|
attr = st.attrs.get(name)
|
|
if attr is not None:
|
|
val = attr.loaded_value
|
|
return None if val is NO_VALUE else val
|
|
if name in st.dict:
|
|
return st.dict.get(name)
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
def _split_field_tokens(fields: Iterable[str]) -> Tuple[List[str], Dict[str, List[str]]]:
|
|
"""
|
|
Split requested fields into:
|
|
- scalars: ["label", "name"]
|
|
- collections: {"updates": ["id", "timestamp","content"], "owner": ["label"]}
|
|
Any dotted token "root.rest.of.path" becomes collections[root].append("rest.of.path").
|
|
Bare tokens ("foo") land in scalars.
|
|
"""
|
|
scalars: List[str] = []
|
|
groups: Dict[str, List[str]] = {}
|
|
for raw in fields:
|
|
f = str(raw).strip()
|
|
if not f:
|
|
continue
|
|
# bare token -> scalar
|
|
if "." not in f:
|
|
scalars.append(f)
|
|
continue
|
|
# dotted token -> group under root
|
|
root, tail = f.split(".", 1)
|
|
if not root or not tail:
|
|
continue
|
|
groups.setdefault(root, []).append(tail)
|
|
return scalars, groups
|
|
|
|
def _deep_get_loaded(obj: Any, dotted: str) -> Any:
|
|
"""
|
|
Deep get with no lazy loads:
|
|
- For all but the final hop, use _safe_get_loaded_attr (mapped-only, no getattr).
|
|
- For the final hop, try _safe_get_loaded_attr first; if None, fall back to getattr()
|
|
to allow computed properties/hybrids that rely on already-loaded columns.
|
|
"""
|
|
parts = dotted.split(".")
|
|
if not parts:
|
|
return None
|
|
|
|
cur = obj
|
|
# Traverse up to the parent of the last token safely
|
|
for part in parts[:-1]:
|
|
if cur is None:
|
|
return None
|
|
cur = _safe_get_loaded_attr(cur, part)
|
|
if cur is None:
|
|
return None
|
|
|
|
last = parts[-1]
|
|
# Try safe fetch on the last hop first
|
|
val = _safe_get_loaded_attr(cur, last)
|
|
if val is not None:
|
|
return val
|
|
# Fall back to getattr for computed/hybrid attributes on an already-loaded object
|
|
try:
|
|
return getattr(cur, last, None)
|
|
except Exception:
|
|
return None
|
|
|
|
def _serialize_leaf(obj: Any) -> Any:
|
|
"""
|
|
Lead serialization for values we put into as_dict():
|
|
- If object has as_dict(), call as_dict() with no args (caller controls field shapes).
|
|
- Else return value as-is (Flask/JSON encoder will handle datetimes, etc., via app config).
|
|
"""
|
|
if obj is None:
|
|
return None
|
|
ad = getattr(obj, "as_dict", None)
|
|
if callable(ad):
|
|
try:
|
|
return ad(None)
|
|
except Exception:
|
|
return str(obj)
|
|
return obj
|
|
|
|
def _serialize_collection(items: Iterable[Any], requested_tails: List[str]) -> List[Dict[str, Any]]:
|
|
"""
|
|
Turn a collection of ORM objects into list[dict] with exactly requested_tails,
|
|
where each tail can be dotted again (e.g., "author.label"). We do NOT lazy-load.
|
|
"""
|
|
out: List[Dict[str, Any]] = []
|
|
# Deduplicate while preserving order
|
|
uniq_tails = list(dict.fromkeys(requested_tails))
|
|
for child in (items or []):
|
|
row: Dict[str, Any] = {}
|
|
for tail in uniq_tails:
|
|
row[tail] = _deep_get_loaded(child, tail)
|
|
# ensure id present if exists and not already requested
|
|
try:
|
|
if "id" not in row and hasattr(child, "id"):
|
|
row["id"] = getattr(child, "id")
|
|
except Exception:
|
|
pass
|
|
out.append(row)
|
|
return out
|
|
|
|
@declarative_mixin
|
|
class CRUDMixin:
|
|
id = Column(Integer, primary_key=True)
|
|
created_at = Column(DateTime, default=func.now(), nullable=False)
|
|
updated_at = Column(DateTime, default=func.now(), nullable=False, onupdate=func.now())
|
|
|
|
def as_dict(self, fields: list[str] | None = None):
|
|
"""
|
|
Serialize the instance.
|
|
|
|
Behavior:
|
|
- If 'fields' (possibly dotted) is provided, emit exactly those keys.
|
|
* Bare tokens (e.g., "label", "owner") return the current loaded value.
|
|
* Dotted tokens for one-to-many (e.g., "updates.id","updates.timestamp")
|
|
produce a single "updates" key containing a list of dicts with the requested child keys.
|
|
* Dotted tokens for many-to-one/one-to-one (e.g., "owner.label") emit the scalar under "owner.label".
|
|
- Else, if '__crudkit_projection__' is set on the instance, use that.
|
|
- Else, fall back to all mapped columns on this class hierarchy.
|
|
|
|
Always includes 'id' when present unless explicitly excluded (i.e., fields explicitly provided without id).
|
|
"""
|
|
req = fields if fields is not None else getattr(self, "__crudkit_projection__", None)
|
|
|
|
if req:
|
|
# Normalize and split into (scalars, groups of dotted by root)
|
|
req_list = [p for p in (str(x).strip() for x in req) if p]
|
|
scalars, groups = _split_field_tokens(req_list)
|
|
|
|
out: Dict[str, Any] = {}
|
|
|
|
# Always include id unless user explicitly listed fields and included id already
|
|
if "id" not in req_list and hasattr(self, "id"):
|
|
try:
|
|
out["id"] = getattr(self, "id")
|
|
except Exception:
|
|
pass
|
|
|
|
for name in scalars:
|
|
# Try loaded value first (never lazy-load)
|
|
val = _safe_get_loaded_attr(self, name)
|
|
|
|
# if still None, allow a final-hop getattr for root scalars
|
|
# so hybrids / @property can compute (they won't traverse relationships).
|
|
if val is None:
|
|
try:
|
|
val = getattr(self, name)
|
|
except Exception:
|
|
val = None
|
|
|
|
# If it's a collection and no subfields were requested, emit a light list
|
|
if isinstance(val, (list, tuple)):
|
|
out[name] = [_serialize_leaf(v) for v in val]
|
|
else:
|
|
out[name] = val
|
|
|
|
# Handle dotted groups: root -> [tails]
|
|
for root, tails in groups.items():
|
|
root_val = _safe_get_loaded_attr(self, root)
|
|
if isinstance(root_val, (list, tuple)):
|
|
# one-to-many collection
|
|
out[root] = _serialize_collection(root_val, tails)
|
|
else:
|
|
# many-to-one or scalar dotted; place each full dotted path as key
|
|
for tail in tails:
|
|
dotted = f"{root}.{tail}"
|
|
out[dotted] = _deep_get_loaded(self, dotted)
|
|
|
|
return out
|
|
|
|
# Fallback: all mapped columns on this class hierarchy
|
|
result: Dict[str, Any] = {}
|
|
for cls in self.__class__.__mro__:
|
|
if hasattr(cls, "__table__"):
|
|
for column in cls.__table__.columns:
|
|
name = column.name
|
|
try:
|
|
result[name] = getattr(self, name)
|
|
except Exception:
|
|
result[name] = None
|
|
return result
|
|
|
|
class Version(Base):
|
|
__tablename__ = "versions"
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
model_name = Column(String, nullable=False)
|
|
object_id = Column(Integer, nullable=False)
|
|
change_type = Column(String, nullable=False)
|
|
data = Column(JSON, nullable=True)
|
|
timestamp = Column(DateTime, default=func.now())
|
|
|
|
actor = Column(String, nullable=True)
|
|
meta = Column('metadata', JSON, nullable=True)
|