Merge branch 'main' of https://github.com/sysid/sse-starlette

Author: sysid

.github/dependabot.yml

@@ -13,3 +13,8 @@ updates:
     commit-message:
       prefix: ":arrow_up:"
     open-pull-requests-limit: 5
+    
+  - package-ecosystem: "github-actions" 
+    directory: "/"
+    schedule:
+      interval: "monthly"

.github/workflows/build.yml

@@ -46,14 +46,15 @@ jobs:
             fastapi \
             tenacity \
             portend \
-            testcontainers
+            testcontainers \
+            async-timeout
 
       - name: Run tests
         run: |
           make test-unit
 
       - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v4
+        uses: codecov/codecov-action@v5
         if: contains(env.USING_COVERAGE, matrix.python-version)
         with:
           token: ${{ secrets.CODECOV_TOKEN }}

README.md

@@ -69,6 +69,39 @@ async def endless(req: Request):
     return EventSourceResponse(event_publisher())
 ```
 
+# Thread Safety with SQLAlchemy Sessions and Similar Objects
+
+The streaming portion of this package is accomplished via anyio TaskGroups. Care
+needs to be taken to avoid passing objects that are not thread-safe to generators
+you use to yield streaming data.
+
+For example, if you are using SQLAlchemy, you should not use/pass an `AsyncSession`
+object to your generator:
+
+```python
+# ❌ This can result in "The garbage collector is trying to clean up non-checked-in connection..." errors
+async def bad_route():
+    async with AsyncSession() as session:
+        async def generator():
+            async for row in session.execute(select(User)):
+                yield dict(data=row)
+
+    return EventSourceResponse(generator)
+```
+
+Instead, ensure you create sessions within the generator itself
+
+```python
+# ✅ This is safe
+async def good_route():
+    async def generator():
+        async with AsyncSession() as session:
+            async for row in session.execute(select(User)):
+                yield dict(data=row)
+
+    return EventSourceResponse(generator)
+```
+
 ## Special use cases
 ### Customize Ping
 By default, the server sends a ping every 15 seconds. You can customize this by:
@@ -112,6 +145,26 @@ Async generators can expose tricky error and cleanup behavior especially when th
 Example [`no_async_generators.py`](https://github.com/sysid/sse-starlette/pull/56#issue-1704495339) shows an alternative implementation
 that does not rely on async generators but instead uses memory channels (`examples/no_async_generators.py`).
 
+### Using pytest to test SSE Endpoints
+When testing more than a single SSE endpoint via pytest, one may encounter the following error: `RuntimeError: <asyncio.locks.Event object at 0xxxx [unset]> is bound to a different event loop`.
+
+A workaround to fix this error is to use the following fixture on all tests that use SSE:
+
+```python
+@pytest.fixture
+def reset_sse_starlette_appstatus_event():
+    """
+    Fixture that resets the appstatus event in the sse_starlette app.
+
+    Should be used on any test that uses sse_starlette to stream events.
+    """
+    # See https://github.com/sysid/sse-starlette/issues/59
+    from sse_starlette.sse import AppStatus
+
+    AppStatus.should_exit_event = None
+```
+
+For details, see [Issue#59](https://github.com/sysid/sse-starlette/issues/59#issuecomment-1961678665).
 
 ## Development, Contributing
 1. install pdm: `pip install pdm`

examples/example_fastapi_sqlalchemy.py

@@ -0,0 +1,47 @@
+import typing as T
+
+import sqlalchemy as sa
+from fastapi import Depends, FastAPI
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from sse_starlette.sse import EventSourceResponse
+
+# Database
+db_bind = create_async_engine("sqlite+aiosqlite://:memory:")
+AsyncSessionLocal = async_sessionmaker(bind=db_bind, expire_on_commit=False)
+
+
+async def async_db_session():
+    async with AsyncSessionLocal() as session:
+        yield session
+
+
+AsyncDbSessionDependency = T.Annotated[AsyncSession, Depends(async_db_session)]
+
+TODOS_CTE_SQL = """
+WITH todo AS (
+    SELECT 1 AS id, 'Task 1' AS title, 'Description 1' AS description, 0 AS completed
+    UNION ALL
+    SELECT 2, 'Task 2', 'Description 2', 1
+    UNION ALL
+    SELECT 3, 'Task 3', 'Description 3', 0
+)
+
+SELECT * FROM todo
+"""
+
+# App
+app = FastAPI()
+
+
+@app.route("/things")
+async def things(db_session: AsyncDbSessionDependency):
+    # Safe to use db_session here to do auth or something else.
+    async def thing_streamer():
+        # Do *NOT* reuse db_session here within the AsyncGenerator, create a
+        # new session instead.
+        async with AsyncSessionLocal() as session:
+            async for row in session.execute(sa.text(TODOS_CTE_SQL)):
+                yield {"data": dict(row)}
+
+    return EventSourceResponse(thing_streamer)