sqlmodel_crud_utils

620def bulk_update_rows(
621    id_str_list: list[str | int],
622    data: dict,
623    session_inst: Session,
624    model: type[SQLModel],
625    pk_field: str = "id",
626    chunk_size: int = 500,
627) -> tuple[bool, int]:
628    """
629    Applies the same column values to every row whose primary key is within
630    the provided list and commits once. IDs with no matching row are ignored.
631
632    This is a Core ``UPDATE``, so ORM-level hooks and change tracking
633    (ADR-0011) do not see it. Column-level ``onupdate`` defaults do fire, so
634    ``updated_at`` is refreshed for both ``TimestampMixin`` and
635    ``AuditMixin``. Nothing else is filled in: ``AuditMixin.updated_by`` is
636    only set when the caller includes it in ``data``, and
637    ``SoftDeleteMixin`` fields (``is_deleted``, ``deleted_at``,
638    ``deleted_by``) are only changed if named in ``data``. The ID list is
639    split into ``chunk_size`` batches. One ``UPDATE ... WHERE pk IN (...)``
640    statement is sent per batch (never per row), and all batches share one
641    transaction.
642
643    :param id_str_list: List of primary key values to update.
644    :param data: Mapping of column name to the new value for every row. Must
645        be non-empty and may not name a primary-key column.
646    :param session_inst: SQLAlchemy Session instance.
647    :param model: SQLModel class representing the table.
648    :param pk_field: Primary-key column to match against (default: "id").
649    :param chunk_size: Maximum IDs per UPDATE statement (default: 500).
650    :return: Tuple[bool, int]: ``True`` and the number of rows matched by the
651        update. The count is 0 when nothing matched or the list was empty.
652    :raises ValueError: If ``pk_field`` is not a primary-key column,
653        ``data`` is empty or names an unknown or primary-key column, or
654        ``chunk_size`` is not positive.
655    :raises Exception: Any database error, after the transaction has been
656        rolled back and the error logged.
657    """
658    validate_primary_key_field(model, pk_field)
659    validate_update_columns(model, data)
660    updated = 0
661    try:
662        for chunk in chunked(id_str_list, chunk_size):
663            stmnt = (
664                update(model)
665                .where(getattr(model, pk_field).in_(chunk))
666                .values(**data)
667            )
668            result = session_inst.exec(stmnt)
669            updated += max(result.rowcount, 0)
670        session_inst.commit()
671    except Exception as e:
672        session_inst.rollback()
673        logger.error(f"Failed to bulk update rows: {type(e)}, {e}")
674        raise
675
676    invalidate_written(session_inst, model)
677    return True, updated

718def bulk_upsert_mappings(
719    payload: list,
720    session_inst: Session,
721    model: type[SQLModel],
722    pk_fields: list[str] | None = None,
723    update_columns: Sequence[str] | None = None,
724):
725    """
726    Insert or update a batch of rows in a single ``INSERT ... ON CONFLICT DO
727    UPDATE ... RETURNING`` statement.
728
729    The ``RETURNING`` rows are fully read (``.all()``) before ``commit()``.
730    SQLite refuses to commit while a result cursor is still unread
731    ("cannot commit transaction - SQL statements in progress"); Postgres
732    tolerates it. The upsert is also executed exactly once rather than once
733    bare and once more to fetch the rows.
734
735    Returned instances are expired by ``commit()`` when the session uses
736    ``expire_on_commit=True`` (the default), so read their attributes while
737    the session is open or create the session with ``expire_on_commit=False``.
738
739    :param payload: Row mappings to upsert; the first mapping's keys decide
740        which columns are overwritten on conflict.
741    :param session_inst: Active session.
742    :param model: Table model being upserted.
743    :param pk_fields: Conflict-target columns. Defaults to ``["id"]``.
744    :param update_columns: Restrict the columns written on conflict to
745        these (the ``SET`` clause of ``ON CONFLICT DO UPDATE``, ADR-0016).
746        Defaults to ``None``: every column in the first mapping is
747        overwritten. New rows are always inserted with every provided
748        column. Names are table column keys, as in ``payload``; duplicates
749        are ignored.
750    :return: ``(True, rows)`` where ``rows`` are the upserted model instances.
751    :raises ValueError: Before any statement is sent, if ``update_columns``
752        is empty or a bare string, names an unknown column, a primary-key
753        column or a ``pk_fields`` column, or names a column the first
754        mapping of ``payload`` does not provide.
755    """
756    if not pk_fields:
757        pk_fields = ["id"]
758    if update_columns is not None:
759        update_columns = validate_upsert_update_columns(
760            model, payload, pk_fields, update_columns
761        )
762    set_columns = (
763        payload[0].keys() if update_columns is None else update_columns
764    )
765    stmnt = upsert(model).values(payload)
766    stmnt = stmnt.on_conflict_do_update(
767        index_elements=[getattr(model, x) for x in pk_fields],
768        set_={k: getattr(stmnt.excluded, k) for k in set_columns},
769    )
770    results = (
771        session_inst.scalars(
772            stmnt.returning(model),
773            execution_options={"populate_existing": True},
774        )
775    ).all()
776
777    session_inst.commit()
778    invalidate_written(session_inst, model)
779
780    return True, results

680def delete_row(
681    id_str: str | int,
682    session_inst: Session,
683    model: type[SQLModel],
684    pk_field: str = "id",
685):
686    """
687
688    :param id_str:
689    :param session_inst:
690    :param model:
691    :param pk_field:
692    :return:
693    """
694    success = False
695    stmnt = select(model).where(getattr(model, pk_field) == id_str)
696    results = session_inst.exec(stmnt)
697
698    row = results.one_or_none()
699
700    if not row:
701        pass
702    else:
703        try:
704            session_inst.delete(row)
705            session_inst.commit()
706            invalidate_written(session_inst, model)
707            success = True
708        except Exception as e:
709            logger.error(
710                f"Failed to delete data row. Please see error messages here: "
711                f"{type(e), e, e.args}"
712            )
713            session_inst.rollback()
714
715    return success

542def delete_rows_within_id_list(
543    id_str_list: list[str | int],
544    session_inst: Session,
545    model: type[SQLModel],
546    pk_field: str = "id",
547    chunk_size: int = 500,
548    soft: bool = False,
549    deleted_by: str | None = None,
550) -> tuple[bool, int]:
551    """
552    Deletes every row whose primary key is within the provided list and
553    commits once. IDs with no matching row are ignored.
554
555    By default this is a hard SQL ``DELETE``: rows are removed, even for a
556    model that uses ``SoftDeleteMixin`` (nothing is marked with
557    ``deleted_at``), and ORM-level hooks do not run. Pass ``soft=True`` to
558    mark the rows deleted instead, exactly as ``SoftDeleteMixin.soft_delete``
559    does: ``is_deleted`` becomes true, ``deleted_at`` is the current UTC time
560    (one timestamp for the whole call) and ``deleted_by`` is the given user.
561    The soft path is a Core ``UPDATE``, so the same bulk-write rules as
562    ``bulk_update_rows`` apply: column-level ``onupdate`` defaults fire (for
563    example ``updated_at`` of ``TimestampMixin`` and ``AuditMixin``), while
564    ``updated_by`` is not set and ORM events and change tracking do not see
565    the write. Rows already soft-deleted are left untouched (their original
566    ``deleted_at`` and ``deleted_by`` are kept) and are not counted.
567
568    The ID list is split into ``chunk_size`` batches to stay under backend
569    bind-parameter limits. One ``DELETE`` (or ``UPDATE``) ``... WHERE pk IN
570    (...)`` statement is sent per batch (never per row), so 1,200 IDs at the
571    default size is 3 statements. All batches share one transaction, so the
572    call is all-or-nothing. The model's cache entries are invalidated after
573    the commit in both modes.
574
575    :param id_str_list: List of primary key values to delete.
576    :param session_inst: SQLAlchemy Session instance.
577    :param model: SQLModel class representing the table.
578    :param pk_field: Primary-key column to match against (default: "id").
579    :param chunk_size: Maximum IDs per statement (default: 500).
580    :param soft: When ``True``, soft-delete instead of removing the rows.
581        The model must have the ``SoftDeleteMixin`` columns ``is_deleted``,
582        ``deleted_at`` and ``deleted_by`` (default: ``False``).
583    :param deleted_by: User recorded in ``deleted_by`` on soft-deleted rows.
584        Only valid with ``soft=True`` (default: ``None``).
585    :return: Tuple[bool, int]: ``True`` and the number of rows affected: rows
586        removed for a hard delete, or rows newly marked deleted for a soft
587        delete (rows that were already soft-deleted are not counted). The
588        count is 0 when nothing matched or the list was empty.
589    :raises ValueError: If ``pk_field`` is not a primary-key column of
590        ``model``, ``chunk_size`` is not positive, ``soft=True`` is used on a
591        model without the ``SoftDeleteMixin`` columns (the message names the
592        missing ones), or ``deleted_by`` is given without ``soft=True``.
593    :raises Exception: Any database error, after the transaction has been
594        rolled back and the error logged.
595    """
596    validate_primary_key_field(model, pk_field)
597    validate_soft_delete_args(model, soft, deleted_by)
598    deleted_at = soft_delete_timestamp() if soft else None
599    deleted = 0
600    try:
601        for chunk in chunked(id_str_list, chunk_size):
602            if soft:
603                stmnt = soft_delete_statement(
604                    model, pk_field, chunk, deleted_at, deleted_by
605                )
606            else:
607                stmnt = delete(model).where(getattr(model, pk_field).in_(chunk))
608            result = session_inst.exec(stmnt)
609            deleted += max(result.rowcount, 0)
610        session_inst.commit()
611    except Exception as e:
612        session_inst.rollback()
613        logger.error(f"Failed to bulk delete rows: {type(e)}, {e}")
614        raise
615
616    invalidate_written(session_inst, model)
617    return True, deleted

