A lightweight, flexible, and type-safe repository pattern implementation for SQLAlchemy Async, featuring seamless integration with Pydantic, Dataclasses, and standard Python objects.
- 🚀 Async-first design
- 🔒 Type-safe CRUD operations
- ⚗ Robust SQLAlchemy Async support
- 📦 Out-of-the-box support for Pydantic (v2), Dataclasses, and Dicts
- 🛠 Extensible mapping logic (
to_repr/to_inner) - 📝 Full type hints support
- 🛡 Attribute protection via Metaclasses
pip install simple-repo-asyncsqlaNow you must control converting proccess
class UserRepository(crud_factory(UserSQLA, UserDomain, UserCreate, UserPatch)):
# Define your custom methods here
def to_repr(self, obj: SA) -> DM:
return self.domain_model.model_validate(obj)
def to_inner(self, data: CS | DM | PS) -> dict:
return data.model_dump(exclude_unset=True)Use dict as domain model
class UserRepository(crud_factory(UserSQLA, dict, UserCreate, UserPatch)):
# Define your custom methods here
def to_repr(self, obj: SA) -> DM:
return {
"id": obj.id,
"name": obj.name,
"email": obj.email,
"is_active": obj.is_active
# your fileds
}
def to_inner(self, data: CS | DM | PS) -> dict:
if isinstance(data, dict):
return data
return data.model_dump(exclude_unset=True)Or use SQLAlchemy model as domain model
class UserRepository(crud_factory(UserSQLA, UserSQLA, UserCreate, UserPatch)):
# Define your custom methods here
def to_repr(self, obj: SA) -> DM:
return obj
def to_inner(self, data: CS | DM | PS) -> dict:
if isinstance(data, UserSQLA):
return {
"id": data.id,
"name": data.name,
"email": data.email,
"is_active": data.is_active
# your fileds
}
return data.model_dump(exclude_unset=True)Or dict for any schema
class UserRepository(crud_factory(UserSQLA, dict, dict, dict)):
# Define your custom methods here
def to_repr(self, obj: SA) -> DM:
return {
"id": obj.id,
"name": obj.name,
"email": obj.email,
"is_active": obj.is_active
# your fileds
}
def to_inner(self, data: CS | DM | PS) -> dict:
return datafrom sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from pydantic import BaseModel, ConfigDict
from simple_repository import crud_factory
from simple_repository.exceptions import NotFoundException
# Define SQLAlchemy model
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
name: Mapped[str]
email: Mapped[str]
is_active: Mapped[bool] = mapped_column(default=True)
# Define domain model
class UserDomain(BaseModel):
id: int
name: str
email: str
is_active: bool
model_config = ConfigDict(from_attributes=True)
# Define schemas for create/update operations
class UserCreate(BaseModel):
name: str
email: str
is_active: bool | None = None
class UserPatch(BaseModel):
name: str | None = None
email: str | None = None
is_active: bool | None = None
# Setup async database
engine = create_async_engine("sqlite+aiosqlite:///./db.sqlite3")
async_session_factory = async_sessionmaker(engine, expire_on_commit=False)
# Create repository class
class UserRepository(crud_factory(User, UserDomain, UserCreate, UserPatch)):
def to_repr(self, obj: SA) -> DM:
return self.domain_model.model_validate(obj)
def to_inner(self, data: CS | DM | PS) -> dict:
return data.model_dump(exclude_unset=True)
# Use in your async application
async def main():
user_repo = UserRepository()
async with async_session_factory() as session:
# Create - single record
new_user = await user_repo.create(
session,
UserCreate(name="John Doe", email="john@example.com")
)
print(f"Created user: {new_user}")
# Create many - multiple records
users_data = [
UserCreate(name="Jane Doe", email="jane@example.com"),
UserCreate(name="Bob Smith", email="bob@example.com"),
]
created_users = await user_repo.create_many(session, users_data)
print(f"Created {len(created_users)} users")
# Get one - by primary key
user = await user_repo.get_one(session, new_user.id)
print(f"Retrieved user: {user}")
# Get many - multiple records by IDs
user_ids = [new_user.id, created_users[0].id]
many_users = await user_repo.get_many(session, user_ids)
print(f"Retrieved {len(many_users)} users by IDs")
# Get all - with pagination
all_users, total = await user_repo.get_all(
session,
offset=0,
limit=10,
order_by="name"
)
print(f"Total users: {total}, Retrieved: {len(all_users)}")
# Count - get total count
count = await user_repo.count(session)
print(f"Total count: {count}")
# Update - replace entire record
user.name = "John Smith"
updated_user = await user_repo.update(session, user)
print(f"Updated user: {updated_user}")
# Patch - partial update
patch_data = UserPatch(email="john.smith@example.com")
patched_user = await user_repo.patch(session, patch_data, new_user.id)
print(f"Patched user: {patched_user}")
# Remove - delete single record
await user_repo.remove(session, created_users[1].id)
print(f"Removed 1 user")
# Remove many - delete multiple records
remove_ids = [created_users[0].id]
removed_count = await user_repo.remove_many(session, remove_ids)
print(f"Removed {removed_count} users")
# Handle not found exception
try:
await user_repo.get_one(session, 9999)
except NotFoundException:
print("User not found (expected)")Extend the base repository with advanced query methods:
from sqlalchemy import select, case, func, text
from sqlalchemy.ext.asyncio import AsyncSession
from simple_repository import crud_factory
from simple_repository.abctract import IAsyncCrud
from .models.user import User as UserSQLA
from .domains.user import UserDomain
from .schemes.user import UserCreate, UserPatch
from .db import async_session_factory
# Define interface for custom operations OR don't
class IUserRepository(IAsyncCrud[UserSQLA, UserDomain, UserCreate, UserPatch]):
@abstractmethod
async def get_user_activity_stats(
self,
session: AsyncSession,
min_orders: int = 5,
days_window: int = 30
) -> list[dict]:
pass
# Inherit from the interface (if defined) and from the class created by the factory
class UserRepository(IUserRepository, crud_factory(UserSQLA, UserDomain, UserCreate, UserPatch)):
def to_inner(self, data: CS | DM | PS) -> dict:
return data.model_dump(exclude_unset=True)
def to_repr(self, object: SA) -> DM:
return self.domain_model.model_validate(object)
"""Custom repository with advanced analytical capabilities."""
async def get_user_activity_stats(
self,
session: AsyncSession,
min_orders: int = 5,
days_window: int = 30
) -> list[dict]:
current_date = func.current_timestamp()
window_date = current_date - text(f"interval '{days_window} days'")
orders_stats = (
select(
Order.user_id,
func.count().label('order_count'),
func.sum(Order.total_amount).label('total_spent'),
func.avg(Order.total_amount).label('avg_order_value'),
func.count(case(
(Order.created_at > window_date, 1)
)).label('recent_orders'),
(func.max(Order.created_at) - func.min(Order.created_at)) /
func.nullif(func.count() - 1, 0)
.label('avg_order_interval')
)
.group_by(Order.user_id)
.having(func.count() >= min_orders)
.alias('orders_stats')
)
query = (
select(
self.sqla_model.id,
self.sqla_model.name,
self.sqla_model.email,
orders_stats.c.order_count,
orders_stats.c.total_spent,
orders_stats.c.avg_order_value,
orders_stats.c.recent_orders,
orders_stats.c.avg_order_interval,
(
orders_stats.c.recent_orders * 0.4 +
func.least(orders_stats.c.total_spent / 1000, 10) * 0.3 +
(orders_stats.c.order_count * 0.3)
).label('engagement_score'),
func.percent_rank().over(
order_by=orders_stats.c.total_spent
).label('spending_percentile')
)
.join(orders_stats, self.sqla_model.id == orders_stats.c.user_id)
.where(self.sqla_model.is_active == True)
.order_by(text('engagement_score DESC'))
)
result = await session.execute(query)
return list(result.mappings().all())
# Usage example
async def analyze_user_activity():
user_repo_instance = UserRepository()
with async_session_factory() as session:
stats = await user_repo_instance.get_user_activity_stats(
session,
min_orders=5,
days_window=30
)
return stats
from fastapi import HTTPException, APIRouter
from simple_repository.exceptions import NotFoundException
from sqlalchemy.ext.asyncio import AsyncSession
from .db import async_session_factory
from .my_repository import user_crud
from .domains.user import UserDomain
router = APIRouter("/user")
@router.get("/{user_id}")
async def get_user(
session: Annotated[AsyncSession, Depends(async_session_factory)],
user_id: int,
) -> UserDomain:
try:
return await user_crud.get_one(session, user_id)
except NotFoundException:
raise HTTPException(status_code=404, detail="User not found")- create
- create_many
- get_one
- get_many
- get_all
- patch
- update
- remove
- remove_many
- count
The library now uses the FrozenClassAttributesMeta metaclass to prevent accidental or unintended reassignment of sqla_model and domain_model attributes at the class level after they have been set by crud_factory. This ensures the stability and predictability of repository behavior.
Attempting to modify these attributes after class creation will result in an AttributeError.
# Example:
MyUserRepoClass = crud_factory(UserSQLA, UserDomain)
try:
MyUserRepoClass.sqla_model = SomeOtherSQLA # This will raise an AttributeError
MyUserRepoClass.domain_model = SomeOtherDomain # This also will raise an AttributeError
except AttributeError as e:
print(f"Error: {e}")Pull requests are welcome! For major changes, please open an issue first to discuss what you would like to change.
This project is licensed under the MIT License.