ad/molodetz/database/soft_delete.py

# retoor <retoor@molodetz.nl>
from molodetz.constants import SOFT_DELETE_TABLES
from molodetz.database.core import db, in_clause, now_iso


def _check(table_name):
    if table_name not in SOFT_DELETE_TABLES:
        raise ValueError(f"{table_name} is not soft-deletable")


def soft_delete(table_name, actor, stamp=None, **criteria):
    _check(table_name)
    if table_name not in db.tables:
        return 0
    stamp = stamp or now_iso()
    table = db[table_name]
    count = 0
    for row in list(table.find(deleted_at=None, **criteria)):
        table.update({"id": row["id"], "deleted_at": stamp, "deleted_by": actor}, ["id"])
        count += 1
    return count


def soft_delete_in(table_name, column, values, actor, stamp=None):
    _check(table_name)
    if not values or table_name not in db.tables:
        return 0
    stamp = stamp or now_iso()
    placeholders, params = in_clause(list(values))
    with db:
        db.query(
            f"UPDATE {table_name} SET deleted_at = :stamp, deleted_by = :actor "
            f"WHERE {column} IN ({placeholders}) AND deleted_at IS NULL",
            stamp=stamp,
            actor=actor,
            **params,
        )
    return len(values)


def restore(table_name, uid):
    _check(table_name)
    if table_name not in db.tables:
        return False
    row = db[table_name].find_one(uid=uid)
    if not row or row.get("deleted_at") is None:
        return False
    db[table_name].update({"id": row["id"], "deleted_at": None, "deleted_by": None}, ["id"])
    return True


def purge(table_name, uid):
    _check(table_name)
    if table_name not in db.tables:
        return False
    return bool(db[table_name].delete(uid=uid))


def list_deleted(table_name, limit=200):
    _check(table_name)
    if table_name not in db.tables:
        return []
    return list(
        db.query(
            f"SELECT * FROM {table_name} WHERE deleted_at IS NOT NULL ORDER BY deleted_at DESC LIMIT :limit",
            limit=limit,
        )
    )


def count_deleted(table_name):
    _check(table_name)
    if table_name not in db.tables:
        return 0
    row = next(iter(db.query(f"SELECT COUNT(*) AS n FROM {table_name} WHERE deleted_at IS NOT NULL")), None)
    return int(row["n"]) if row else 0


def restore_event(stamp):
    restored = []
    tables = set(db.tables)
    for table_name in SOFT_DELETE_TABLES:
        if table_name not in tables:
            continue
        with db:
            db.query(
                f"UPDATE {table_name} SET deleted_at = NULL, deleted_by = NULL WHERE deleted_at = :stamp",
                stamp=stamp,
            )
        restored.append(table_name)
    return restored


def purge_event(stamp):
    purged = []
    tables = set(db.tables)
    for table_name in SOFT_DELETE_TABLES:
        if table_name not in tables:
            continue
        rows = list(db[table_name].find(deleted_at=stamp))
        if rows:
            db[table_name].delete(deleted_at=stamp)
            purged.append((table_name, rows))
    return purged