Redis pub/sub for real-time notifications in FastAPI

<p>Need to push real-time updates to connected clients when events occur (e.g., trace validated, vote received). Using Server-Sent Events (SSE) for the client side, need Redis pub/sub to fan out events from any worker to all API instances.</p>

<p>Use Redis pub/sub with asyncio to stream events via SSE:</p> <div class="highlight"><pre><span></span><code><span class="kn">import</span><span class="w"> </span><span class="nn">asyncio</span> <span class="kn">import</span><span class="w"> </span><span class="nn">json</span> <span class="kn">from</span><span class="w"> </span><span class="nn">fastapi</span><span class="w"> </span><span class="kn">import</span> <span class="n">APIRouter</span><span class="p">,</span> <span class="n">Depends</span> <span class="kn">from</span><span class="w"> </span><span class="nn">fastapi.responses</span><span class="w"> </span><span class="kn">import</span> <span class="n">StreamingResponse</span> <span class="kn">from</span><span class="w"> </span><span class="nn">redis.asyncio</span><span class="w"> </span><span class="kn">import</span> <span class="n">Redis</span> <span class="kn">from</span><span class="w"> </span><span class="nn">app.dependencies</span><span class="w"> </span><span class="kn">import</span> <span class="n">get_redis</span> <span class="n">router</span> <span class="o">=</span> <span class="n">APIRouter</span><span class="p">()</span> <span class="c1"># Publisher (call from anywhere in the app)</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">publish_event</span><span class="p">(</span><span class="n">redis</span><span class="p">:</span> <span class="n">Redis</span><span class="p">,</span> <span class="n">channel</span><span class="p">:</span> <span class="nb">str</span><span class="p">,</span> <span class="n">data</span><span class="p">:</span> <span class="nb">dict</span><span class="p">)</span> <span class="o">-&gt;</span> <span class="kc">None</span><span class="p">:</span> <span class="k">await</span> <span class="n">redis</span><span class="o">.</span><span class="n">publish</span><span class="p">(</span><span class="n">channel</span><span class="p">,</span> <span class="n">json</span><span class="o">.</span><span class="n">dumps</span><span class="p">(</span><span class="n">data</span><span class="p">))</span> <span class="c1"># SSE endpoint</span> <span class="nd">@router</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="s1">'/events/</span><span class="si">{user_id}</span><span class="s1">'</span><span class="p">)</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">stream_events</span><span class="p">(</span> <span class="n">user_id</span><span class="p">:</span> <span class="nb">str</span><span class="p">,</span> <span class="n">redis</span><span class="p">:</span> <span class="n">Redis</span> <span class="o">=</span> <span class="n">Depends</span><span class="p">(</span><span class="n">get_redis</span><span class="p">)</span> <span class="p">)</span> <span class="o">-&gt;</span> <span class="n">StreamingResponse</span><span class="p">:</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">event_generator</span><span class="p">():</span> <span class="n">pubsub</span> <span class="o">=</span> <span class="n">redis</span><span class="o">.</span><span class="n">pubsub</span><span class="p">()</span> <span class="k">await</span> <span class="n">pubsub</span><span class="o">.</span><span class="n">subscribe</span><span class="p">(</span><span class="sa">f</span><span class="s1">'user:</span><span class="si">{</span><span class="n">user_id</span><span class="si">}</span><span class="s1">'</span><span class="p">)</span> <span class="k">try</span><span class="p">:</span> <span class="k">while</span> <span class="kc">True</span><span class="p">:</span> <span class="n">message</span> <span class="o">=</span> <span class="k">await</span> <span class="n">pubsub</span><span class="o">.</span><span class="n">get_message</span><span class="p">(</span><span class="n">ignore_subscribe_messages</span><span class="o">=</span><span class="kc">True</span><span class="p">,</span> <span class="n">timeout</span><span class="o">=</span><span class="mf">1.0</span><span class="p">)</span> <span class="k">if</span> <span class="n">message</span> <span class="ow">is</span> <span class="ow">not</span> <span class="kc">None</span><span class="p">:</span> <span class="n">data</span> <span class="o">=</span> <span class="n">message</span><span class="p">[</span><span class="s1">'data'</span><span class="p">]</span> <span class="k">if</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">data</span><span class="p">,</span> <span class="nb">bytes</span><span class="p">):</span> <span class="n">data</span> <span class="o">=</span> <span class="n">data</span><span class="o">.</span><span class="n">decode</span><span class="p">()</span> <span class="k">yield</span> <span class="sa">f</span><span class="s1">'data: </span><span class="si">{</span><span class="n">data</span><span class="si">}</span><span class="se">\n\n</span><span class="s1">'</span> <span class="k">else</span><span class="p">:</span> <span class="c1"># Keepalive ping</span> <span class="k">yield</span> <span class="s1">': ping</span><span class="se">\n\n</span><span class="s1">'</span> <span class="k">await</span> <span class="n">asyncio</span><span class="o">.</span><span class="n">sleep</span><span class="p">(</span><span class="mi">15</span><span class="p">)</span> <span class="k">finally</span><span class="p">:</span> <span class="k">await</span> <span class="n">pubsub</span><span class="o">.</span><span class="n">unsubscribe</span><span class="p">(</span><span class="sa">f</span><span class="s1">'user:</span><span class="si">{</span><span class="n">user_id</span><span class="si">}</span><span class="s1">'</span><span class="p">)</span> <span class="k">await</span> <span class="n">pubsub</span><span class="o">.</span><span class="n">close</span><span class="p">()</span> <span class="k">return</span> <span class="n">StreamingResponse</span><span class="p">(</span> <span class="n">event_generator</span><span class="p">(),</span> <span class="n">media_type</span><span class="o">=</span><span class="s1">'text/event-stream'</span><span class="p">,</span> <span class="n">headers</span><span class="o">=</span><span class="p">{</span><span class="s1">'Cache-Control'</span><span class="p">:</span> <span class="s1">'no-cache'</span><span class="p">,</span> <span class="s1">'X-Accel-Buffering'</span><span class="p">:</span> <span class="s1">'no'</span><span class="p">}</span> <span class="p">)</span> <span class="c1"># Usage: publish when a trace is validated</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">on_trace_validated</span><span class="p">(</span><span class="n">trace</span><span class="p">:</span> <span class="n">Trace</span><span class="p">,</span> <span class="n">redis</span><span class="p">:</span> <span class="n">Redis</span><span class="p">)</span> <span class="o">-&gt;</span> <span class="kc">None</span><span class="p">:</span> <span class="k">await</span> <span class="n">publish_event</span><span class="p">(</span><span class="n">redis</span><span class="p">,</span> <span class="sa">f</span><span class="s1">'user:</span><span class="si">{</span><span class="n">trace</span><span class="o">.</span><span class="n">contributor_id</span><span class="si">}</span><span class="s1">'</span><span class="p">,</span> <span class="p">{</span> <span class="s1">'type'</span><span class="p">:</span> <span class="s1">'trace_validated'</span><span class="p">,</span> <span class="s1">'trace_id'</span><span class="p">:</span> <span class="nb">str</span><span class="p">(</span><span class="n">trace</span><span class="o">.</span><span class="n">id</span><span class="p">),</span> <span class="s1">'title'</span><span class="p">:</span> <span class="n">trace</span><span class="o">.</span><span class="n">title</span><span class="p">,</span> <span class="p">})</span> </code></pre></div> <p>The <code>X-Accel-Buffering: no</code> header disables nginx buffering so events reach the client immediately. Always handle client disconnects by catching <code>asyncio.CancelledError</code> in the generator.</p>