Mecanismo sort para tablas
This commit is contained in:
61
backend/api/v1/modules/a76/common/sorting.py
Normal file
61
backend/api/v1/modules/a76/common/sorting.py
Normal file
@@ -0,0 +1,61 @@
|
||||
from typing import Any, Optional, Type
|
||||
from sqlalchemy import asc, desc, inspect
|
||||
from sqlalchemy.orm import Query, RelationshipProperty
|
||||
|
||||
def apply_sorting(
|
||||
query: Query,
|
||||
model: Type[Any],
|
||||
sort_by: Optional[str] = None,
|
||||
sort_desc: bool = True,
|
||||
default_sort_col: str = "created_at"
|
||||
) -> Query:
|
||||
"""
|
||||
Applies dynamic sorting to a SQLAlchemy query.
|
||||
Supports nested attributes via dot notation (e.g., 'compliance_mx.remesa').
|
||||
Automatically handles joins if necessary.
|
||||
"""
|
||||
if not sort_by:
|
||||
if hasattr(model, default_sort_col):
|
||||
col = getattr(model, default_sort_col)
|
||||
return query.order_by(desc(col))
|
||||
return query
|
||||
|
||||
try:
|
||||
parts = sort_by.split('.')
|
||||
current_model = model
|
||||
|
||||
# Traverse relationships if dot notation is used
|
||||
for i, part in enumerate(parts[:-1]):
|
||||
# Check if relationship exists
|
||||
mapper = inspect(current_model)
|
||||
if part in mapper.relationships:
|
||||
rel = mapper.relationships[part]
|
||||
# Join the relationship
|
||||
query = query.join(rel.entity.class_)
|
||||
current_model = rel.entity.class_
|
||||
else:
|
||||
# If part is not a relationship, we can't go deeper
|
||||
# Fallback to default sorting
|
||||
if hasattr(model, default_sort_col):
|
||||
return query.order_by(desc(getattr(model, default_sort_col)))
|
||||
return query
|
||||
|
||||
# The last part is the actual column
|
||||
last_part = parts[-1]
|
||||
if hasattr(current_model, last_part):
|
||||
col = getattr(current_model, last_part)
|
||||
if sort_desc:
|
||||
query = query.order_by(desc(col))
|
||||
else:
|
||||
query = query.order_by(asc(col))
|
||||
else:
|
||||
# Fallback to default sorting on the original model
|
||||
if hasattr(model, default_sort_col):
|
||||
query = query.order_by(desc(getattr(model, default_sort_col)))
|
||||
except Exception as e:
|
||||
# Log error or handle gracefully
|
||||
print(f"Error applying sort for {sort_by}: {e}")
|
||||
if hasattr(model, default_sort_col):
|
||||
query = query.order_by(desc(getattr(model, default_sort_col)))
|
||||
|
||||
return query
|
||||
Reference in New Issue
Block a user