JezK

Edit File: track_modifications.py

from __future__ import annotations

import typing as t

import sqlalchemy as sa
import sqlalchemy.event
import sqlalchemy.orm
from flask import current_app
from flask import has_app_context
from flask.signals import Namespace  # type: ignore[attr-defined]

if t.TYPE_CHECKING:
    from .session import Session

_signals = Namespace()

models_committed = _signals.signal("models-committed")
"""This Blinker signal is sent after the session is committed if there were changed
models in the session.

The sender is the application that emitted the changes. The receiver is passed the
``changes`` argument with a list of tuples in the form ``(instance, operation)``.
The operations are ``"insert"``, ``"update"``, and ``"delete"``.
"""

before_models_committed = _signals.signal("before-models-committed")
"""This signal works exactly like :data:`models_committed` but is emitted before the
commit takes place.
"""

def _listen(session: sa.orm.scoped_session[Session]) -> None:
    sa.event.listen(session, "before_flush", _record_ops, named=True)
    sa.event.listen(session, "before_commit", _record_ops, named=True)
    sa.event.listen(session, "before_commit", _before_commit)
    sa.event.listen(session, "after_commit", _after_commit)
    sa.event.listen(session, "after_rollback", _after_rollback)

def _record_ops(session: Session, **kwargs: t.Any) -> None:
    if not has_app_context():
        return

if not current_app.config["SQLALCHEMY_TRACK_MODIFICATIONS"]:
        return

for targets, operation in (
        (session.new, "insert"),
        (session.dirty, "update"),
        (session.deleted, "delete"),
    ):
        for target in targets:
            state = sa.inspect(target)
            key = state.identity_key if state.has_identity else id(target)
            session._model_changes[key] = (target, operation)

def _before_commit(session: Session) -> None:
    if not has_app_context():
        return

app = current_app._get_current_object()  # type: ignore[attr-defined]

if not app.config["SQLALCHEMY_TRACK_MODIFICATIONS"]:
        return

if session._model_changes:
        changes = list(session._model_changes.values())
        before_models_committed.send(app, changes=changes)

def _after_commit(session: Session) -> None:
    if not has_app_context():
        return

app = current_app._get_current_object()  # type: ignore[attr-defined]

if not app.config["SQLALCHEMY_TRACK_MODIFICATIONS"]:
        return

if session._model_changes:
        changes = list(session._model_changes.values())
        models_committed.send(app, changes=changes)
        session._model_changes.clear()

def _after_rollback(session: Session) -> None:
    session._model_changes.clear()

Anmeldung Newsletter | Fairnetzt-Hochrhein