👀 Views: 22 💬 Answers: 1 📅 Created: 2025-06-16

FastAPI SQLAlchemy asyncio Python

I'm a bit lost with I'm currently developing an API using FastAPI and SQLAlchemy, and I'm running into issues with asynchronous database queries. I want to utilize async/await for non-blocking operations, but I'm working with a question when trying to create and manage the database session. I have the following setup: ```python from fastapi import FastAPI, Depends from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker, declarative_base DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname" engine = create_async_engine(DATABASE_URL, echo=True) Base = declarative_base() AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False) app = FastAPI() async def get_db(): async with AsyncSessionLocal() as session: yield session @app.post("/items/") async def create_item(item: Item, db: AsyncSession = Depends(get_db)): db_item = ItemModel(**item.dict()) db.add(db_item) await db.commit() await db.refresh(db_item) return db_item ``` However, I'm getting an behavior that says `RuntimeError: This event loop is already running`. This happens when I make requests to create items. I suspect it's due to the event loop being handled incorrectly, but I'm unsure how to resolve this scenario. I've tried ensuring that I'm using the latest versions of FastAPI (0.68.0), SQLAlchemy (1.4.22), and asyncpg (0.24.0), but the question continues. Also, I've checked that I have the proper database setup, and it's working with synchronous queries. I would appreciate any advice on how to properly manage the session in an async context or if there are additional configurations I might need to adjust. Thanks! For context: I'm using Python on Ubuntu 22.04. Thanks for any help you can provide!