shopdb-flask/scripts/site_imports/journal.py

"""Shared, import-agnostic journal + undo for reversible bulk imports.

IMPORT-AGNOSTIC. Nothing here knows about WINC, Maximo, machines, or any one
source. An import writer records the primitive mutations it makes - "set column
X on row Y of table T from old to new" (update) or "insert row Y into table T"
(insert) - each tied to a batch id, and a generic undo replays them in reverse.
That is the whole mechanism: importers emit primitives; undo reverses primitives.

Identity: one value threads approval, write, and undo. The approval packet's
`batchid` becomes the importbatches primary key, becomes the journal foreign key,
becomes the argument the "Undo import" button passes. Nothing to reconcile.

Contract (extends the approval packet's 6-step writer contract):
  open_batch()  -> after the packet passes, write the importbatches row.
  record_update()/record_insert() -> append a journal entry in the SAME
                  transaction as the mutation it describes (ledger rule: the
                  entry and the change it records commit together).
  undo_batch()  -> reverse-replay this batch's entries, conservatively.

Undo is deliberately conservative and never destructive beyond what the import
owns:
  - update reverts a cell ONLY if it still holds the value we wrote; a value a
    human changed since is left as-is and reported as a conflict (never clobbered).
  - insert hard-deletes ONLY the row this batch created, and only when nothing
    outside the batch still references it; otherwise it is left and reported.

All access goes through a SQLAlchemy connection executing parameterised text, so
this reuses the same connection an importer already opened and pulls in no ORM
models - keeping it a true platform primitive any importer can share.
"""

from collections import namedtuple
from datetime import datetime, timezone

BATCH_APPLIED = 'applied'
BATCH_UNDONE = 'undone'
# See shopdb/core/models/importbatch.py for why this exists.
BATCH_PARTIAL = 'partially-undone'

OP_UPDATE = 'update'
OP_INSERT = 'insert'

# Result of an undo pass, for a report and a UI.
UndoResult = namedtuple('UndoResult', ('batchid', 'reverted', 'conflicts'))

# A single thing undo could not safely reverse.
Conflict = namedtuple('Conflict', ('operation', 'table', 'targetid',
                                    'column', 'reason'))


def _utcnow():
    return datetime.now(timezone.utc).replace(tzinfo=None)


def _as_text(value):
    """Store every journalled value as text so a pre/post image compares the
    same way it will read back, regardless of the column's declared type."""
    return None if value is None else str(value)


def open_batch(connection, packet, createdby=None):
    """Write the importbatches row for an approved packet.

    Call once, after the packet has been verified against the current plan and
    acknowledged. Returns the batchid so the caller stamps it on every journal
    entry it then records. Import-agnostic: the packet's own fields carry all
    the identity this needs.
    """
    from sqlalchemy import text

    connection.execute(text(
        'INSERT INTO importbatches '
        '(batchid, kind, planhash, rowcount, createdat, createdby, status) '
        'VALUES (:batchid, :kind, :planhash, :rowcount, :createdat, '
        ':createdby, :status)'),
        {
            'batchid': packet['batchid'],
            'kind': packet.get('kind'),
            'planhash': packet.get('planhash'),
            'rowcount': packet.get('rowcount', 0),
            'createdat': _utcnow(),
            'createdby': createdby,
            'status': BATCH_APPLIED,
        })
    return packet['batchid']


def record_update(connection, batchid, sequence, table, targetid, column,
                  oldvalue, newvalue):
    """Append an 'update' journal entry describing one cell change.

    Record the pre-image (oldvalue) you read BEFORE applying the change and the
    post-image (newvalue) you are about to write, in the same transaction as the
    UPDATE itself.
    """
    _insert_entry(connection, batchid, sequence, OP_UPDATE, table, targetid,
                  column, oldvalue, newvalue)


def record_insert(connection, batchid, sequence, table, targetid):
    """Append an 'insert' journal entry for a row this batch created.

    Only ever emitted for a needs-create row the writer itself inserted, so undo
    can hard-delete strictly what the import created.
    """
    _insert_entry(connection, batchid, sequence, OP_INSERT, table, targetid,
                  None, None, None)


def _insert_entry(connection, batchid, sequence, operation, table, targetid,
                  column, oldvalue, newvalue):
    from sqlalchemy import text

    connection.execute(text(
        'INSERT INTO importjournalentries '
        '(batchid, sequence, operation, targettable, targetid, columnname, '
        'oldvalue, newvalue, isreverted) '
        'VALUES (:batchid, :sequence, :operation, :targettable, :targetid, '
        ':columnname, :oldvalue, :newvalue, 0)'),
        {
            'batchid': batchid,
            'sequence': sequence,
            'operation': operation,
            'targettable': table,
            'targetid': targetid,
            'columnname': column,
            'oldvalue': _as_text(oldvalue),
            'newvalue': _as_text(newvalue),
        })


