Metadata-Version: 2.4
Name: fastapi-async-db-plugin
Version: 0.1.0
Summary: A production-grade async SQLAlchemy database plugin for FastAPI.
Author-email: vaibhav734 <vaibhav734@users.noreply.github.com>
Project-URL: Homepage, https://github.com/vaibhav734/astapi-async-db-plugin
Project-URL: Bug Tracker, https://github.com/vaibhav734/astapi-async-db-plugin/issues
Classifier: Programming Language :: Python :: 3
Classifier: Framework :: FastAPI
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Topic :: Database
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: fastapi>=0.100.0
Requires-Dist: sqlalchemy[asyncio]>=2.0.0
Requires-Dist: asyncpg>=0.28.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: pydantic-settings>=2.0.0
Dynamic: license-file

# FastAPI Async DB Plugin

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

## Installation

```bash
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`:

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

### Example `main.py`
```python
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]
```
