I'm trying to initialize a ConnectionPool using asyncpg and then I'm using this pool to create a connection each time I need to read something from my DB. It was working fine until I moved my SQL endpoints to an APIRouter. I'm a bit lost now since I couldn't find docs for this specific case.
To create the connection, I'm using the app's lifecycle on main.py:
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.db_pool = await create_async_pool()
yield
await app.state.db_pool.close()
Then I'm declaring two dependencies in main.py:
async def get_database():
yield app.state.db_pool
async def get_db_connection(pool: asyncpg.pool.Pool = Depends(get_database)):
async with pool.acquire() as connection:
yield connection
Up until here, everything works perfectly fine and I can use my database on my app's endpoints written on main.py. Now I want to migrate these endpoints to a new API router on a file called analytics.py. Importing these dependencies directly from my analytics.py returns circular imports errors, but I'm completely unable to pass on the latest dependency into my new Analytics router. My file tree currently looks like this:
.
├── app
│ ├── __init__.py
│ ├── main.py
│ └── analytics
│ ├── __init__.py
│ └── analytics.py
Any idea how I could fix this? There must definitely be something that I'm missing
My complete files look like this, main.py:
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.db_pool = await create_async_pool()
yield
await app.state.db_pool.close()
app = FastAPI(lifespan=lifespan)
app.include_router(analytics.router)
async def get_database():
yield app.state.db_pool
async def get_db_connection(pool: asyncpg.pool.Pool = Depends(get_database)):
async with pool.acquire() as connection:
yield connection
@app.get("/"):
async def hello():
return {"message": "Hello world!"}
analytics.py:
from fastapi import APIRouter
from app.main import get_db_connection
router = APIRouter(
prefix="/analytics"
)
@router.post("/test")
async def test_analytics(input_data, db = Depends(get_db_connection)):
return "OK"
The error with this configuration is:
Import Error: cannot import name 'get_db_connection' from partially initialized module 'app.main' (most likely due to a circular import)
We don't see the full code of your main and analytics modules. So, it's hard to say for sure why you have this circular imports errors.
But moving these two dependencies into separate file (e.g. dependencies.py) should help you.
Also, modify lifespant to:
@asynccontextmanager
async def lifespan(app: FastAPI):
db_pool = await create_async_pool()
yield {"db_pool": db_pool}
await db_pool.close()
And use request.state instead of app.state in get_database:
from fastapi import Request
async def get_database(request: Request):
yield request.state.db_pool