823def get_change_history(
824    session_inst: Session,
825    model: type[SQLModel],
826    id_str: int | str,
827):
828    """
829    Return the recorded change history of one row, oldest first.
830
831    Requires ``register_change_tracking()`` and a model inheriting
832    ``TrackChangesMixin`` (ADR-0011). Changes made by the bulk helpers are not
833    recorded, so they do not appear here.
834
835    :param session_inst: Session
836    :param model: type[SQLModel]
837    :param id_str: primary key value of the row; a tuple or list in
838        column order for a composite key
839    :return: Tuple[bool, list[dict]]
840    """
841    stmnt = history_select(model, id_str)
842    try:
843        result = session_inst.connection(
844            bind_arguments={"mapper": model}
845        ).execute(stmnt)
846        return True, [dict(r) for r in result.mappings().all()]
847    except Exception as e:
848        logger.error(
849            f"Reading change history failed. See error message: "
850            f"{type(e), e, e.args}"
851        )
852        session_inst.rollback()
853        return False, []

 68def get_one_or_create(
 69    session_inst: Session,
 70    model: type[SQLModel],
 71    create_method_kwargs: dict | None = None,
 72    selectin: bool = False,
 73    select_in_key: str | None = None,
 74    **kwargs,
 75):
 76    """
 77    This function either returns an existing data row from the database or
 78    creates a new instance and saves it to the DB.
 79
 80    :param session_inst: Session
 81    :param model: SQLModel ORM
 82    :param create_method_kwargs: dict
 83    :param selectin: bool
 84    :param select_in_key: str | None
 85    :param kwargs: keyword args
 86    :return: Tuple[Row, bool]
 87    """
 88
 89    def _get_entry(sqlmodel, **key_args):
 90        stmnt = select(sqlmodel).filter_by(**key_args)
 91        results = get_result_from_query(query=stmnt, session=session_inst)
 92
 93        if results:
 94            if selectin and select_in_key:
 95                stmnt = stmnt.options(
 96                    selectinload(getattr(sqlmodel, select_in_key))
 97                )
 98                results = get_result_from_query(
 99                    query=stmnt, session=session_inst
100                )
101            return results, True
102        else:
103            return results, False
104
105    results, exists = _get_entry(model, **kwargs)
106    if results:
107        return results, exists
108    else:
109        kwargs.update(create_method_kwargs or {})
110        created = model()
111        [setattr(created, k, v) for k, v in kwargs.items()]
112        session_inst.add(created)
113        session_inst.commit()
114        invalidate_written(session_inst, model)
115        return created, False

