RunTimeError: got Future <Future pending> attached to a different loop when using custom loop in sync fixtures when upgrading from 0.14.2 to 0.15.0 #1315
Comments
sevaho
changed the title
|
I get the same issue with FastAPI (based on starlette) and mongo motor client. Just downgraded to 14.2 and fixed the problem as well. |
|
I've got similar issue in tests with FastAPI (0.68.2 --> 0.69.0 which means Starlette upgrading from 0.14.2 to 0.15.0) and databases library (based on asyncpg): ...
[stacktrace here]
...
asyncpg/protocol/protocol.pyx:323: in query
???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress
asyncpg/protocol/protocol.pyx:707: InterfaceErrorWith FastAPI 0.68.2 (Starlette 0.14.2) works like a charm. |
|
I think the problem is that In my case it poses a serious problem because within the definition of my app, I instantiate |
|
Shouldn't this be moved to the |
|
@sevaho Can you test if this version works without the error? @aminalaee No, I don't think it's related to Basically, I could be wrong but my initial understanding is:
This means that if you create any objects that bind to the default event loop before calling Does this sound like an accurate description of the situation? Asking because this is how I understand it, but I'm not 100% sure I understand |
|
@MatthewScholefield another error: Tested with starlette 0.15.0 and 0.16.0 and 0.17.1. |
|
Also faced with this problem. Described part of my project: main.py from fastapi import FastAPI
from app.api.api_v1.api import router as api_router
app = FastAPI()
app.include_router(api_router)apy/api.py from fastapi import APIRouter
from app.api.api_v1.endpoints.employee import router as user_router
router = APIRouter()
router.include_router(user_router)api/api_v1/endpoints/employee.py from fastapi import APIRouter
from app.crud.employee import EmployeeCRUD
router = APIRouter()
@router.get("/employee/count/", response_model=int, tags=["Data"])
async def count_employees():
return await EmployeeCRUD.count_employees()crud/employee.py from dataclasses import dataclass
from fastapi import APIRouter
from app.database.database import engine
from app.model.employee import Employee
router = APIRouter()
@dataclass
class EmployeeCRUD:
model = Employee
@classmethod
async def count_employees(cls):
return await engine.count(cls.model)database/database.py from motor.motor_asyncio import AsyncIOMotorClient
from odmantic import AIOEngine
from app.core.config import MONGO_DB_URL
client = AsyncIOMotorClient(MONGO_DB_URL)
engine = AIOEngine(motor_client=client)test from fastapi.testclient import TestClient
from app.main import app
def test_ping():
client = TestClient(app)
response = client.get('/api/employee/count/')
assert response.status_code == 200 |
|
@Hazzari have you tried setting the loop on Motor client? client = AsyncIOMotorClient()
client.get_io_loop = asyncio.get_event_loop
engine = AIOEngine(motor_client=client) |
Thank you so much! That solved the problem! |
I have the same issue about this with using SQLAlchemy async connect to to fix the problem, but obviously not a solution |
|
Here's a very minimal reproduction: import pytest
import asyncio
from starlette.responses import JSONResponse
from starlette.applications import Starlette
from starlette.routing import Route
async def fn(request):
print('in endpoint', id(asyncio.get_running_loop()))
return JSONResponse({})
def make_app():
app = Starlette(routes=[Route("/", endpoint=fn)])
return app
#######################
from starlette.testclient import TestClient
pytestmark = pytest.mark.anyio
@pytest.fixture
def anyio_backend():
return 'asyncio'
@pytest.fixture
async def tclient():
with TestClient(app=make_app()) as c:
yield c
@pytest.mark.anyio
async def test_bug(tclient):
print('in test', id(asyncio.get_running_loop()))
print(tclient.get('/'))when running this under This makes testing impossible for my use-case -- create a |
|
I'm not using Mongo so for now my only solution was to downgrade and pin fastapi to |
|
How to reproduce with |
|
I am also facing similar issue in my test cases |
|
My solution for async SQLAlchemy + FastApi + Pytest: in 'main.py' @app.on_event('startup')
async def startup_event():
await db.init()
await db.create_all()
@app.on_event('shutdown')
async def shutdown_event():
await db.close()in db file Base = declarative_base()
class Database:
def __init__(self, url):
self._session = None
self._engine = None
self.url = url
async def create_all(self):
async with self._engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
def __getattr__(self, name) -> AsyncSession:
return getattr(self._session, name)
async def init(self):
# closes connections if a session is created,
# so as not to create repeated connections
if self._session:
await self._session.close()
self._engine = create_async_engine(self.url, future=True)
self._session = sessionmaker(
self._engine,
expire_on_commit=False,
class_=AsyncSession
)()
db: AsyncSession | Database = Database(settings.db_url())and client object in "test.py" @pytest.fixture(scope="module")
def client() -> TestClient:
with TestClient(app) as client:
yield client
def test_get_files_without_params(client: TestClient):
response = client.get("/api/v1/file") |
|
I am seeing similar issues, where I have my db_session fixture that ensures unittest changes are rolled back after every test, this requires the app to use the same session to be able to see the data loaded into the database. With the change to anyio, I am yet to find a combination of things that will allow me to have the fixture db_session on the same loop as the TestClient. The following worked prior to anyio. server.py def build_app():
app = FastAPI()
@app.on_event("startup")
async def app_setup():
app.state.database_pool = await asyncpg.create_pool(
dsn=config.postgres_dsn,
min_size=config.postgres_pool_min_size,
max_size=config.postgres_pool_max_size,
server_settings={
"application_name": "tfx_backend_v{}".format(tfx_backend.__version__),
},
)
@app.on_event("shutdown")
async def app_shutdown():
# cleanly shutdown the connections in the pool
await app.state.database_pool.close()
@app.middleware("http")
async def middleware_asyncpg(request, call_next):
# on request, inject a database transaction
async with request.app.state.database_pool.acquire() as database_connection:
async with database_connection.transaction():
request.state.database_connection = database_connection
return await call_next(request)
return apptests/conftest.py @pytest.fixture(scope="session")
def engine(settings):
backend = get_backend(settings.postgres_dsn)
_upgrade(backend)
yield
# Rollback all migrations
_downgrade(backend, count=None)
@pytest.fixture(scope="function", autouse=True)
async def db_session(engine, settings):
conn = await asyncpg.connect(settings.postgres_dsn)
tr = conn.transaction()
await tr.start()
try:
yield conn
except Exception:
pass
finally:
await tr.rollback()tests/views/conftest.py import contextlib
class MockPool:
def __init__(self, db_session):
self.db_session = db_session
@contextlib.asynccontextmanager
async def acquire(self):
yield self.db_session
async def close(self):
pass
# We don't want to build the app for every test, create a session level fixture for the app
@pytest.fixture(scope="session")
def app():
return server.build_app()
# We do want to patch the db_session for every test, create a funcion level fixture for the client.
@pytest.fixture
def client(monkeypatch, db_session, app, tmp_path):
mock_pool = MockPool(db_session)
monkeypatch.setattr(server.asyncpg, "create_pool", AsyncMock(return_value=mock_pool))
with TestClient(app=app, raise_server_exceptions=False, base_url="http://localhost") as client:
yield clientThis allowed me to populate the database, then make a request on the TestClient and see the same data. If there was a way to access the loop that TestClient started up, I could create a new db_session for the client as asyncpg will accept a loop parameter, allowing me to run them on the same loop (hopefully). As long as the TestClient doesn't spawn new loops on requests. |
you can use: @contextlib.asynccontextmanager
async def some_function(app):
async with some_resource() as v:
app.v = v
yield
with TestClient(app=app) as tc, tc.portal.wrap_async_context_manager(some_function(app)):
... |
|
I have a feeling that has worked, so I now get the same error in other fixtures that use the db_session, as its now connected to the TestClient loop. @contextlib.asynccontextmanager
async def session_factory(app):
conn = await asyncpg.connect(app.state.config.postgres_dsn)
tr = conn.transaction()
await tr.start()
try:
app.state.database_pool = MockPool(conn)
yield
except Exception:
pass
finally:
await tr.rollback()
# We don't want to build the app for every test, create a session level fixture for the app
@pytest.fixture(scope="session")
def app():
return server.build_app()
# We do want to patch the db_session for every test, create a funcion level fixture for the client.
@pytest.fixture
def client(engine, app):
with TestClient(
app=app,
raise_server_exceptions=False,
base_url="http://localhost",
) as client, client.portal.wrap_async_context_manager(session_factory(app)):
yield client
@pytest.fixture
def db_session(client):
return client.app.state.database_pool.db_sessionThe RuntimeError now raises from the database factory's creating users etc. Rather than from within the view when calling the TestClient. Seems like I need to influence the loop that TestClient creates (like it used to attach to the test loop). |
- Using this version of FastAPI gives us Starlette 0.14.2 which fixes a bug we faced - Bug described at encode/starlette#1315 - Downgrading FastAPI means `UploadFile` is no longer supported. I've reverted to `File()` as an alternative solution, although `UploadFile` is probably easier to use, and I'd reimplement it in that way if we could upgrade in the future
|
I have faced the same issues in my unit tests. I fixed the issue by using a context manager for Doesn't work: @pytest.fixture(scope="session")
def client() -> TestClient:
return TestClient(app)Works: @pytest.fixture(scope="session")
def client() -> TestClient:
with TestClient(app) as c:
yield c |
|
Any time I am doing async stuff outside of my app I'll just switch to httpx so that I don't have to deal with juggling event looks. I realize it requires rewriting of tests so it may not be a great solution for existing tests that were broken, but it might be a good idea for new tests. |
The latter calls the startup event, while the former doesn't. |
|
Same issue. Had to address by pinning Fastapi to |
I continue to see the error even after making this change. Could you please share an example of how you're using that |
|
FastAPI needs context manager support for lifespan |
|
For testing FastAPI with SQLAlchemy, one option is overriding the engine to use a NullPool (https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html#using-multiple-asyncio-event-loops). |
Thanks for this! This helped me figure out that the issue is creating asyncio.Lock or similar objects at the module level (so before the event loop is created). I've created a fix for the redis package: redis/redis-py#2471. I think this will have to be solved at the package level, or users should not define the connections from the offending packages at the module level. |
|
Still actual in 2023, if you encountered with the same issue, here is workaround that helped me: first: set up asgi-lifespan https://pypi.org/project/asgi-lifespan/ If you have any connection initialization actions in fixtures with session level make sure you override event_loop fixture. otherwise you will receive ScopeMismatch error. Startup your test client with LifespanManager: |
|
This comment shows the issue: #1315 (comment) The event loop ID is different on both prints. |
and others
Checklist
master.Describe the bug
Upgrading starlette>=0.15.0 breaks current testing strategy. The setup is mocking a nats subscription by actually using the nats server.
The code works with starlette 0.14.2, upgradign to 0.15.0 gives
RunTumeError got Future <Future pending> attached to a different loop. When upgrading to starlette 0.16.0 it gives TimeOut errors. I would love to keep tests sync.To reproduce
requirements.txt
code
Run:
Expected behavior
The test to work.
Actual behavior
Test does not work.
Debugging material
output running with starlette 0.15.0:
output when running with starlette 0.16.0:
Environment
Additional context
Important
The text was updated successfully, but these errors were encountered: