Skip to main content

​
Overview

The Repository class is the heart of FluxCRUD’s data layer. It wraps SQLAlchemy operations into a clean, type-safe API that handles:
  • Type Validation: Using Pydantic schemas.
  • Caching: Automatic read-through and write-through caching.
  • Batching: Optimized database writes.
  • N+1 Prevention: Optional DataLoader support (opt-in).

​
Basic Usage

By default, the Repository uses direct SQL queries for maximum performance. 
# Create
repo = Repository(session, model=User, schema=UserSchema)
user = await repo.create(UserCreate(name="Alice"))

# Read (Direct SQL)
user = await repo.get(1)

​
Enabling DataLoader (N+1 Protection)

When resolving relationships in loops (e.g., GraphQL or nested API responses), enable the DataLoader to batch queries automatically. 
repo = Repository(..., use_loader=True)
# Queries will be batched
user = await repo.get(1)

​
Advanced Features

​
Caching

When CacheManager is provided, get() calls check the cache first. 
repo = Repository(..., cache_manager=redis_cache)
# This will hit Redis first, then DB if missing
user = await repo.get(1)

​
Batching

Use the batch_writer context manager for high-performance inserts. 
async with repo.batch_writer() as writer:
    for item in big_list:
        await writer.add(item)

​
Learn More