47def get_result_from_query(query: SelectOfScalar, session: Session):
48    """
49    Processes an SQLModel query object and returns a singular result from the
50    return payload. If more than one row is returned, then only the first row is
51    returned. If no rows are available, then a null value is returned.
52
53    :param query: SelectOfScalar
54    :param session: Session
55
56    :return: Row
57    """
58    results = session.exec(query)
59    try:
60        results = results.one_or_none()
61    except MultipleResultsFound:
62        results = session.exec(query)
63        results = results.first()
64
65    return results

183def get_row(
184    id_str: str | int,
185    session_inst: Session,
186    model: type[SQLModel],
187    selectin: bool = False,
188    select_in_keys: list[str] | None = None,
189    lazy: bool = False,
190    lazy_load_keys: list[str] | None = None,
191    pk_field: str = "id",
192    use_cache: bool = False,
193    cache_ttl: float | None = None,
194):
195    """
196    Fetch a single row by primary key.
197
198    The optional parameters follow the same order as ``a_sync.get_row`` and
199    both ``get_rows`` functions, so positional calls bind identically in the
200    sync and async APIs.
201
202    :param id_str: Primary key value to look up.
203    :param session_inst: SQLModel Session instance.
204    :param model: SQLModel class representing the table.
205    :param selectin: Apply ``selectinload`` for ``select_in_keys``.
206    :param select_in_keys: Relationship name(s) to eager load.
207    :param lazy: Apply ``lazyload`` for ``lazy_load_keys``.
208    :param lazy_load_keys: Relationship name(s) to lazy load.
209    :param pk_field: Primary key field name (default: "id").
210    :param use_cache: Read from and fill the cache set by ``configure_cache``
211        (ADR-0012). Bypassed with ``selectin`` or ``lazy``.
212    :param cache_ttl: Seconds the entry lives; backend default when None.
213    :return: Tuple[bool, SQLModel | None]: Success flag and the row, if found.
214    """
215    backend = active_backend(
216        use_cache
217        and not (selectin and select_in_keys)
218        and not (lazy and lazy_load_keys)
219    )
220    if backend:
221        namespace = namespace_for(model)
222        cache_key = make_key(
223            session_inst, model, "get_row", pk_field=pk_field, id=id_str
224        )
225        hit = lookup(backend, namespace, cache_key)
226        row = load_row(model, hit) if hit is not None else None
227        if row is not None:
228            return True, row
229    stmnt = select(model).where(getattr(model, pk_field) == id_str)
230    if selectin and select_in_keys:
231        if isinstance(select_in_keys, list) is False:
232            select_in_keys = [select_in_keys]
233
234        for key in select_in_keys:
235            stmnt = stmnt.options(selectinload(getattr(model, key)))
236    if lazy and lazy_load_keys:
237        if isinstance(lazy_load_keys, list) is False:
238            lazy_load_keys = [lazy_load_keys]
239        for key in lazy_load_keys:
240            stmnt = stmnt.options(lazyload(getattr(model, key)))
241    results = session_inst.exec(stmnt)
242
243    row = results.one_or_none()
244
245    if not row:
246        success = False
247    else:
248        success = True
249        if backend:
250            store(backend, namespace, cache_key, dump_row(row), cache_ttl)
251
252    return success, row