def undo_batch(connection, batchid, references=None):
    """Reverse-replay an applied batch's journal. Returns an UndoResult.

    references (optional): { table -> [(reftable, refcolumn), ...] } describing
    which columns in other tables point at a created row's id. Used to decide
    whether an inserted row is safe to hard-delete: if any reference OUTSIDE
    this batch's own updates still points at it, the row is left and reported as
    a conflict rather than deleted. Omit it to require that the batch's own
    update-reverts have already cleared every reference (the common case, since a
    created businessunit is only ever pointed at by the same batch's asset
    updates).

    Idempotent: a batch already 'undone' is a no-op. Entries are walked in
    reverse sequence so inserts are removed only after the updates that pointed
    at them are reverted.
    """
    from sqlalchemy import text

    batch = connection.execute(text(
        'SELECT status FROM importbatches WHERE batchid = :batchid'),
        {'batchid': batchid}).fetchone()
    if batch is None:
        raise ValueError('unknown batchid: %s' % batchid)
    # Only a FULLY undone batch is finished. A partially-undone one still has
    # entries with isreverted = 0, and the whole point of retrying is to pick
    # them up once whatever conflicted has been resolved.
    if batch[0] == BATCH_UNDONE:
        return UndoResult(batchid, 0, [])

    entries = connection.execute(text(
        'SELECT entryid, sequence, operation, targettable, targetid, '
        'columnname, oldvalue, newvalue FROM importjournalentries '
        'WHERE batchid = :batchid AND isreverted = 0 '
        'ORDER BY sequence DESC, entryid DESC'),
        {'batchid': batchid}).fetchall()

    reverted = 0
    conflicts = []
    for row in entries:
        (entryid, _sequence, operation, table, targetid, column,
         oldvalue, newvalue) = row

        if operation == OP_UPDATE:
            conflict = _revert_update(connection, table, targetid, column,
                                      oldvalue, newvalue)
        elif operation == OP_INSERT:
            conflict = _revert_insert(connection, batchid, table, targetid,
                                      references)
        else:
            conflict = Conflict(operation, table, targetid, column,
                                'unknown operation')

        if conflict is None:
            connection.execute(text(
                'UPDATE importjournalentries SET isreverted = 1 '
                'WHERE entryid = :entryid'), {'entryid': entryid})
            reverted += 1
        else:
            conflicts.append(conflict)

    # Record what actually happened, not what was attempted. Writing 'undone'
    # unconditionally is what made a conflicted entry unrevertable: this call
    # is the only way back, and the early return above refuses a second attempt
    # on an undone batch.
    #
    # undoneat is stamped only when the batch is genuinely finished; on a
    # partial it stays null, so "when was this reversed" keeps its meaning.
    status = BATCH_PARTIAL if conflicts else BATCH_UNDONE
    connection.execute(text(
        'UPDATE importbatches SET status = :status, undoneat = :undoneat '
        'WHERE batchid = :batchid'),
        {'status': status,
         'undoneat': None if conflicts else _utcnow(),
         'batchid': batchid})

    return UndoResult(batchid, reverted, conflicts)


def _revert_update(connection, table, targetid, column, oldvalue, newvalue):
    """Revert one cell to its pre-image, but only if it still holds what we
    wrote. Returns a Conflict if a human changed it since, else None."""
    from sqlalchemy import text

    pk = _primary_key(connection, table)
    current = connection.execute(text(
        'SELECT %s FROM %s WHERE %s = :targetid' % (column, table, pk)),
        {'targetid': targetid}).fetchone()
    if current is None:
        return Conflict(OP_UPDATE, table, targetid, column, 'row missing')

    if _as_text(current[0]) != _as_text(newvalue):
        return Conflict(OP_UPDATE, table, targetid, column,
                        'value changed since import')

    connection.execute(text(
        'UPDATE %s SET %s = :oldvalue WHERE %s = :targetid'
        % (table, column, pk)),
        {'oldvalue': oldvalue, 'targetid': targetid})
    return None


def _revert_insert(connection, batchid, table, targetid, references):
    """Hard-delete a row this batch created, only if nothing outside the batch
    still references it. Returns a Conflict when it is not safe, else None."""
    from sqlalchemy import inspect, text

    pk = _primary_key(connection, table)
    # A reference table may not exist here. Some referrers are plugin-owned -
    # notifications, for one - and a lean site drops the tables of plugins it
    # does not install (ADR-014). Querying an absent table raises 1146 on MySQL,
    # which rolls the whole undo back and leaves the batch permanently
    # un-undoable on that site. Skip what is not there; a table that does not
    # exist cannot hold a reference.
    present = set(inspect(connection).get_table_names())
    for reftable, refcolumn in (references or {}).get(table, []):
        if reftable not in present:
            continue
        referencing = connection.execute(text(
            'SELECT COUNT(*) FROM %s WHERE %s = :targetid'
            % (reftable, refcolumn)), {'targetid': targetid}).fetchone()[0]
        if referencing:
            return Conflict(OP_INSERT, table, targetid, None,
                            'still referenced by %s.%s' % (reftable, refcolumn))

    connection.execute(text(
        'DELETE FROM %s WHERE %s = :targetid' % (table, pk)),
        {'targetid': targetid})
    return None


def _primary_key(connection, table):
    """Single-column primary key name for a table, via the live schema."""
    from sqlalchemy import inspect

    pk = inspect(connection).get_pk_constraint(table)
    columns = pk.get('constrained_columns') or []
    if len(columns) != 1:
        raise ValueError(
            'journal undo needs a single-column primary key on %s' % table)
    return columns[0]


def format_undo_result(result):
    """Human-readable rendering of an UndoResult for the console or a UI panel."""
    lines = []
    lines.append('UNDO batch %s' % result.batchid)
    lines.append('  reverted : %d entries' % result.reverted)
    if result.conflicts:
        lines.append('  conflicts: %d (left as-is, not reverted)'
                     % len(result.conflicts))
        for conflict in result.conflicts:
            target = conflict.table
            if conflict.column:
                target = '%s.%s' % (conflict.table, conflict.column)
            lines.append('    ! %-8s %s #%s - %s' % (
                conflict.operation, target, conflict.targetid,
                conflict.reason))
    else:
        lines.append('  conflicts: none')
    return '\n'.join(lines)