SimpleFastCrud

SimpleFastCrud is a powerful tool that automatically generates complete CRUD (Create, Read, Update, Delete) endpoints for your FastAPI applications using SQLAlchemy models. It eliminates boilerplate code and provides advanced features like filtering, pagination, multi-tenancy support, and relationship handling.

Features

• 🚀 Automatic CRUD endpoint generation
• 🔍 Advanced filtering and search capabilities
• 📄 Built-in pagination support
• 🏢 Multi-tenant application support
• 🔐 Authentication and authorization integration
• 🔗 Automatic relationship handling with eager loading
• 📊 Customizable query parameters

Installation

pip install SimpleFastCrud

Requirements

• FastAPI
• SQLAlchemy
• Pydantic

Quick Start

Basic Setup

from fastapi import APIRouter, FastAPI
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from SimpleFastCrud import SimpleFastCrud

# Database setup
Base = declarative_base()
engine = create_engine('sqlite:///./test.db')
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Database dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# FastAPI setup
app = FastAPI()
api_router = APIRouter()

# Define your model
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

# Create tables
Base.metadata.create_all(bind=engine)

# Initialize SimpleFastCrud and add your model
crud = SimpleFastCrud(api_router=api_router, get_db=get_db)
crud.add(model=User)

# Include the router in your app
app.include_router(api_router)

This simple setup automatically generates the following endpoints:

• GET /users - List all users (with pagination and filtering)
• GET /users/{id} - Get a specific user by ID
• POST /users - Create a new user
• PUT /users/{id} - Update an existing user
• DELETE /users/{id} - Delete a user

Advanced Features

1. Authentication and Authorization (auth_dep)

Add authentication to your CRUD endpoints by providing an authentication dependency. This ensures that only authenticated users can access the endpoints.

from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """
    Verify the authentication token and return user information.
    This is a simplified example - implement your own authentication logic.
    """
    token = credentials.credentials
    # Verify token here (JWT, OAuth, etc.)
    # For this example, we'll return a mock user
    if token != "valid_token":
        raise HTTPException(status_code=401, detail="Invalid authentication")

    return {
        "user_id": 1,
        "username": "john_doe",
        "email": "john@example.com"
    }

# Add CRUD with authentication
crud.add(
    model=User,
    auth_dep=Depends(get_current_user)
)

What it does:

• All endpoints will require authentication
• The authenticated user information will be available in the endpoint handlers
• Unauthenticated requests will receive a 401 error

2. Multi-Tenant Filtering (filter_param)

Perfect for SaaS applications where data must be isolated per tenant/organization. This parameter automatically filters all queries based on a field in your model.

from sqlalchemy import Column, Integer, String, ForeignKey

class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Integer)
    tenant_id = Column(Integer, index=True)  # The tenant identifier

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    # Your authentication logic
    return {
        "user_id": 1,
        "tenant_id": 42,  # The tenant this user belongs to
        "username": "john_doe"
    }

# Add CRUD with multi-tenant support
crud.add(
    model=Product,
    auth_dep=Depends(get_current_user),
    filter_param='tenant_id'  # Automatically filter by tenant_id
)

What it does:

• GET requests: Only returns records where tenant_id matches the authenticated user's tenant_id
• POST requests: Automatically sets the tenant_id from the authenticated user
• PUT/DELETE requests: Only allows modifications to records belonging to the user's tenant
• Prevents data leakage between tenants

Example:

• User from Tenant A cannot see, modify, or delete products from Tenant B
• When creating a product, the tenant_id is automatically set from the user's token

3. Search Functionality (filter_query_search)

Enable text search on a specific field. Users can search using the search query parameter.

class Article(Base):
    __tablename__ = 'articles'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    content = Column(String)
    author = Column(String)

crud.add(
    model=Article,
    filter_query_search='title'  # Enable search on the title field
)

How to use:

# Search for articles with "Python" in the title
GET /articles?search=Python

# The search is case-insensitive and uses LIKE %search%

What it does:

• Adds a search query parameter to the GET endpoint
• Performs case-insensitive partial matching (ILIKE)
• Example: search=python will find "Introduction to Python", "Python Basics", etc.

4. Advanced Filtering (filter_fields)

Create custom filters with different comparison operators. Users can filter results using query parameters.

from sqlalchemy import Column, Integer, String, Date, Float

class Order(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True)
    customer_name = Column(String)
    total = Column(Float)
    status = Column(String)
    created_at = Column(Date)

crud.add(
    model=Order,
    filter_fields={
        'status': {
            'comparison': 'eq',  # Equal comparison
            'type': 'string',
            'description': 'Filter by order status (pending, completed, cancelled)'
        },
        'total': {
            'comparison': 'ge',  # Greater than or equal
            'type': 'number',
            'description': 'Minimum order total'
        },
        'created_at': {
            'comparison': 'between',  # Date range
            'type': 'date',
            'description': 'Filter by creation date range'
        }
    }
)

Available comparison operators:

• eq: Equal to (=)
• ne: Not equal to (!=)
• lt: Less than (<)
• le: Less than or equal (<=)
• gt: Greater than (>)
• ge: Greater than or equal (>=)
• between: Range query (for dates and numbers)

How to use:

# Filter by status
GET /orders?status=completed

# Filter by minimum total
GET /orders?total=100

# Filter by date range (format: DD-MM-YYYY)
GET /orders?created_at_from=01-01-2024&created_at_to=31-12-2024

# Combine multiple filters
GET /orders?status=completed&total=100&created_at_from=01-01-2024

5. Pagination

Enable pagination to handle large datasets efficiently.

crud.add(
    model=User,
    pagination=True,  # Enable pagination
    steps=25          # Items per page (default: 10)
)

How to use:

