sqladbx 0.0.4

Modern async SQLAlchemy database context manager for FastAPI, Taskiq, Temporal, and more

sqladbx 🚀

Modern async SQLAlchemy database context manager for FastAPI, Litestar, Taskiq, Temporal, and more.

💡 Inspired by fastapi-async-sqlalchemy - reimagined with modern async patterns, ContextVar-based session management, and extended framework support.

✨ Features

• 🎯 Simple API - Clean, intuitive interface for database operations
• ⚡ Async-first - Built for async/await with SQLAlchemy 2.0+
• 🔄 Auto Context Management - Automatic session lifecycle with middleware
• 🌐 Framework Agnostic - Use with web frameworks, CLI scripts, background tasks, or any async Python code
• 🎭 Multi-Session Support - Handle multiple concurrent sessions when needed
• 🔒 Type Safe - Full type hints and mypy support
• 🧪 Well Tested - Comprehensive test coverage

📦 Installation

pip install sqladbx

🚀 Quick Start

FastAPI Example

from fastapi import FastAPI
from sqlmodel import Field, SQLModel
from sqladbx import SQLAlchemyMiddleware, db

# Define your model
class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

# Create FastAPI app
app = FastAPI()

# Add SQLAlchemy middleware
app.add_middleware(
    SQLAlchemyMiddleware,
    db_url="postgresql+asyncpg://user:password@localhost/dbname"
)

# Use db.session in your endpoints - no context manager needed!
@app.post("/users")
async def create_user(name: str, email: str):
    user = User(name=name, email=email)
    db.session.add(user)
    await db.session.commit()
    await db.session.refresh(user)
    return user

@app.get("/users")
async def list_users():
    result = await db.session.scalars(select(User))
    return result.all()

Without Middleware (Taskiq, Temporal, etc.)

from sqladbx import db

# Initialize once at startup
db.initialize(db_url="postgresql+asyncpg://user:pass@localhost/dbname")

# Use in your code
async def process_data():
    async with db(commit_on_exit=True):
        result = await db.session.scalars(select(User))
        users = result.all()
        # Process users...

# Cleanup on shutdown (optional but recommended)
await db.dispose()

Multi-Session Mode (Concurrent Queries)

import asyncio
from sqladbx import db

async def query1():
    result = await db.session.execute(select(User))
    return result.scalars().all()

async def query2():
    result = await db.session.execute(select(Post))
    return result.scalars().all()

# Run queries concurrently
async with db(multi_sessions=True):
    users, posts = await asyncio.gather(query1(), query2())

Multiple Databases

from sqladbx import create_middleware_and_db

# Create db instances and middlewares
MainMiddleware, main_db = create_middleware_and_db()
ReplicaMiddleware, replica_db = create_middleware_and_db()

# Add to app
app.add_middleware(MainMiddleware, db_url="postgresql://main")
app.add_middleware(ReplicaMiddleware, db_url="postgresql://replica")

@app.get("/users")
async def list_users():
    # Use main_db for writes
    result = await main_db.session.scalars(select(User))
    return result.all()

@app.get("/stats")
async def get_stats():
    # Use replica_db for reads
    result = await replica_db.session.scalars(select(Stats))
    return result.all()

More Examples

See the examples/ directory for complete working examples:

• examples/fastapi_basic.py
• examples/fastapi_multi_db.py
• examples/litestar_basic.py
• examples/litestar_multi_db.py
• examples/taskiq_basic.py
• examples/taskiq_multi_sessions.py

🎓 Best Practices

✅ DO

• Use middleware for web applications (automatic session management)
• Use manual context (async with db()) for CLI/background tasks
• Enable commit_on_exit=True for simple CRUD operations
• Use separate db proxies for master/replica setups
• Implement proper error handling with try/except

❌ DON'T

• Don't mix middleware and manual initialization
• Don't create sessions manually - use db.session
• Don't use blocking I/O inside database contexts
• Don't share sessions between requests

🤝 Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🔗 Links

• PyPI: https://pypi.org/project/sqladbx/
• GitHub: https://github.com/swaldom/sqladbx
• Documentation: https://github.com/swaldom/sqladbx#readme
• Issues: https://github.com/swaldom/sqladbx/issues

💬 Support

If you have any questions or need help, please:

• Open an issue on GitHub
• Check existing issues and discussions
• Read the documentation carefully

---

Made with ❤️ by Oleksii Svichkar