This document outlines proposed enhancements for v0.2.0 of sqlmodel-crud-utils. The focus is on improving developer experience, adding production-ready features, and maintaining backward compatibility while expanding functionality.
- ✅ Comprehensive CRUD operations (sync & async)
- ✅ 100% test coverage (56 tests)
- ✅ Type checking enabled and passing
- ✅ Good separation of sync/async implementations
- ✅ Flexible filtering and pagination
- ✅ Relationship loading support
- ✅ Pre-commit hooks configured
- ✅ CI/CD with GitHub Actions
- ❌ No public API exports (empty
__init__.py) - ❌ No CHANGELOG.md file
- ❌ Limited transaction management
- ❌ No custom exception hierarchy
- ❌ No lifecycle hooks (pre/post operations)
- ❌ No audit trail helpers (created_at, updated_at)
- ❌ No soft delete support
- ❌ Minimal inline documentation
- ❌ No query builder interface
Motivation: Users should be able to import commonly used functions directly from the package.
Implementation:
# sqlmodel_crud_utils/__init__.py
"""SQLModel CRUD Utilities - Simplified database operations for SQLModel."""
__version__ = "0.2.0"
# Sync exports
from sqlmodel_crud_utils.sync import (
delete_row,
get_one_or_create,
get_row,
get_rows,
get_rows_within_id_list,
insert_data_rows,
update_row,
write_row,
bulk_upsert_mappings,
)
# Async exports (with 'a_' prefix to avoid conflicts)
from sqlmodel_crud_utils.a_sync import (
delete_row as a_delete_row,
get_one_or_create as a_get_one_or_create,
get_row as a_get_row,
get_rows as a_get_rows,
get_rows_within_id_list as a_get_rows_within_id_list,
insert_data_rows as a_insert_data_rows,
update_row as a_update_row,
write_row as a_write_row,
bulk_upsert_mappings as a_bulk_upsert_mappings,
)
__all__ = [
"__version__",
# Sync
"delete_row",
"get_one_or_create",
"get_row",
"get_rows",
"get_rows_within_id_list",
"insert_data_rows",
"update_row",
"write_row",
"bulk_upsert_mappings",
# Async
"a_delete_row",
"a_get_one_or_create",
"a_get_row",
"a_get_rows",
"a_get_rows_within_id_list",
"a_insert_data_rows",
"a_update_row",
"a_write_row",
"a_bulk_upsert_mappings",
]Testing: Add tests to verify imports work correctly.
Motivation: Better error handling and debugging for users.
Design:
# sqlmodel_crud_utils/exceptions.py
class SQLModelCRUDError(Exception):
"""Base exception for all sqlmodel-crud-utils errors."""
pass
class RecordNotFoundError(SQLModelCRUDError):
"""Raised when a requested record is not found."""
def __init__(self, model: type, id_value: any, pk_field: str = "id"):
self.model = model
self.id_value = id_value
self.pk_field = pk_field
super().__init__(
f"{model.__name__} with {pk_field}={id_value} not found"
)
class MultipleRecordsError(SQLModelCRUDError):
"""Raised when multiple records found where one expected."""
def __init__(self, model: type, count: int):
self.model = model
self.count = count
super().__init__(
f"Expected 1 {model.__name__}, found {count}"
)
class ValidationError(SQLModelCRUDError):
"""Raised when data validation fails."""
pass
class BulkOperationError(SQLModelCRUDError):
"""Raised when bulk operations fail."""
def __init__(self, total: int, failed: int, errors: list):
self.total = total
self.failed = failed
self.errors = errors
super().__init__(
f"Bulk operation failed: {failed}/{total} records failed"
)
class TransactionError(SQLModelCRUDError):
"""Raised when transaction operations fail."""
passImplementation:
- Modify existing functions to raise these exceptions
- Add
raise_on_errorparameter (default False for backward compatibility) - Update all error handling to use new exceptions
Motivation: Track changes across versions for transparency.
Implementation:
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [0.2.0] - TBD
### Added
- Public API exports in `__init__.py` for easier imports
- Custom exception hierarchy for better error handling
- CHANGELOG.md file for version tracking
- Transaction context managers for safer operations
- Audit trail support with `AuditMixin`
- Soft delete support with `SoftDeleteMixin`
- Lifecycle hooks (pre/post operations)
- Enhanced type hints throughout codebase
### Changed
- Improved error messages with detailed context
- Enhanced documentation with more examples
### Fixed
- Type checking errors with modern Python type hints
- `write_row` signature (was Type[SQLModel], now SQLModel instance)
- Logger type compatibility with optional loguru dependency
- Deprecated `session.execute()` replaced with `session.exec()`
### Security
- None
## [0.1.0] - 2024-XX-XX
### Added
- Initial release with sync and async CRUD operations
- Support for filtering, pagination, and relationship loading
- Bulk upsert operations
- 100% test coverageMotivation: Simplify transaction management and ensure proper rollback.
Design:
# sqlmodel_crud_utils/transactions.py
from contextlib import contextmanager, asynccontextmanager
from typing import Generator, AsyncGenerator
from sqlmodel import Session
from sqlmodel.ext.asyncio.session import AsyncSession
from .exceptions import TransactionError
@contextmanager
def transaction(session: Session) -> Generator[Session, None, None]:
"""
Context manager for handling transactions with automatic rollback.
Usage:
with transaction(session) as tx:
write_row(data, tx)
update_row(id, data, tx)
# Automatically commits on success, rolls back on error
"""
try:
yield session
session.commit()
except Exception as e:
session.rollback()
raise TransactionError(f"Transaction failed: {e}") from e
@asynccontextmanager
async def a_transaction(
session: AsyncSession
) -> AsyncGenerator[AsyncSession, None]:
"""
Async context manager for handling transactions.
Usage:
async with a_transaction(session) as tx:
await a_write_row(data, tx)
await a_update_row(id, data, tx)
"""
try:
yield session
await session.commit()
except Exception as e:
await session.rollback()
raise TransactionError(f"Transaction failed: {e}") from eMotivation: Common requirement for tracking record creation and updates.
Design:
# sqlmodel_crud_utils/mixins.py
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel
class AuditMixin:
"""
Mixin for automatic audit trail tracking.
Usage:
class MyModel(SQLModel, AuditMixin, table=True):
id: int = Field(primary_key=True)
name: str
"""
created_at: datetime = Field(
default_factory=datetime.utcnow,
nullable=False,
index=True,
)
updated_at: Optional[datetime] = Field(
default=None,
sa_column_kwargs={"onupdate": datetime.utcnow},
)
created_by: Optional[str] = Field(default=None, max_length=100)
updated_by: Optional[str] = Field(default=None, max_length=100)
class SoftDeleteMixin:
"""
Mixin for soft delete functionality.
Usage:
class MyModel(SQLModel, SoftDeleteMixin, table=True):
id: int = Field(primary_key=True)
name: str
"""
deleted_at: Optional[datetime] = Field(default=None, index=True)
deleted_by: Optional[str] = Field(default=None, max_length=100)
is_deleted: bool = Field(default=False, index=True)
def soft_delete(self, user: Optional[str] = None):
"""Mark record as deleted."""
self.is_deleted = True
self.deleted_at = datetime.utcnow()
self.deleted_by = user
def restore(self):
"""Restore soft-deleted record."""
self.is_deleted = False
self.deleted_at = None
self.deleted_by = NoneImplementation:
- Add helper functions for soft delete operations
- Modify
get_rowsto automatically exclude soft-deleted records (with override option)
Motivation: Allow users to inject custom logic before/after operations.
Design:
# sqlmodel_crud_utils/hooks.py
from typing import Protocol, TypeVar, Optional
from sqlmodel import SQLModel
T = TypeVar("T", bound=SQLModel)
class OperationHook(Protocol):
"""Protocol for operation hooks."""
def before_create(self, instance: T) -> T:
"""Called before creating a record."""
...
def after_create(self, instance: T) -> None:
"""Called after creating a record."""
...
def before_update(self, instance: T, data: dict) -> dict:
"""Called before updating a record."""
...
def after_update(self, instance: T) -> None:
"""Called after updating a record."""
...
def before_delete(self, instance: T) -> None:
"""Called before deleting a record."""
...
def after_delete(self, instance: T) -> None:
"""Called after deleting a record."""
...
# Global hook registry
_hooks: dict[type, list[OperationHook]] = {}
def register_hook(model: type[SQLModel], hook: OperationHook):
"""Register a hook for a specific model."""
if model not in _hooks:
_hooks[model] = []
_hooks[model].append(hook)
def get_hooks(model: type[SQLModel]) -> list[OperationHook]:
"""Get all hooks for a model."""
return _hooks.get(model, [])Motivation: Fluent interface for complex queries.
Design:
# sqlmodel_crud_utils/query_builder.py
class QueryBuilder:
"""Fluent interface for building queries."""
def __init__(self, session, model):
self.session = session
self.model = model
self._filters = []
self._order_by = []
self._limit = None
self._offset = None
def where(self, **kwargs):
"""Add WHERE conditions."""
self._filters.extend(kwargs.items())
return self
def order_by(self, field: str, desc: bool = False):
"""Add ORDER BY clause."""
self._order_by.append((field, desc))
return self
def limit(self, count: int):
"""Add LIMIT clause."""
self._limit = count
return self
def offset(self, count: int):
"""Add OFFSET clause."""
self._offset = count
return self
def all(self):
"""Execute query and return all results."""
# Implementation using existing get_rows
pass
def first(self):
"""Execute query and return first result."""
pass
def count(self):
"""Return count of matching records."""
passMotivation: Simplify connection management.
Design:
# sqlmodel_crud_utils/connection.py
from sqlmodel import create_engine
from sqlalchemy.pool import QueuePool
def create_pooled_engine(
database_url: str,
pool_size: int = 5,
max_overflow: int = 10,
pool_timeout: int = 30,
**kwargs
):
"""
Create a database engine with connection pooling.
Args:
database_url: Database connection URL
pool_size: Number of connections to maintain
max_overflow: Max connections beyond pool_size
pool_timeout: Timeout waiting for connection
"""
return create_engine(
database_url,
poolclass=QueuePool,
pool_size=pool_size,
max_overflow=max_overflow,
pool_timeout=pool_timeout,
**kwargs
)Motivation: Better user experience with comprehensive examples.
Implementation:
- Add detailed docstrings to all public functions
- Create usage examples for each feature
- Add troubleshooting guide
- Create migration guide from v0.1.0
- ✅ Fix type checking errors (COMPLETED)
- Implement public API exports (
__init__.py) - Create custom exception hierarchy
- Add CHANGELOG.md
- Update version to 0.2.0
- Implement transaction context managers
- Add audit trail mixins
- Add soft delete support
- Implement lifecycle hooks system
- Update all existing functions to use new features
- Write tests for all new features
- Update documentation
- Create migration guide
- Add usage examples
- Update README with new features
- Run full test suite
- Build and publish to PyPI
- Tag release on GitHub
- Announce release
NONE - v0.2.0 maintains full backward compatibility with v0.1.0.
All new features are opt-in:
- Exception handling:
raise_on_error=Falseby default - Hooks: Only active when registered
- Mixins: Only when inherited
- Transaction managers: Optional context managers
- All new exceptions with various scenarios
- Transaction context managers (success and failure)
- Audit mixins with datetime tracking
- Soft delete operations
- Lifecycle hooks execution
- Public API imports
- Benchmark bulk operations
- Test connection pooling efficiency
- Measure hook overhead
- Add "What's New in v0.2.0" section
- Update installation instructions
- Add new feature examples
- Update feature list
- CHANGELOG.md
- MIGRATION_GUIDE.md (v0.1.0 → v0.2.0)
- CONTRIBUTING.md updates
- API reference generation with pdoc
- SQL Injection: Already handled by SQLAlchemy/SQLModel
- Audit Trails: Ensure user IDs don't leak sensitive info
- Soft Deletes: Ensure deleted data isn't exposed in queries
- Connection Pooling: Secure credential management
- Hook Overhead: Minimal impact, hooks are optional
- Audit Fields: Automatic timestamp updates via database
- Soft Deletes: Index on
is_deletedfor query performance - Transaction Managers: No overhead vs manual commit/rollback
- Caching Layer: Redis/memcached integration
- Change Tracking: Track field-level changes
- Batch Operations: Enhanced bulk operations with progress
- Migration Utilities: Schema migration helpers
- GraphQL Support: GraphQL query integration
- OpenTelemetry: Tracing and metrics
- Rate Limiting: Built-in rate limiting for operations
- Validation Hooks: Data validation before operations
v0.2.0 represents a significant step toward production-readiness while maintaining 100% backward compatibility. The focus is on developer experience, reliability, and common production needs like audit trails, transactions, and error handling.
Estimated Development Time: 3-4 weeks Target Release Date: TBD Risk Level: Low (backward compatible, well-tested)
- Architecture Review
- Security Review
- Performance Review
- Documentation Review
- Stakeholder Approval
Document Version: 1.0 Last Updated: 2026-02-16 Author: Claude Sonnet 4.5 Status: DRAFT - Awaiting Review