Function update_object

backend/app/database_ops/utils.py:11–37

(conn, update_dict: Dict, update_time: bool, table_name: str, equal_filters: Dict)

async def update_object(conn, update_dict: Dict, update_time: bool, table_name: str, equal_filters: Dict) -> None:
    # 2. build update dict
    pg_update_dict = update_dict.copy()
    for key, value in update_dict.items():
        if isinstance(value, (dict, list)):
            pg_update_dict[key] = json.dumps(value)

    timestamp_int = current_timestamp_int_milliseconds()

    # 3. Prepare the SET clause for the update query
    updates = ", ".join(
        f"{key} = ${idx}" for idx, key in enumerate(pg_update_dict.keys(), start=len(equal_filters) + 2)
    )
    if update_time:
        updates += ", updated_timestamp = $1"

    # 4. Prepare the WHERE clause for the update query
    conditions = " AND ".join(f"{key} = ${idx}" for idx, key in enumerate(equal_filters.keys(), start=2))

    # 5. Prepare the final query
    query = f"UPDATE {table_name} SET {updates} WHERE {conditions}"

    # 6. Prepare the arguments for the query
    args = [timestamp_int, *equal_filters.values(), *pg_update_dict.values()]

    # 7. Execute the query
    await conn.execute(query, *args)


async def get_object_total(