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: Integrated DataLoader support.

​
Basic Usage

repo = Repository(model=User, schema=UserSchema)

# Find by ID
user = await repo.get(session, 1)

# Create
user = await repo.create(session, UserCreate(name="Alice"))

​
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(session, 1)

​
Batching

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