feat: add support for custom async iterable objects

examples/custom_generator.py

@@ -0,0 +1,43 @@
+import asyncio
+
+from fastapi import FastAPI, Depends
+from sse_starlette import EventSourceResponse, ServerSentEvent
+from starlette import status
+
+
+class Stream:
+    def __init__(self) -> None:
+        self._queue = asyncio.Queue[ServerSentEvent]()
+
+    def __aiter__(self) -> "Stream":
+        return self
+
+    async def __anext__(self) -> ServerSentEvent:
+        return await self._queue.get()
+
+    async def asend(self, value: ServerSentEvent) -> None:
+        await self._queue.put(value)
+
+
+app = FastAPI()
+
+_stream = Stream()
+app.dependency_overrides[Stream] = lambda: _stream
+
+
+@app.get("/sse")
+async def sse(stream: Stream = Depends()) -> EventSourceResponse:
+    return EventSourceResponse(stream)
+
+
+@app.post("/message", status_code=status.HTTP_201_CREATED)
+async def send_message(message: str, stream: Stream = Depends()) -> None:
+    await stream.asend(
+        ServerSentEvent(data=message)
+    )
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="127.0.0.1", port=8000)

sse_starlette/sse.py

@@ -1,5 +1,4 @@
 import enum
-import inspect
 import io
 import logging
 import re
@@ -152,7 +151,7 @@ def __init__(
     ) -> None:
         self.sep = sep
         self.ping_message_factory = ping_message_factory
-        if inspect.isasyncgen(content):
+        if isinstance(content, AsyncIterable):
             self.body_iterator = (
                 content
             )  # type: AsyncIterable[Union[Any,dict,ServerSentEvent]]

tests/integration_testing.http

@@ -2,3 +2,15 @@ GET http://localhost:8000/endless
 Accept: application/json
 
 ###
+
+# curl -X 'POST'
+#  'http://127.0.0.1:8000/message?message=xxx'
+#  -H 'accept: application/json'
+#  -d ''
+POST http://127.0.0.1:8000/message?message=xxx
+accept: application/json
+Content-Type: application/x-www-form-urlencoded
+
+###
+
+