255def get_rows(
256    session_inst: Session,
257    model: type[SQLModel],
258    selectin: bool = False,
259    select_in_keys: list[str] | None = None,
260    lazy: bool = False,
261    lazy_load_keys: list[str] | None = None,
262    page_size: int = 100,
263    page: int = 1,
264    text_field: str | None = None,
265    stmnt: SelectOfScalar | None = None,
266    use_cache: bool = False,
267    cache_ttl: float | None = None,
268    **kwargs,
269):
270    """
271    Retrieve rows for ``model``, building the query from ``kwargs`` or
272    running a caller-supplied statement.
273
274    Pagination applies only to statements built internally (``stmnt`` is
275    ``None``): ``page`` and ``page_size`` translate to OFFSET and LIMIT. A
276    caller-supplied ``stmnt`` is executed exactly as given; the caller owns
277    its pagination, so ``page``, ``page_size`` and any filter, sort or
278    loading arguments are ignored. Add ``.limit()``/``.offset()`` to the
279    statement yourself when you want a bounded result.
280
281    .. versionchanged:: Unreleased
282        A custom ``stmnt`` is no longer silently capped at ``page_size``
283        rows (issue #11).
284
285    :param session_inst: Active session used to execute the query.
286    :param model: SQLModel table class to query.
287    :param selectin: Enable ``selectinload`` for ``select_in_keys``.
288    :param select_in_keys: Relationship names to eager load via selectin.
289    :param lazy: Enable ``lazyload`` for ``lazy_load_keys``.
290    :param lazy_load_keys: Relationship names to lazy load.
291    :param page_size: Rows per page; ignored when ``stmnt`` is supplied.
292    :param page: 1-based page number; ignored when ``stmnt`` is supplied.
293    :param text_field: Field matched with a text search from ``kwargs``.
294    :param stmnt: Custom statement, executed as-is and never paginated.
295    :param use_cache: Read from and fill the cache set by ``configure_cache``
296        (ADR-0012). Bypassed with ``stmnt``, ``selectin`` or ``lazy``.
297    :param cache_ttl: Seconds the entry lives; backend default when None.
298    :param kwargs: Filters and sort options for the built query.
299    :return: ``(success, rows)`` where ``success`` is True if rows exist.
300    """
301    paginate = stmnt is None
302    backend = active_backend(
303        use_cache
304        and stmnt is None
305        and not (selectin and select_in_keys)
306        and not (lazy and lazy_load_keys)
307    )
308    if backend:
309        namespace = namespace_for(model)
310        cache_key = make_key(
311            session_inst,
312            model,
313            "get_rows",
314            page=page,
315            page_size=page_size,
316            text_field=text_field,
317            filters=repr(sorted(kwargs.items(), key=lambda kv: kv[0])),
318        )
319        hit = lookup(backend, namespace, cache_key)
320        rows = load_rows(model, hit) if hit is not None else None
321        if rows is not None:
322            return True, rows
323    # Inside get_rows (sync and async versions)
324
325    # ... existing code ...
326    if stmnt is None:
327        stmnt = select(model)
328        if kwargs:
329            # Separate special filter keys from exact match keys
330            exact_match_kwargs = {}
331            special_filters = {}
332
333            keys_to_process = list(kwargs.keys())  # Iterate over a copy
334
335            for key in keys_to_process:
336                val = kwargs[key]
337                if "__like" in key:
338                    model_key = key.replace("__like", "")
339                    special_filters[key] = (
340                        getattr(model, model_key).like,
341                        f"%{val}%",
342                    )  # Adapt for like
343                elif "__gte" in key:
344                    model_key = key.replace("__gte", "")
345                    parsed_val = (
346                        date_parse(val)
347                        if "date" in key
348                        and isinstance(val, str)
349                        and is_date(val, fuzzy=False)
350                        else (
351                            int(val)
352                            if isinstance(val, str) and val.isdigit()
353                            else val
354                        )
355                    )
356                    special_filters[key] = (
357                        getattr(model, model_key).__ge__,
358                        parsed_val,
359                    )
360                elif "__lte" in key:
361                    model_key = key.replace("__lte", "")
362                    parsed_val = (
363                        date_parse(val)
364                        if "date" in key
365                        and isinstance(val, str)
366                        and is_date(val, fuzzy=False)
367                        else (
368                            int(val)
369                            if isinstance(val, str) and val.isdigit()
370                            else val
371                        )
372                    )
373                    special_filters[key] = (
374                        getattr(model, model_key).__le__,
375                        parsed_val,
376                    )
377                elif "__gt" in key:  # Add __gt if needed
378                    model_key = key.replace("__gt", "")
379                    parsed_val = (
380                        date_parse(val)
381                        if "date" in key
382                        and isinstance(val, str)
383                        and is_date(val, fuzzy=False)
384                        else (
385                            int(val)
386                            if isinstance(val, str) and val.isdigit()
387                            else val
388                        )
389                    )
390                    special_filters[key] = (
391                        getattr(model, model_key).__gt__,
392                        parsed_val,
393                    )
394                elif "__lt" in key:  # Add __lt if needed
395                    model_key = key.replace("__lt", "")
396                    parsed_val = (
397                        date_parse(val)
398                        if "date" in key
399                        and isinstance(val, str)
400                        and is_date(val, fuzzy=False)
401                        else (
402                            int(val)
403                            if isinstance(val, str) and val.isdigit()
404                            else val
405                        )
406                    )
407                    special_filters[key] = (
408                        getattr(model, model_key).__lt__,
409                        parsed_val,
410                    )
411                elif "__in" in key:  # Add __in if needed
412                    model_key = key.replace("__in", "")
413                    if isinstance(val, list):
414                        special_filters[key] = (
415                            getattr(model, model_key).in_,
416                            val,
417                        )
418                    else:
419                        logger.warning(
420                            f"Value for __in filter '{key}' is not a list, "
421                            f"skipping."
422                        )
423                elif key not in ("sort_desc", "sort_field") and (
424                    not text_field or key != text_field
425                ):
426                    # Collect keys for filter_by, excluding sort/text search
427                    # keys
428                    exact_match_kwargs[key] = val
429
430            # Apply special filters using filter()
431            for _filter_key, (
432                filter_method,
433                filter_value,
434            ) in special_filters.items():
435                stmnt = stmnt.filter(filter_method(filter_value))
436
437            # Apply sorting
438            sort_desc = kwargs.get("sort_desc")
439            sort_field = kwargs.get("sort_field")
440            if sort_field:
441                sort_attr = getattr(model, sort_field)
442                stmnt = stmnt.order_by(
443                    sort_attr.desc() if sort_desc else sort_attr
444                )
445
446            # Apply text search if applicable (assuming .match() is correct)
447            if text_field and text_field in kwargs:
448                search_val = kwargs[text_field]
449                stmnt = stmnt.where(
450                    getattr(model, text_field).match(search_val)
451                )
452                # Remove from exact_match_kwargs if it ended up there
453                exact_match_kwargs.pop(text_field, None)
454
455            # Apply exact matches using filter_by()
456            if exact_match_kwargs:
457                stmnt = stmnt.filter_by(**exact_match_kwargs)
458
459        # Apply relationship loading options (Check if key is a relationship
460        # first - simplified check)
461        if selectin and select_in_keys:
462            for key in select_in_keys:
463                # Basic check: Does the attribute exist and is it likely a
464                # relationship?
465                # A more robust check might involve inspecting
466                # model.__sqlmodel_relationships__
467                attr = getattr(model, key, None)
468                if (
469                    attr is not None
470                    and hasattr(attr, "property")
471                    and hasattr(attr.property, "mapper")
472                ):
473                    stmnt = stmnt.options(selectinload(attr))
474                else:
475                    logger.warning(
476                        f"Skipping selectinload for non-relationship "
477                        f"attribute '{key}' on model {model.__name__}"
478                    )
479
480        if lazy and lazy_load_keys:
481            for key in lazy_load_keys:
482                attr = getattr(model, key, None)
483                if (
484                    attr is not None
485                    and hasattr(attr, "property")
486                    and hasattr(attr.property, "mapper")
487                ):
488                    stmnt = stmnt.options(lazyload(attr))
489                else:
490                    logger.warning(
491                        f"Skipping lazyload for non-relationship attribute  "
492                        f"'{key}' on model {model.__name__}"
493                    )
494
495    if paginate:
496        stmnt = stmnt.offset((page - 1) * page_size).limit(page_size)
497
498    _result = session_inst.exec(stmnt)
499    results = _result.all()
500    success = True if len(results) > 0 else False
501    if success and backend:
502        store(
503            backend,
504            namespace,
505            cache_key,
506            dump_rows(results),
507            cache_ttl,
508        )
509
510    return success, results

