Db tools

def bulk_upsert_image_cache_fast(
    *,
    list_upsert_objects: list[dict[str, Any]],
    db: Session,
) -> None:
    """
    Insert or update many objects into `HistoryImageCache` and commit

    This function is an optimized version of

    ```python
    for obj in list_upsert_objects:
        db.merge(**obj)
    db.commit()
    ```

    See docs at
    https://docs.sqlalchemy.org/en/20/dialects/postgresql.html#insert-on-conflict-upsert

    NOTE: we tried to replace `index_elements` with
    `constraint="pk_historyimagecache"`, but it did not work as expected.

    Args:
        list_upsert_objects:
            List of dictionaries for objects to be upsert-ed.
        db: A sync database session
    """
    len_list_upsert_objects = len(list_upsert_objects)

    logger.debug(f"[bulk_upsert_image_cache_fast] {len_list_upsert_objects=}.")

    if len_list_upsert_objects == 0:
        return None

    for ind in range(0, len_list_upsert_objects, _CHUNK_SIZE):
        stmt = pg_insert(HistoryImageCache).values(
            list_upsert_objects[ind : ind + _CHUNK_SIZE]
        )
        stmt = stmt.on_conflict_do_update(
            index_elements=[
                HistoryImageCache.zarr_url,
                HistoryImageCache.dataset_id,
                HistoryImageCache.workflowtask_id,
            ],
            set_=dict(
                latest_history_unit_id=stmt.excluded.latest_history_unit_id
            ),
        )
        db.execute(stmt)
        db.commit()

def update_executor_error_log_safe(
    *,
    job_id: int,
    executor_error_log: str | None,
    db: Session,
) -> None:
    """
    Update `JobV2.executor_error_log` with a `DataError` fallback.
    """
    job_db = db.get_one(JobV2, job_id)
    job_db.executor_error_log = executor_error_log
    try:
        db.merge(job_db)
        db.commit()
    except DataError as exc:
        logger.warning(
            f"Cannot update `executor_error_log` for job {job_id}, due to {exc}"
        )
        db.rollback()