common-crud 0.1.1

Common CRUD operations for SQLAlchemy and FastAPI

Common CRUD

A lightweight, flexible library for common CRUD (Create, Read, Update, Delete) operations with SQLAlchemy and FastAPI.

Overview

Common CRUD provides a clean, reusable base class (BaseCRUD) that handles standard database operations for SQLAlchemy models. It's designed to:

• Reduce boilerplate code for common database operations
• Provide consistent error handling and logging
• Support both simple and advanced filtering options
• Work seamlessly with SQLAlchemy's async functionality

Installation

pip install common-crud

Or with Poetry:

poetry add common-crud

Dependencies

• Python 3.11+
• SQLAlchemy 2.0+

Usage

Basic Setup

Create a CRUD class for your SQLAlchemy model:

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from common_crud import BaseCRUD

# Define your SQLAlchemy model
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String, unique=True)

# Create a CRUD class for your model
class UserCRUD(BaseCRUD):
    model = User
    # Optional: Define fields that should use ILIKE for case-insensitive search
    filter_fields = {"name": "ilike", "email": "ilike"}

Basic Operations

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

# Create engine and session
engine = create_async_engine("sqlite+aiosqlite:///database.db")
async_session = async_sessionmaker(engine, expire_on_commit=False)

async def main():
    async with async_session() as session:
        # Create
        new_user = await UserCRUD.add(
            session=session, 
            name="John Doe",
            email="[e-mail]"
        )
        print(f"Added user: {new_user.id}")
        
        # Read by ID
        user = await UserCRUD.find_one_or_none_by_id(
            session=session,
            data_id=new_user.id
        )
        print(f"Found user: {user.name}")
        
        # Read with filter
        users = await UserCRUD.find_all(
            session=session, 
            name="John"  # Will use ILIKE since it's in filter_fields
        )
        print(f"Found {len(users)} users")
        
        # Update
        rows_updated = await UserCRUD.update(
            session=session,
            filter_by={"id": new_user.id},
            name="John Smith"
        )
        print(f"Updated {rows_updated} rows")
        
        # Delete
        rows_deleted = await UserCRUD.delete(
            session=session,
            id=new_user.id
        )
        print(f"Deleted {rows_deleted} rows")

# Run the async function
asyncio.run(main())

Advanced Features

Time Range Filtering

Filter records based on a date/time field:

# Find all records created between two dates
records = await MyCRUD.find_all_in_time_range(
    session=session,
    start_time=datetime(2023, 1, 1),
    end_time=datetime(2023, 12, 31),
    param="created_at"  # Default field name
)

Bulk Insert

Efficiently insert multiple records:

rows_inserted = await UserCRUD.bulk_insert(
    session=session,
    data=[
        {"name": "User 1", "email": "[e-mail]"},
        {"name": "User 2", "email": "[e-mail]"},
        {"name": "User 3", "email": "[e-mail]"}
    ]
)

Error Handling

The library provides specialized exceptions:

from common_crud.exceptions import CRUDError, ModelNotSetError, InvalidFilterError, DatabaseError

try:
    await UserCRUD.update(session=session, filter_by={}, name="New Name")
except InvalidFilterError:
    print("Need to provide filter criteria for update")
except DatabaseError as e:
    print(f"Database error: {e}")
except CRUDError as e:
    print(f"Generic error: {e}")

FastAPI Integration

Works seamlessly with FastAPI:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_session  # Your session dependency
from models import User, UserCRUD
from pydantic import BaseModel

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    
    class Config:
        orm_mode = True

@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate, session: AsyncSession = Depends(get_session)):
    try:
        db_user = await UserCRUD.add(session=session, **user.dict())
        return db_user
    except DatabaseError as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, session: AsyncSession = Depends(get_session)):
    user = await UserCRUD.find_one_or_none_by_id(session=session, data_id=user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

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.