513def get_rows_within_id_list(
514    id_str_list: list[str | int],
515    session_inst: Session,
516    model: type[SQLModel],
517    pk_field: str = "id",
518):
519    """
520    Retrieves rows from the database whose primary key is within the provided
521    list.
522
523    :param id_str_list: List of primary key values to fetch.
524    :param session_inst: SQLAlchemy Session instance.
525    :param model: SQLModel class representing the table.
526    :param pk_field: Name of the primary key field (default: "id").
527    :return: Tuple[bool, list[SQLModel]]: A tuple containing a success flag
528             (True if rows were found, False otherwise) and a list of the
529             found model instances.
530    """
531    if not id_str_list:  # Handle empty input list
532        return False, []
533
534    stmnt = select(model).where(getattr(model, pk_field).in_(id_str_list))
535    results = session_inst.exec(stmnt).all()  # Fetch all results into a list
536
537    success = len(results) > 0  # Success is true only if results were found
538
539    return success, results

143def insert_data_rows(data_rows, session_inst: Session):
144    """
145
146    :param data_rows:
147    :param session_inst:
148    :return:
149    """
150    try:
151        session_inst.add_all(data_rows)
152        session_inst.commit()
153        invalidate_written(session_inst, *{type(r) for r in data_rows})
154
155        return True, data_rows
156
157    except Exception as e:
158        logger.error(
159            f"Writing data rows to table failed. See error message: "
160            f"{type(e), e, e.args}"
161        )
162        logger.info(
163            "Attempting to write individual entries. This can be a "
164            "bit taxing, so please consider your payload to the DB"
165        )
166
167        session_inst.rollback()
168        processed_rows, failed_rows = [], []
169        for row in data_rows:
170            success, processed_row = write_row(row, session_inst=session_inst)
171            if not success:
172                failed_rows.append(row)
173            else:
174                processed_rows.append(row)
175
176        if processed_rows:
177            status = True
178        else:
179            status = (False,)
180        return status, {"success": processed_rows, "failed": failed_rows}

