105 lines
3.0 KiB
Python
105 lines
3.0 KiB
Python
# 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
|