Python SQLAlchemy event listeners for audit logging

<p>I need to automatically track when rows are inserted or updated in PostgreSQL for audit purposes. I want this without modifying every query in my codebase.</p>

<p>SQLAlchemy ORM event listeners:</p> <div class="highlight"><pre><span></span><code><span class="kn">from</span><span class="w"> </span><span class="nn">sqlalchemy</span><span class="w"> </span><span class="kn">import</span> <span class="n">event</span><span class="p">,</span> <span class="n">inspect</span> <span class="kn">import</span><span class="w"> </span><span class="nn">structlog</span> <span class="n">log</span> <span class="o">=</span> <span class="n">structlog</span><span class="o">.</span><span class="n">get_logger</span><span class="p">()</span> <span class="nd">@event</span><span class="o">.</span><span class="n">listens_for</span><span class="p">(</span><span class="n">Trace</span><span class="p">,</span> <span class="s1">'after_insert'</span><span class="p">)</span> <span class="k">def</span><span class="w"> </span><span class="nf">after_trace_insert</span><span class="p">(</span><span class="n">mapper</span><span class="p">,</span> <span class="n">connection</span><span class="p">,</span> <span class="n">target</span><span class="p">):</span> <span class="n">log</span><span class="o">.</span><span class="n">info</span><span class="p">(</span> <span class="s1">'trace_created'</span><span class="p">,</span> <span class="n">trace_id</span><span class="o">=</span><span class="nb">str</span><span class="p">(</span><span class="n">target</span><span class="o">.</span><span class="n">id</span><span class="p">),</span> <span class="n">is_seed</span><span class="o">=</span><span class="n">target</span><span class="o">.</span><span class="n">is_seed</span><span class="p">,</span> <span class="n">contributor_id</span><span class="o">=</span><span class="nb">str</span><span class="p">(</span><span class="n">target</span><span class="o">.</span><span class="n">contributor_id</span><span class="p">),</span> <span class="p">)</span> <span class="nd">@event</span><span class="o">.</span><span class="n">listens_for</span><span class="p">(</span><span class="n">Trace</span><span class="p">,</span> <span class="s1">'after_update'</span><span class="p">)</span> <span class="k">def</span><span class="w"> </span><span class="nf">after_trace_update</span><span class="p">(</span><span class="n">mapper</span><span class="p">,</span> <span class="n">connection</span><span class="p">,</span> <span class="n">target</span><span class="p">):</span> <span class="n">state</span> <span class="o">=</span> <span class="n">inspect</span><span class="p">(</span><span class="n">target</span><span class="p">)</span> <span class="n">history</span> <span class="o">=</span> <span class="n">state</span><span class="o">.</span><span class="n">attrs</span><span class="o">.</span><span class="n">status</span><span class="o">.</span><span class="n">history</span> <span class="k">if</span> <span class="n">history</span><span class="o">.</span><span class="n">has_changes</span><span class="p">():</span> <span class="n">log</span><span class="o">.</span><span class="n">info</span><span class="p">(</span> <span class="s1">'trace_status_changed'</span><span class="p">,</span> <span class="n">trace_id</span><span class="o">=</span><span class="nb">str</span><span class="p">(</span><span class="n">target</span><span class="o">.</span><span class="n">id</span><span class="p">),</span> <span class="n">old</span><span class="o">=</span><span class="n">history</span><span class="o">.</span><span class="n">deleted</span><span class="p">[</span><span class="mi">0</span><span class="p">]</span> <span class="k">if</span> <span class="n">history</span><span class="o">.</span><span class="n">deleted</span> <span class="k">else</span> <span class="kc">None</span><span class="p">,</span> <span class="n">new</span><span class="o">=</span><span class="n">target</span><span class="o">.</span><span class="n">status</span><span class="p">,</span> <span class="p">)</span> <span class="c1"># Mixin for all models:</span> <span class="k">class</span><span class="w"> </span><span class="nc">AuditMixin</span><span class="p">:</span> <span class="nd">@classmethod</span> <span class="k">def</span><span class="w"> </span><span class="nf">__init_subclass__</span><span class="p">(</span><span class="bp">cls</span><span class="p">,</span> <span class="o">**</span><span class="n">kwargs</span><span class="p">):</span> <span class="nb">super</span><span class="p">()</span><span class="o">.</span><span class="n">__init_subclass__</span><span class="p">(</span><span class="o">**</span><span class="n">kwargs</span><span class="p">)</span> <span class="n">event</span><span class="o">.</span><span class="n">listen</span><span class="p">(</span><span class="bp">cls</span><span class="p">,</span> <span class="s1">'after_insert'</span><span class="p">,</span> <span class="n">AuditMixin</span><span class="o">.</span><span class="n">_log_insert</span><span class="p">)</span> <span class="nd">@staticmethod</span> <span class="k">def</span><span class="w"> </span><span class="nf">_log_insert</span><span class="p">(</span><span class="n">mapper</span><span class="p">,</span> <span class="n">connection</span><span class="p">,</span> <span class="n">target</span><span class="p">):</span> <span class="n">log</span><span class="o">.</span><span class="n">info</span><span class="p">(</span><span class="s1">'inserted'</span><span class="p">,</span> <span class="n">model</span><span class="o">=</span><span class="nb">type</span><span class="p">(</span><span class="n">target</span><span class="p">)</span><span class="o">.</span><span class="vm">__name__</span><span class="p">,</span> <span class="nb">id</span><span class="o">=</span><span class="nb">str</span><span class="p">(</span><span class="n">target</span><span class="o">.</span><span class="n">id</span><span class="p">))</span> <span class="k">class</span><span class="w"> </span><span class="nc">Trace</span><span class="p">(</span><span class="n">AuditMixin</span><span class="p">,</span> <span class="n">Base</span><span class="p">):</span> <span class="o">...</span> </code></pre></div> <p>Key points: - ORM events fire AFTER SQL executes but BEFORE commit - inspect(target).attrs.field.history gives before/after values for update events - Keep event listeners lightweight -- they run synchronously in the DB transaction - For async sessions, use async_object_session(target) to get the session</p>