783def update_row(
784    id_str: int | str,
785    data: dict,
786    session_inst: Session,
787    model: type[SQLModel],
788    pk_field: str = "id",
789):
790    """
791
792    :param id_str:
793    :param data:
794    :param session_inst:
795    :param model:
796    :param pk_field:
797    :return:
798    """
799    success = False
800    stmnt = select(model).where(getattr(model, pk_field) == id_str)
801    results = session_inst.exec(stmnt)
802
803    row = results.one_or_none()
804
805    if row:
806        [setattr(row, k, v) for k, v in data.items()]
807        try:
808            session_inst.add(row)
809            session_inst.commit()
810            invalidate_written(session_inst, model)
811            success = True
812        except Exception as e:
813            session_inst.rollback()
814            logger.error(
815                f"Updating the data row failed. See error messages: "
816                f"{type(e), e, e.args}"
817            )
818        return success, row
819    else:
820        return success, None

118def write_row(data_row: SQLModel, session_inst: Session):
119    """
120    Writes a new instance of an SQLModel ORM model to the database, with an
121    exception catch that rolls back the session in the event of failure.
122
123    :param data_row: SQLModel
124    :param session_inst: Session
125    :return: Tuple[bool, ScalarResult]
126    """
127    try:
128        session_inst.add(data_row)
129        session_inst.commit()
130        invalidate_written(session_inst, type(data_row))
131
132        return True, data_row
133    except Exception as e:
134        session_inst.rollback()
135        logger.error(
136            f"Writing data row to table failed. See error message: "
137            f"{type(e), e, e.args}"
138        )
139
140        return False, None

