When implementing HTTP/multipart requests and streaming (SSE/chunked) parsing with aiohttp, follow these rules:
Example patterns:
# Multipart: use aiohttp primitive rather than ad-hoc dict encoding
writer = aiohttp.MultipartWriter()
# ... append parts to writer ...
# Response lifecycle: release to pool (avoid breaking pooling semantics)
resp = await session.request(...)
try:
# read/parse as needed
...
finally:
if not resp.closed:
resp.release() # return connection to pool
# Streaming parser: be chunk-boundary agnostic
async def parse_stream_async(reader):
async for chunk in reader:
for line in chunk.splitlines():
# handle b"data: " prefix / DONE sentinel as appropriate
...
Enter the URL of a public GitHub repository