sqlite_fk.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310
  1. """Selected ON DELETE rules of the foreign keys, applied by hand on SQLite.
  2. The SQLite database runs with ``PRAGMA foreign_keys`` off: turning it on breaks
  3. the migrations' table rebuilds (measured: every test errors with "no such table:
  4. main.print_archives"). So a delete there applies none of the models' ON DELETE
  5. CASCADE / SET NULL rules. Rows pointing at the deleted row keep its id, and
  6. SQLite hands the highest freed id to the next row created, which then inherits
  7. them: a queue item prints another archive, a project shows another project's
  8. files.
  9. Only the references in APPLIED are handled: those where a dangling id does harm
  10. and the model's rule is what the app wants. Not every model rule is (deleting a
  11. printer would make a printer-scoped notification provider global, and delete
  12. finished queue history), so a reference is added here deliberately, never by
  13. default.
  14. Hooks apply them at every ORM delete, so no delete path needs to remember:
  15. - ``after_delete`` on every mapped class collects the ids of ``db.delete(obj)``,
  16. and ``after_flush`` applies the rules once per flush, one statement per rule;
  17. - ``do_orm_execute`` applies them before ``db.execute(delete(Model)...)``.
  18. Rule statements go straight to the connection and don't trigger the hooks again.
  19. ``sweep_orphans`` repairs what earlier deletes left behind; init_db runs it at
  20. every start. Nothing here does anything on PostgreSQL, which enforces the rules
  21. itself.
  22. """
  23. import logging
  24. from functools import cache
  25. from typing import Any
  26. from sqlalchemy import Column, Connection, Delete, ForeignKey, Table, delete, event, exists, select, text, update
  27. from sqlalchemy.orm import Mapper, ORMExecuteState, Session
  28. from backend.app.core.database import Base
  29. logger = logging.getLogger(__name__)
  30. # (child table, child column) whose ON DELETE rule is applied on SQLite.
  31. APPLIED: frozenset[tuple[str, str]] = frozenset(
  32. {
  33. # print_archives
  34. ("active_print_spoolman", "archive_id"),
  35. ("budget_reservations", "print_archive_id"),
  36. ("library_folders", "archive_id"),
  37. ("pending_uploads", "archived_id"),
  38. ("pipeline_runs", "source_archive_id"),
  39. ("print_batches", "archive_id"),
  40. ("print_log_entries", "archive_id"),
  41. ("print_queue", "archive_id"),
  42. ("project_bom_items", "archive_id"),
  43. ("spool_usage_history", "archive_id"),
  44. ("telegram_pending_verdicts", "archive_id"),
  45. ("wallet_transactions", "print_archive_id"),
  46. # library_files. print_queue.library_file_id is CASCADE in the model, but
  47. # library deletes run release_queue_references first, which keeps the rows
  48. # (finished plates count towards a batch); the rule then finds nothing.
  49. ("library_file_tags", "file_id"),
  50. ("pipeline_runs", "sliced_library_file_id"),
  51. ("pipeline_runs", "source_library_file_id"),
  52. ("print_archives", "library_file_id"),
  53. ("print_batches", "library_file_id"),
  54. ("print_queue", "library_file_id"),
  55. ("print_queue_variants", "library_file_id"),
  56. # print_queue
  57. ("pipeline_jobs", "queue_entry_id"),
  58. ("print_log_entries", "queue_item_id"),
  59. ("print_queue_variants", "queue_item_id"),
  60. ("wallet_transactions", "print_queue_id"),
  61. # print_batches
  62. ("print_batch_plates", "batch_id"),
  63. ("print_queue", "batch_id"),
  64. # projects
  65. ("library_files", "project_id"),
  66. ("library_folders", "project_id"),
  67. ("pending_uploads", "project_id"),
  68. ("print_archives", "project_id"),
  69. ("print_batches", "project_id"),
  70. ("print_queue", "project_id"),
  71. ("project_bom_items", "project_id"),
  72. # printers: what the next printer given the id would otherwise take
  73. # over. Only these: providers, plugs and virtual printers go with the
  74. # printer in its delete route, and the rest is decided separately. The
  75. # print being tracked would be restored for a new printer that is already
  76. # printing, charging its filament to the deleted printer's spools.
  77. ("active_print_sessions", "printer_id"),
  78. ("active_print_spoolman", "printer_id"),
  79. ("pipeline_jobs", "assigned_printer_id"),
  80. ("slicer_pipelines", "target_printer_id"),
  81. # smart_plugs: a printer delete now deletes its plugs; their energy
  82. # history would pass to the next plug given the id.
  83. ("smart_plug_energy_snapshots", "plug_id"),
  84. # users: everything a user owns goes with them or loses its owner, or a
  85. # new user given the freed id inherits it (wallet, cost centres, private
  86. # folders). Without an owner, folders and private cost centres are left
  87. # to admins, not opened to everyone.
  88. ("announcement_reads", "user_id"),
  89. ("api_keys", "user_id"),
  90. ("connected_app_grants", "user_id"),
  91. ("connected_apps", "created_by_id"),
  92. ("cost_center_members", "user_id"),
  93. ("cost_centers", "owner_user_id"),
  94. ("file_variant_groups", "created_by_id"),
  95. ("library_files", "created_by_id"),
  96. ("library_folders", "created_by_id"),
  97. ("long_lived_tokens", "user_id"),
  98. ("pipeline_runs", "created_by"),
  99. ("print_archives", "created_by_id"),
  100. ("print_batches", "created_by_id"),
  101. ("print_log_entries", "created_by_id"),
  102. ("print_queue", "created_by_id"),
  103. ("scheduled_dryings", "created_by_id"),
  104. ("slicer_pipelines", "created_by"),
  105. ("sponsor_toast_state", "user_id"),
  106. ("user_email_preferences", "user_id"),
  107. ("user_groups", "user_id"),
  108. ("user_oidc_links", "user_id"),
  109. ("user_otp_codes", "user_id"),
  110. ("user_totp", "user_id"),
  111. ("user_wallets", "user_id"),
  112. ("wallet_transactions", "created_by_user_id"),
  113. ("wallet_transactions", "user_id"),
  114. }
  115. )
  116. _PENDING = "bambuddy_sqlite_fk_pending"
  117. _TABLES = "bambuddy_sqlite_fk_tables"
  118. # Keeps any one statement's IN list small.
  119. _CHUNK = 500
  120. _MAX_DEPTH = 32
  121. def _applies(connection: Connection) -> bool:
  122. """Only SQLite: PostgreSQL enforces the rules itself."""
  123. return connection.dialect.name == "sqlite"
  124. def _existing_tables(connection: Connection) -> set[str]:
  125. """Tables the database has (cached on the connection); a model without one has no rows to fix."""
  126. tables = connection.info.get(_TABLES)
  127. if tables is None:
  128. result = connection.execute(text("SELECT name FROM sqlite_master WHERE type = 'table'"))
  129. tables = connection.info[_TABLES] = {row[0] for row in result}
  130. return tables
  131. def _single_primary_key(table: Table) -> Column | None:
  132. columns = list(table.primary_key.columns)
  133. return columns[0] if len(columns) == 1 else None
  134. @cache
  135. def _applied_foreign_keys() -> tuple[ForeignKey, ...]:
  136. """The model foreign keys of APPLIED references to a single-column primary key."""
  137. from backend.app.core.database import load_all_models
  138. found = []
  139. for table in load_all_models().sorted_tables:
  140. for fk in table.foreign_keys:
  141. if (
  142. (table.name, fk.parent.name) in APPLIED
  143. and fk.ondelete in ("CASCADE", "SET NULL")
  144. and isinstance(fk.column.table, Table)
  145. and fk.column is _single_primary_key(fk.column.table)
  146. ):
  147. found.append(fk)
  148. return tuple(found)
  149. @cache
  150. def _rules(parent_table: str) -> tuple[tuple[Table, Column, str], ...]:
  151. """(child table, child column, ON DELETE action) of the APPLIED references to *parent_table*."""
  152. return tuple(
  153. (fk.parent.table, fk.parent, str(fk.ondelete))
  154. for fk in _applied_foreign_keys()
  155. if fk.column.table.name == parent_table and isinstance(fk.parent.table, Table)
  156. )
  157. def _apply(connection: Connection, parent_table: str, ids: Any, depth: int = 0) -> None:
  158. """Apply the rules for rows of *parent_table* whose primary key is in *ids* (a list or a select)."""
  159. rules = _rules(parent_table)
  160. if not rules or depth > _MAX_DEPTH:
  161. return
  162. if isinstance(ids, list):
  163. for start in range(0, len(ids), _CHUNK):
  164. _apply_chunk(connection, rules, ids[start : start + _CHUNK], depth)
  165. else:
  166. _apply_chunk(connection, rules, ids, depth)
  167. def _apply_chunk(connection: Connection, rules: tuple, ids: Any, depth: int) -> None:
  168. tables = _existing_tables(connection)
  169. for child, column, action in rules:
  170. if child.name not in tables:
  171. continue
  172. if action == "SET NULL":
  173. connection.execute(update(child).where(column.in_(ids)).values({column.name: None}))
  174. continue
  175. child_key = _single_primary_key(child)
  176. if child_key is not None:
  177. # By depth, not by table: a table that refers to itself recurses too.
  178. _apply(connection, child.name, select(child_key).where(column.in_(ids)), depth + 1)
  179. connection.execute(delete(child).where(column.in_(ids)))
  180. @event.listens_for(Base, "after_delete", propagate=True)
  181. def _collect_orm_delete(mapper: Mapper, connection: Connection, target: Any) -> None:
  182. table = mapper.local_table
  183. session = Session.object_session(target)
  184. if session is None or not isinstance(table, Table) or not _rules(table.name) or not _applies(connection):
  185. return
  186. pending = session.info.setdefault(_PENDING, {})
  187. pending.setdefault(table.name, []).append(mapper.primary_key_from_instance(target)[0])
  188. @event.listens_for(Session, "after_flush")
  189. def _apply_orm_deletes(session: Session, flush_context: Any) -> None:
  190. pending = session.info.pop(_PENDING, None)
  191. if not pending:
  192. return
  193. connection = session.connection()
  194. # The parent rows are already gone; with foreign keys off that doesn't stop
  195. # the rules from finding their dependents.
  196. for table, ids in pending.items():
  197. _apply(connection, table, ids)
  198. @event.listens_for(Session, "after_soft_rollback")
  199. def _forget_pending(session: Session, previous_transaction: Any) -> None:
  200. # A flush that failed after collecting must not leave ids for the next one.
  201. session.info.pop(_PENDING, None)
  202. @event.listens_for(Session, "do_orm_execute")
  203. def _before_bulk_delete(state: ORMExecuteState) -> None:
  204. statement = state.statement
  205. if not state.is_delete or not isinstance(statement, Delete):
  206. return
  207. table = statement.table
  208. if not isinstance(table, Table) or not _rules(table.name):
  209. return
  210. key = _single_primary_key(table)
  211. if key is None:
  212. return
  213. connection = state.session.connection()
  214. if not _applies(connection):
  215. return
  216. # The session would flush pending changes only after this hook; flush now so
  217. # the rules see them (a queue item just moved off the archive must not go).
  218. if state.session.autoflush:
  219. state.session.flush()
  220. where = statement.whereclause
  221. _apply(connection, table.name, select(key).where(where) if where is not None else select(key))
  222. def _still_not_null(connection: Connection, column: Column) -> bool:
  223. """Whether the column is NOT NULL in the database, whatever the model says."""
  224. rows = connection.execute(text(f'PRAGMA table_info("{column.table.name}")'))
  225. return any(row[1] == column.name and row[3] for row in rows)
  226. async def sweep_orphans(conn: Any) -> None:
  227. """Repair APPLIED references to rows that no longer exist (SQLite only).
  228. A SET NULL reference to a missing row is cleared and a CASCADE row is
  229. deleted, which is what the rule would have done at the delete. Repeated
  230. until a pass changes nothing, since a deleted row can leave its own
  231. dependents behind.
  232. """
  233. def run(sync: Any) -> None:
  234. connection = sync.connection() if isinstance(sync, Session) else sync
  235. if not _applies(connection):
  236. return
  237. tables = _existing_tables(connection)
  238. rules = [
  239. fk for fk in _applied_foreign_keys() if fk.parent.table.name in tables and fk.column.table.name in tables
  240. ]
  241. refused = {fk for fk in rules if fk.ondelete == "SET NULL" and _still_not_null(connection, fk.parent)}
  242. for fk in refused:
  243. # A migration that would have dropped it failed (and said so); clearing
  244. # the column would fail too, and take the start of the app with it.
  245. logger.warning("Not clearing %s.%s: the column is still NOT NULL", fk.parent.table.name, fk.parent.name)
  246. rules = [fk for fk in rules if fk not in refused]
  247. for _ in range(10):
  248. changed = 0
  249. for fk in rules:
  250. child, column, referenced = fk.parent.table, fk.parent, fk.column
  251. missing = column.is_not(None) & ~exists(
  252. select(1).select_from(referenced.table).where(referenced == column)
  253. )
  254. if fk.ondelete == "SET NULL":
  255. result = connection.execute(update(child).where(missing).values({column.name: None}))
  256. else:
  257. result = connection.execute(delete(child).where(missing))
  258. if result.rowcount:
  259. changed += result.rowcount
  260. logger.info(
  261. "%s %d %s row(s) whose %s no longer exists",
  262. "Cleared" if fk.ondelete == "SET NULL" else "Deleted",
  263. result.rowcount,
  264. child.name,
  265. referenced.table.name,
  266. )
  267. if not changed:
  268. return
  269. await conn.run_sync(run)