fastapi-async-db-plugin 0.1.0

A production-grade async SQLAlchemy database plugin for FastAPI.

FastAPI Async DB Plugin

A reusable, boilerplate-free async database plugin for FastAPI using SQLAlchemy 2.0 and asyncpg.

Installation

pip install fastapi-async-db-plugin

Setup & Usage

Simply import the plugin functions into your FastAPI app and provide a .env file with DATABASE_URL:

DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/dbname

Example main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from fastapi_db_plugin import get_db_session, register_db_events
from fastapi_db_plugin.models_example import User

app = FastAPI(title="Reusable DB App Example")

# Automatically wire up connection pools and dispose them on shutdown.
register_db_events(app, create_all=True)

@app.post("/users/")
async def create_user(username: str, email: str, db: AsyncSession = Depends(get_db_session)):
    new_user = User(username=username, email=email)
    db.add(new_user)
    
    await db.commit()
    await db.refresh(new_user)
    
    return {"id": new_user.id, "username": new_user.username, "email": new_user.email}

@app.get("/users/")
async def list_users(db: AsyncSession = Depends(get_db_session)):
    result = await db.execute(select(User))
    users = result.scalars().all()
    return [{"id": u.id, "username": u.username, "email": u.email} for u in users]