PyAsyncDB

Modern asynchronous database library for Python with ORM, query builder, connection pooling, migrations, and more.

✨ Features

• ✅ Async/Await - Full async support for PostgreSQL, MySQL, SQLite
• ✅ ORM - Django-like model definitions with relationships
• ✅ Query Builder - Fluent SQL builder
• ✅ Connection Pooling - Efficient connection management
• ✅ Migrations - Schema version control with CLI
• ✅ Transactions - ACID transactions with nested savepoints
• ✅ Query Logging - Slow query detection and profiling
• ✅ Batch Operations - Bulk insert/update/delete
• ✅ Pydantic Integration - Automatic validation and serialization
• ✅ CLI Tool - Database management from command line
• ✅ Type Hints - Full type annotations

📦 Installation

pip install pyasyncdb

For Pydantic support:

pip install pyasyncdb[pydantic]

For development:

pip install pyasyncdb[dev]

🚀 Quick Start

Basic Usage

import asyncio
from asyncdb import Database

async def main():
    db = Database("postgresql://user:pass@localhost/mydb")
    
    async with db:
        # Create table
        await db.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id SERIAL PRIMARY KEY,
                name VARCHAR(100),
                email VARCHAR(255) UNIQUE
            )
        """)
        
        # Insert
        await db.execute(
            "INSERT INTO users (name, email) VALUES ($1, $2)",
            ("John", "john@example.com")
        )
        
        # Select
        users = await db.fetch_all("SELECT * FROM users")
        print(users)

asyncio.run(main())

Transactions

from asyncdb import Database

db = Database("postgresql://localhost/mydb")

async with db:
    async with db.transaction():
        # All queries in transaction
        await db.execute("INSERT INTO users (name) VALUES ($1)", ("Alice",))
        await db.execute("UPDATE accounts SET balance = balance - 100 WHERE id = $1", (1,))
        await db.execute("UPDATE accounts SET balance = balance + 100 WHERE id = $2", (2,))
        # Automatically commits on success, rollbacks on error

ORM with Validation

from asyncdb import Database, Model, Field, PydanticModelMixin

db = Database("postgresql://localhost/mydb")

class User(Model, PydanticModelMixin):
    __tablename__ = "users"
    
    id = Field(primary_key=True, auto_increment=True)
    name = Field(type="VARCHAR", max_length=100, nullable=False)
    email = Field(type="VARCHAR", max_length=255, unique=True)
    age = Field(type="INTEGER", default=0)

async with db:
    await User.create_table(db)
    
    # Create instance
    user = User(name="Alice", email="alice@example.com", age=25)
    await user.insert(db)
    
    # Query with filters
    users = await User.objects().filter(age__gte=18).all(db)
    
    # Generate Pydantic schema
    UserSchema = User.pydantic_model()
    validated = UserSchema(name="Bob", email="bob@example.com")

Query Builder

from asyncdb import select, insert, update, delete

# SELECT with joins
q = (
    select("users")
    .columns("id", "name", "email")
    .left_join("orders", "users.id = orders.user_id")
    .where("age > $1", 18)
    .and_where("status = $2", "active")
    .order_by("name")
    .limit(10)
)
sql, params = q.build()

# INSERT
q = insert("users").columns("name", "email").values("John", "john@example.com")
sql, params = q.build()

# UPDATE
q = update("users").set("email", "new@example.com").where("id = $1", 1)
sql, params = q.build()

# DELETE
q = delete("users").where("id = $1", 1)
sql, params = q.build()

Batch Operations

from asyncdb import Database, Model, Field

db = Database("postgresql://localhost/mydb")

class User(Model):
    __tablename__ = "users"
    name = Field(type="VARCHAR", max_length=100)
    email = Field(type="VARCHAR", max_length=255)

async with db:
    # Bulk insert
    users = [
        User(name="Alice", email="alice@example.com"),
        User(name="Bob", email="bob@example.com"),
        User(name="Charlie", email="charlie@example.com"),
    ]
    await User.bulk_insert(db, users)
    
    # Bulk update
    for u in users:
        u.name = u.name.upper()
    await User.bulk_update(db, users)
    
    # Bulk delete
    await User.bulk_delete(db, [1, 2, 3])

Query Logging & Debugging

from asyncdb import Database
import logging

logging.basicConfig(level=logging.INFO)

# Enable query logging
db = Database("postgresql://localhost/mydb", echo=True, slow_query_threshold=0.5)

async with db:
    await db.execute("SELECT * FROM users")
    await db.execute("SELECT * FROM orders WHERE amount > 1000")  # Slow query
    
    # Get slow queries
    slow = db.get_slow_queries()
    print(f"Slow queries: {len(slow)}")
    
    # Get all query log
    log = db.get_query_log()
    for entry in log:
        print(f"{entry['duration']:.3f}s - {entry['query']}")

🛠 CLI Commands

# Set database URL
export DATABASE_URL="postgresql://user:pass@localhost/mydb"

# Run migrations
pyasyncdb migrate-up

# Rollback migrations
pyasyncdb migrate-down --steps 2

# Check migration status
pyasyncdb migrate-status

# Create new migration
pyasyncdb migrate-create add_users_table

# Interactive shell
pyasyncdb shell

# Inspect database
pyasyncdb inspect

📊 Database URLs

postgresql://user:pass@host:5432/dbname
mysql://user:pass@host:3306/dbname
sqlite:///path/to/database.db

📖 API Reference

Database

Model

QuerySet

📝 License

MIT License - see LICENSE for details.

🤝 Contributing

Contributions welcome! Please read our contributing guidelines first.

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

📧 Support

• Documentation: https://github.com/yourusername/asyncdb#readme
• Issues: https://github.com/yourusername/asyncdb/issues
• PyPI: https://pypi.org/project/pyasyncdb/