616async def bulk_update_rows(
617    id_str_list: list[str | int],
618    data: dict,
619    session_inst: AsyncSession,
620    model: type[SQLModel],
621    pk_field: str = "id",
622    chunk_size: int = 500,
623) -> tuple[bool, int]:
624    """
625    Applies the same column values to every row whose primary key is within
626    the provided list and commits once. IDs with no matching row are ignored.
627
628    This is a Core ``UPDATE``, so ORM-level hooks and change tracking
629    (ADR-0011) do not see it. Column-level ``onupdate`` defaults do fire, so
630    ``updated_at`` is refreshed for both ``TimestampMixin`` and
631    ``AuditMixin``. Nothing else is filled in: ``AuditMixin.updated_by`` is
632    only set when the caller includes it in ``data``, and
633    ``SoftDeleteMixin`` fields (``is_deleted``, ``deleted_at``,
634    ``deleted_by``) are only changed if named in ``data``. The ID list is
635    split into ``chunk_size`` batches. One ``UPDATE ... WHERE pk IN (...)``
636    statement is sent per batch (never per row), and all batches share one
637    transaction.
638
639    :param id_str_list: List of primary key values to update.
640    :param data: Mapping of column name to the new value for every row. Must
641        be non-empty and may not name a primary-key column.
642    :param session_inst: SQLModel AsyncSession instance.
643    :param model: SQLModel class representing the table.
644    :param pk_field: Primary-key column to match against (default: "id").
645    :param chunk_size: Maximum IDs per UPDATE statement (default: 500).
646    :return: Tuple[bool, int]: ``True`` and the number of rows matched by the
647        update. The count is 0 when nothing matched or the list was empty.
648    :raises ValueError: If ``pk_field`` is not a primary-key column,
649        ``data`` is empty or names an unknown or primary-key column, or
650        ``chunk_size`` is not positive.
651    :raises Exception: Any database error, after the transaction has been
652        rolled back and the error logged.
653    """
654    validate_primary_key_field(model, pk_field)
655    validate_update_columns(model, data)
656    updated = 0
657    try:
658        for chunk in chunked(id_str_list, chunk_size):
659            stmnt = (
660                update(model)
661                .where(getattr(model, pk_field).in_(chunk))
662                .values(**data)
663            )
664            result = await session_inst.exec(stmnt)
665            updated += max(result.rowcount, 0)
666        await session_inst.commit()
667    except Exception as e:
668        await session_inst.rollback()
669        logger.error(f"Failed to bulk update rows: {type(e)}, {e}")
670        raise
671
672    await a_invalidate_written(session_inst, model)
673    return True, updated

714async def bulk_upsert_mappings(
715    payload: list,
716    session_inst: AsyncSession,
717    model: type[SQLModel],
718    pk_fields: list[str] | None = None,
719    update_columns: Sequence[str] | None = None,
720):
721    """
722    Insert or update a batch of rows in a single ``INSERT ... ON CONFLICT DO
723    UPDATE ... RETURNING`` statement.
724
725    The ``RETURNING`` rows are fully read (``.all()``) before ``commit()``.
726    SQLite refuses to commit while a result cursor is still unread
727    ("cannot commit transaction - SQL statements in progress"); Postgres
728    tolerates it. The upsert is also executed exactly once rather than once
729    bare and once more to fetch the rows.
730
731    Returned instances are expired by ``commit()`` when the session uses
732    ``expire_on_commit=True`` (the default), so read their attributes while
733    the session is open or create the session with ``expire_on_commit=False``.
734
735    :param payload: Row mappings to upsert; the first mapping's keys decide
736        which columns are overwritten on conflict.
737    :param session_inst: Active session.
738    :param model: Table model being upserted.
739    :param pk_fields: Conflict-target columns. Defaults to ``["id"]``.
740    :param update_columns: Restrict the columns written on conflict to
741        these (the ``SET`` clause of ``ON CONFLICT DO UPDATE``, ADR-0016).
742        Defaults to ``None``: every column in the first mapping is
743        overwritten. New rows are always inserted with every provided
744        column. Names are table column keys, as in ``payload``; duplicates
745        are ignored.
746    :return: ``(True, rows)`` where ``rows`` are the upserted model instances.
747    :raises ValueError: Before any statement is sent, if ``update_columns``
748        is empty or a bare string, names an unknown column, a primary-key
749        column or a ``pk_fields`` column, or names a column the first
750        mapping of ``payload`` does not provide.
751    """
752    if not pk_fields:
753        pk_fields = ["id"]
754    if update_columns is not None:
755        update_columns = validate_upsert_update_columns(
756            model, payload, pk_fields, update_columns
757        )
758    set_columns = (
759        payload[0].keys() if update_columns is None else update_columns
760    )
761    stmnt = upsert(model).values(payload)
762    stmnt = stmnt.on_conflict_do_update(
763        index_elements=[getattr(model, x) for x in pk_fields],
764        set_={k: getattr(stmnt.excluded, k) for k in set_columns},
765    )
766    results = (
767        await session_inst.scalars(
768            stmnt.returning(model),
769            execution_options={"populate_existing": True},
770        )
771    ).all()
772
773    await session_inst.commit()
774    await a_invalidate_written(session_inst, model)
775
776    return True, results