More fixes and additions for forms. We are also haunted by detached sessions constantly.

This commit is contained in:
Yaro Kasear 2025-09-23 13:49:59 -05:00
parent 979a329d6a
commit a3f2c794f5
6 changed files with 160 additions and 49 deletions

View file

@ -74,18 +74,32 @@ def apply_pagination(sel: Select, backend: BackendInfo, *, page: int, per_page:
per_page = max(1, int(per_page))
offset = (page - 1) * per_page
if backend.requires_order_by_for_offset and not sel._order_by_clauses:
if default_order_by is None:
sel = sel.order_by(text("1"))
else:
sel = sel.order_by(default_order_by)
if backend.requires_order_by_for_offset:
# Avoid private attribute if possible:
has_order = bool(getattr(sel, "_order_by_clauses", ())) # fallback for SA < 2.0.30
try:
has_order = has_order or bool(sel.get_order_by())
except Exception:
pass
if not has_order:
if default_order_by is not None:
sel = sel.order_by(default_order_by)
else:
# Try to find a primary key from the FROMs; fall back to a harmless literal.
try:
first_from = sel.get_final_froms()[0]
pk = next(iter(first_from.primary_key.columns))
sel = sel.order_by(pk)
except Exception:
sel = sel.order_by(text("1"))
return sel.limit(per_page).offset(offset)
@contextmanager
def maybe_identify_insert(session: Session, table, backend: BackendInfo):
"""
For MSSQL tables with IDENTIFY PK when you need to insert explicit IDs.
For MSSQL tables with IDENTITY PK when you need to insert explicit IDs.
No-op elsewhere.
"""
if not backend.is_mssql:
@ -93,7 +107,7 @@ def maybe_identify_insert(session: Session, table, backend: BackendInfo):
return
full_name = f"{table.schema}.{table.name}" if table.schema else table.name
session.execute(text(f"SET IDENTIFY_INSERT {full_name} ON"))
session.execute(text(f"SET IDENTITY_INSERT {full_name} ON"))
try:
yield
finally:
@ -101,7 +115,7 @@ def maybe_identify_insert(session: Session, table, backend: BackendInfo):
def chunked_in(column, values: Iterable, backend: BackendInfo, chunk_size: Optional[int] = None) -> ClauseElement:
"""
Build a safe large IN() filter respecting bund param limits.
Build a safe large IN() filter respecting bind param limits.
Returns a disjunction of chunked IN clauses if needed.
"""
vals = list(values)
@ -120,3 +134,12 @@ def chunked_in(column, values: Iterable, backend: BackendInfo, chunk_size: Optio
for p in parts[1:]:
expr = expr | p
return expr
def sql_trim(expr, backend: BackendInfo):
"""
Portable TRIM. SQL Server before compat level 140 lacks TRIM().
Emit LTRIM(RTRIM(...)) there; use TRIM elsewhere
"""
if backend.is_mssql:
return func.ltrim(func.rtrim(expr))
return func.trim(expr)