SQLAlchemy event listeners for audit logging

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

<p>Use 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="kn">from</span><span class="w"> </span><span class="nn">sqlalchemy.orm</span><span class="w"> </span><span class="kn">import</span> <span class="n">Session</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="c1"># Per-model listener:</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">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="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="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="c1"># Only log status changes</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_status</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_status</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"># Session-level listener (all models):</span> <span class="nd">@event</span><span class="o">.</span><span class="n">listens_for</span><span class="p">(</span><span class="n">Session</span><span class="p">,</span> <span class="s1">'after_bulk_delete'</span><span class="p">)</span> <span class="k">def</span><span class="w"> </span><span class="nf">after_bulk_delete</span><span class="p">(</span><span class="n">delete_context</span><span class="p">):</span> <span class="n">log</span><span class="o">.</span><span class="n">warning</span><span class="p">(</span><span class="s1">'bulk_delete'</span><span class="p">,</span> <span class="n">table</span><span class="o">=</span><span class="n">delete_context</span><span class="o">.</span><span class="n">primary_table</span><span class="o">.</span><span class="n">name</span><span class="p">)</span> </code></pre></div> <p>For a mixin-based approach:</p> <div class="highlight"><pre><span></span><code><span class="k">class</span><span class="w"> </span><span class="nc">AuditMixin</span><span class="p">:</span> <span class="nd">@staticmethod</span> <span class="k">def</span><span class="w"> </span><span class="nf">_after_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="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="bp">cls</span><span class="o">.</span><span class="n">_after_insert</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 the SQL executes but BEFORE commit - Use <code>inspect(target).attrs.field.history</code> to get before/after values - Keep event listeners lightweight — they run synchronously in the DB transaction - For async sessions, use <code>async_object_session(target)</code> to get the session</p>