|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import text
|
|
|
|
from .core import db
|
|
|
|
|
|
def conditional_update_row(
|
|
table_name: str, row_uid: str, set_clause: str, where_clause: str, params: dict
|
|
) -> int:
|
|
sql = (
|
|
f"UPDATE {table_name} SET {set_clause}, updated_at = :updated_at "
|
|
f"WHERE uid = :row_uid AND ({where_clause})"
|
|
)
|
|
bind = {
|
|
**params,
|
|
"updated_at": datetime.now(timezone.utc).isoformat(),
|
|
"row_uid": row_uid,
|
|
}
|
|
with db:
|
|
result = db.executable.execute(text(sql), bind)
|
|
return result.rowcount
|