42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
from typing import List
|
|
from sqlalchemy.inspection import inspect
|
|
from sqlalchemy.orm import Load, joinedload, selectinload
|
|
|
|
def default_eager_policy(Model, expand: List[str]) -> List[Load]:
|
|
"""
|
|
Heuristic:
|
|
- many-to-one / one-to-one: joinedload
|
|
- one-to-many / many-to-many: selectinload
|
|
Accepts dotted paths like "author.publisher".
|
|
"""
|
|
if not expand:
|
|
return []
|
|
|
|
opts: List[Load] = []
|
|
|
|
for path in expand:
|
|
parts = path.split(".")
|
|
current_model = Model
|
|
current_inspect = inspect(current_model)
|
|
|
|
# first hop
|
|
rel = current_inspect.relationships.get(parts[0])
|
|
if not rel:
|
|
continue # silently skip bad names
|
|
attr = getattr(current_model, parts[0])
|
|
loader: Load = selectinload(attr) if rel.uselist else joinedload(attr)
|
|
current_model = rel.mapper.class_
|
|
|
|
# nested hops, if any
|
|
for name in parts[1:]:
|
|
current_inspect = inspect(current_model)
|
|
rel = current_inspect.relationships.get(name)
|
|
if not rel:
|
|
break
|
|
attr = getattr(current_model, name)
|
|
loader = loader.selectinload(attr) if rel.uselist else loader.joinedload(attr)
|
|
current_model = rel.mapper.class_
|
|
|
|
opts.append(loader)
|
|
|
|
return opts
|