# Get first page (25 items)
GET /users?page=1&per_page=25

# Get second page
GET /users?page=2&per_page=25

# Disable pagination for a specific request
GET /users?pagination=false

Response format with pagination:

{
    "data": [...],
    "message": "",
    "metadata": {
        "page": 1,
        "per_page": 25,
        "total": 150,
        "total_pages": 6
    }
}

Response format without pagination:

{
    "data": [...],
    "message": "",
    "metadata": {
        "total": 150
    }
}

6. Relationship Handling

By default, SimpleFastCrud automatically loads related models using eager loading (joinedload).

from sqlalchemy.orm import relationship

class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    books = relationship("Book", back_populates="author")

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    author_id = Column(Integer, ForeignKey('authors.id'))
    author = relationship("Author", back_populates="books")

# With relationships (default)
crud.add(
    model=Author,
    relationship=True  # Includes related books in the response
)

# Without relationships (faster for simple queries)
crud.add(
    model=Book,
    relationship=False  # Only returns book data, no author details
)

What it does:

• relationship=True: Uses SQLAlchemy's joinedload to fetch related data in a single query
• relationship=False: Only fetches the main model data (better performance if you don't need related data)

7. Custom Pydantic Schemas

Override the auto-generated schemas with your own Pydantic models.

from pydantic import BaseModel, EmailStr, validator

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

    @validator('name')
    def name_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError('Name cannot be empty')
        return v

crud.add(
    model=User,
    schema=UserCreate  # Use custom schema for input and output
)

Complete Example: E-commerce API

Here's a comprehensive example combining multiple features:

from fastapi import APIRouter, Depends, FastAPI, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy import Column, Integer, String, Float, Date, ForeignKey, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from SimpleFastCrud import SimpleFastCrud

# Database setup
Base = declarative_base()
engine = create_engine('sqlite:///./ecommerce.db')
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Authentication
security = HTTPBearer()

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    # Implement your JWT/OAuth validation here
    return {
        "user_id": 1,
        "tenant_id": 42,  # For multi-tenancy
        "username": "shop_owner"
    }

# Models
class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String, index=True)
    description = Column(String)
    price = Column(Float)
    stock = Column(Integer)
    category = Column(String)
    tenant_id = Column(Integer, index=True)  # Multi-tenant field
    orders = relationship("Order", back_populates="product")

class Order(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True)
    customer_name = Column(String)
    product_id = Column(Integer, ForeignKey('products.id'))
    quantity = Column(Integer)
    total = Column(Float)
    status = Column(String)
    created_at = Column(Date)
    tenant_id = Column(Integer, index=True)
    product = relationship("Product", back_populates="orders")

# Create tables
Base.metadata.create_all(bind=engine)

# FastAPI setup
app = FastAPI(title="E-commerce API")
api_router = APIRouter()

# Initialize CRUD
crud = SimpleFastCrud(api_router=api_router, get_db=get_db)

# Products CRUD with full features
crud.add(
    model=Product,
    auth_dep=Depends(get_current_user),    # Require authentication
    filter_param='tenant_id',               # Multi-tenant filtering
    filter_query_search='name',             # Search by product name
    filter_fields={                         # Advanced filtering
        'category': {
            'comparison': 'eq',
            'type': 'string',
            'description': 'Filter by category'
        },
        'price': {
            'comparison': 'le',
            'type': 'number',
            'description': 'Maximum price'
        }
    },
    pagination=True,                        # Enable pagination
    steps=20,                               # 20 items per page
    relationship=True                       # Include related orders
)

# Orders CRUD
crud.add(
    model=Order,
    auth_dep=Depends(get_current_user),
    filter_param='tenant_id',
    filter_query_search='customer_name',
    filter_fields={
        'status': {
            'comparison': 'eq',
            'type': 'string',
            'description': 'Filter by order status'
        },
        'created_at': {
            'comparison': 'between',
            'type': 'date',
            'description': 'Filter by date range'
        },
        'total': {
            'comparison': 'ge',
            'type': 'number',
            'description': 'Minimum order total'
        }
    },
    pagination=True,
    steps=50
)

app.include_router(api_router)

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

Example API calls:

# Get products with search and filters
GET /products?search=laptop&category=electronics&price=1000&page=1

# Get orders by date range and status
GET /orders?status=completed&created_at_from=01-01-2024&created_at_to=31-12-2024&page=1

# Create a new product (automatically sets tenant_id from authenticated user)
POST /products
{
    "name": "Gaming Laptop",
    "description": "High-performance laptop",
    "price": 1299.99,
    "stock": 10,
    "category": "electronics"
}

# Update a product (only if it belongs to your tenant)
PUT /products/1
{
    "price": 1199.99,
    "stock": 8
}

API Response Format

All endpoints return a consistent response format:

{
    "data": {},        // or [] for lists
    "message": "",     // Success or error message
    "metadata": {}     // Pagination info, counts, etc.
}

Parameters Reference

add() Method Parameters

Parameter	Type	Default	Description
model	SQLAlchemy Model	Required	The SQLAlchemy model class
auth_dep	Depends	None	Authentication dependency for securing endpoints
filter_param	str	None	Field name for multi-tenant filtering (e.g., 'tenant_id')
schema	Pydantic Model	None	Custom Pydantic schema (auto-generated if not provided)
dependencies	List[Depends]	[]	Additional FastAPI dependencies
relationship	bool	True	Enable/disable eager loading of relationships
pagination	bool	False	Enable pagination for GET all endpoint
steps	int	10	Default number of items per page
filter_query_search	str	None	Field name for text search functionality
filter_fields	dict	None	Advanced filtering configuration

Contributing

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

License

This project is licensed under the MIT License.

Support

If you encounter any issues or have questions, please open an issue on the GitHub repository.