from __future__ import annotations from dataclasses import dataclass from typing import Tuple, Optional, Iterable from contextlib import contextmanager from sqlalchemy import text, func from sqlalchemy.engine import Engine from sqlalchemy.orm import Session from sqlalchemy.sql.elements import ClauseElement from sqlalchemy.sql import Select @dataclass(frozen=True) class BackendInfo: name: str version: Tuple[int, ...] paramstyle: str is_sqlite: bool is_postgres: bool is_mysql: bool is_mssql: bool supports_returning: bool supports_ilike: bool requires_order_by_for_offset: bool max_bind_params: Optional[int] @classmethod def from_engine(cls, engine: Engine) -> "BackendInfo": d = engine.dialect name = d.name version = tuple(getattr(d, "server_version_info", ()) or ()) is_pg = name in {"postgresql", "postgres"} is_my = name == "mysql" is_sq = name == "sqlite" is_ms = name == "mssql" supports_ilike = is_pg or is_my supports_returning = is_pg or (is_sq and version >= (3, 35)) requires_order_by_for_offset = is_ms max_bind_params = 999 if is_sq else None return cls( name=name, version=version, paramstyle=d.paramstyle, is_sqlite=is_sq, is_postgres=is_pg, is_mysql=is_my, is_mssql=is_ms, supports_returning=supports_returning, supports_ilike=supports_ilike, requires_order_by_for_offset=requires_order_by_for_offset, max_bind_params=max_bind_params, ) def make_backend_info(engine: Engine) -> BackendInfo: return BackendInfo.from_engine(engine) def ci_like(column, value: str, backend: BackendInfo) -> ClauseElement: """ Portable save-insensitive LIKE. Uses ILIKE where available, else lower() dance. """ pattern = f"%{value}%" if backend.supports_ilike: return column.ilike(pattern) return func.lower(column).like(func.lower(text(":pattern"))).params(pattern=pattern) def apply_pagination(sel: Select, backend: BackendInfo, *, page: int, per_page: int, default_order_by=None) -> Select: """ Portable pagination. MSSQL requires ORDER BY when using OFFSET """ page = max(1, int(page)) per_page = max(1, int(per_page)) offset = (page - 1) * per_page 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 IDENTITY PK when you need to insert explicit IDs. No-op elsewhere. """ if not backend.is_mssql: yield return full_name = f"{table.schema}.{table.name}" if table.schema else table.name session.execute(text(f"SET IDENTITY_INSERT {full_name} ON")) try: yield finally: session.execute(text(f"SET IDENTITY_INSERT {full_name} OFF")) def chunked_in(column, values: Iterable, backend: BackendInfo, chunk_size: Optional[int] = None) -> ClauseElement: """ Build a safe large IN() filter respecting bind param limits. Returns a disjunction of chunked IN clauses if needed. """ vals = list(values) if not vals: return text("1=0") limit = chunk_size or backend.max_bind_params or len(vals) if len(vals) <= limit: return column.in_(vals) parts = [] for i in range(0, len(vals), limit): parts.append(column.in_(vals[i:i + limit])) expr = parts[0] 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)