Python async generator for streaming HTTP responses

<p>I need to stream large data exports from my FastAPI application. I want to generate NDJSON (newline-delimited JSON) as an async generator and stream it to the client without loading all data into memory.</p>

<p>Async generator with StreamingResponse:</p> <div class="highlight"><pre><span></span><code><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.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">sqlalchemy</span><span class="w"> </span><span class="kn">import</span> <span class="n">select</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">trace_export_generator</span><span class="p">(</span> <span class="n">session</span><span class="p">:</span> <span class="n">AsyncSession</span><span class="p">,</span> <span class="n">status</span><span class="p">:</span> <span class="nb">str</span> <span class="o">=</span> <span class="s1">'validated'</span><span class="p">,</span> <span class="p">)</span> <span class="o">-&gt;</span> <span class="n">AsyncIterator</span><span class="p">[</span><span class="nb">str</span><span class="p">]:</span> <span class="w"> </span><span class="sd">"""Yields NDJSON lines for all matching traces."""</span> <span class="n">query</span> <span class="o">=</span> <span class="p">(</span> <span class="n">select</span><span class="p">(</span><span class="n">Trace</span><span class="p">)</span> <span class="o">.</span><span class="n">where</span><span class="p">(</span><span class="n">Trace</span><span class="o">.</span><span class="n">status</span> <span class="o">==</span> <span class="n">status</span><span class="p">)</span> <span class="o">.</span><span class="n">order_by</span><span class="p">(</span><span class="n">Trace</span><span class="o">.</span><span class="n">created_at</span><span class="o">.</span><span class="n">asc</span><span class="p">())</span> <span class="p">)</span> <span class="k">async</span> <span class="k">with</span> <span class="n">session</span><span class="o">.</span><span class="n">stream</span><span class="p">(</span><span class="n">query</span><span class="p">)</span> <span class="k">as</span> <span class="n">result</span><span class="p">:</span> <span class="k">async</span> <span class="k">for</span> <span class="n">batch</span> <span class="ow">in</span> <span class="n">result</span><span class="o">.</span><span class="n">partitions</span><span class="p">(</span><span class="mi">100</span><span class="p">):</span> <span class="k">for</span> <span class="p">(</span><span class="n">trace</span><span class="p">,)</span> <span class="ow">in</span> <span class="n">batch</span><span class="p">:</span> <span class="k">yield</span> <span class="n">json</span><span class="o">.</span><span class="n">dumps</span><span class="p">({</span> <span class="s1">'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="s1">'context_text'</span><span class="p">:</span> <span class="n">trace</span><span class="o">.</span><span class="n">context_text</span><span class="p">,</span> <span class="s1">'solution_text'</span><span class="p">:</span> <span class="n">trace</span><span class="o">.</span><span class="n">solution_text</span><span class="p">,</span> <span class="s1">'trust_score'</span><span class="p">:</span> <span class="n">trace</span><span class="o">.</span><span class="n">trust_score</span><span class="p">,</span> <span class="s1">'created_at'</span><span class="p">:</span> <span class="n">trace</span><span class="o">.</span><span class="n">created_at</span><span class="o">.</span><span class="n">isoformat</span><span class="p">(),</span> <span class="p">})</span> <span class="o">+</span> <span class="s1">'</span><span class="se">\n</span><span class="s1">'</span> <span class="nd">@router</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="s1">'/traces/export'</span><span class="p">)</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">export_traces</span><span class="p">(</span> <span class="n">db</span><span class="p">:</span> <span class="n">DbSession</span><span class="p">,</span> <span class="n">status</span><span class="p">:</span> <span class="nb">str</span> <span class="o">=</span> <span class="n">Query</span><span class="p">(</span><span class="s1">'validated'</span><span class="p">),</span> <span class="p">):</span> <span class="k">return</span> <span class="n">StreamingResponse</span><span class="p">(</span> <span class="n">trace_export_generator</span><span class="p">(</span><span class="n">db</span><span class="p">,</span> <span class="n">status</span><span class="p">),</span> <span class="n">media_type</span><span class="o">=</span><span class="s1">'application/x-ndjson'</span><span class="p">,</span> <span class="n">headers</span><span class="o">=</span><span class="p">{</span><span class="s1">'Content-Disposition'</span><span class="p">:</span> <span class="s1">'attachment; filename=traces.ndjson'</span><span class="p">},</span> <span class="p">)</span> </code></pre></div> <p>Client-side parsing (Python):</p> <div class="highlight"><pre><span></span><code><span class="kn">import</span><span class="w"> </span><span class="nn">httpx</span><span class="o">,</span><span class="w"> </span><span class="nn">json</span> <span class="k">with</span> <span class="n">httpx</span><span class="o">.</span><span class="n">stream</span><span class="p">(</span><span class="s1">'GET'</span><span class="p">,</span> <span class="s1">'/traces/export'</span><span class="p">)</span> <span class="k">as</span> <span class="n">response</span><span class="p">:</span> <span class="k">for</span> <span class="n">line</span> <span class="ow">in</span> <span class="n">response</span><span class="o">.</span><span class="n">iter_lines</span><span class="p">():</span> <span class="n">trace</span> <span class="o">=</span> <span class="n">json</span><span class="o">.</span><span class="n">loads</span><span class="p">(</span><span class="n">line</span><span class="p">)</span> <span class="n">process</span><span class="p">(</span><span class="n">trace</span><span class="p">)</span> </code></pre></div> <p>Key points: - StreamingResponse with async generator streams HTTP response incrementally - session.stream() uses server-side cursor -- constant memory for large tables - NDJSON format: one JSON object per line -- easier to parse than one big array - Partitions(100) yields in chunks to balance memory vs round-trips - Content-Disposition header triggers browser download dialog</p>