RB-Commons
RB-Commons is a lightweight async Python library that simplifies database operations through a clean manager interface built on top of SQLAlchemy's async capabilities. It provides a robust foundation for handling common database operations while maintaining full type safety through Python's typing system.
Features
Async-First Design
- Built on top of SQLAlchemy's async functionality
- Efficient handling of async database operations
- Proper transaction and session management
- Type-safe operations with Generic types
Core Functionality
- CRUD Operations: Complete set of Create, Read, Update, and Delete operations
- Flexible Filtering: Support for dynamic query filtering
- Instance Management: Both instance-level and query-level updates
- Error Handling: Comprehensive error handling with custom exceptions
- Transaction Safety: Automatic rollbacks on failures
Installation
pip install rb-commons
Quick Start
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from rb_commons.orm import BaseManager
# Define your model
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)
# Create your manager
class UserManager(BaseManager[User]):
model = User
# Usage in async context
async def main():
# Setup database connection
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async with AsyncSession(engine) as session:
# Initialize manager
user_manager = UserManager(session)
# Create user
user = await user_manager.create(
name="John Doe",
email="john@example.com"
)
# Get user
user = await user_manager.get(id=1)
# Filter users
users = await user_manager.filter(name="John Doe")
# Update user by filters
updated_user = await user_manager.update_by_filters(
filters={"id": 1},
name="Jane Doe"
)
# Delete user
success = await user_manager.delete(id=1)
Core Operations
Get and Filter
# Get single instance (returns Optional[ModelType])
user = await user_manager.get(id=1)
# Filter multiple instances (returns List[ModelType])
active_users = await user_manager.filter(is_active=True)
# Check existence
exists = await user_manager.is_exists(email="john@example.com")
Create
try:
user = await user_manager.create(
name="John Doe",
email="john@example.com"
)
except DatabaseException as e:
print(f"Database error: {e}")
except InternalException as e:
print(f"Internal error: {e}")
Update
RB-Commons provides three different ways to update records:
# 1. Update by filters - updates records matching filters and returns updated instance
updated_user = await user_manager.update_by_filters(
filters={"id": 1},
name="New Name"
)
# 2. Update instance with specific fields
user = await user_manager.get(id=1)
if user:
updated_user = await user_manager.update(
instance=user,
name="New Name"
)
# 3. Save modified instance
user = await user_manager.get(id=1)
if user:
user.name = "New Name"
saved_user = await user_manager.save(user)
Delete
# Delete by ID
success = await user_manager.delete(id=1)
# Delete by filters
success = await user_manager.delete(email="old@example.com")
# Bulk delete
deleted_count = await user_manager.bulk_delete(is_active=False)
Error Handling
RB-Commons provides custom exceptions for better error handling:
DatabaseException: For SQLAlchemy and database-related errorsInternalException: For internal operation errors
try:
user = await user_manager.create(name="John")
except DatabaseException as e:
# Handle database errors (e.g., constraint violations)
print(f"Database error: {e}")
except InternalException as e:
# Handle internal errors
print(f"Internal error: {e}")
Method Return Types
get():Optional[ModelType]filter():List[ModelType]create():ModelTypedelete():bool | Nonebulk_delete():intupdate():Optional[ModelType]update_by_filters():Optional[ModelType]save():Optional[ModelType]is_exists():bool
Best Practices
- Choose the Right Update Method:
- Use
update_by_filters()for query-based updates - Use
update()for updating specific fields of an instance - Use
save()for saving modified instances
- Use
- Always Use Async Context: The library is designed for async operations, ensure you're running within an async context
- Session Management: Properly manage your database sessions using async context managers
- Error Handling: Implement proper error handling using the provided exception classes
- Type Safety: Utilize the generic typing system for better IDE support and type safety
Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
License
This project is licensed under the MIT License - see the LICENSE file for details.
1 maintainer
Metadata
1 maintainer
Release files for rb-commons 0.7.32
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| rb_commons-0.7.32.tar.gz | 19.8 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| rb_commons-0.7.32-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 42.4 kB
Release files / rb_commons-0.7.32.tar.gz
| Download URL | rb_commons-0.7.32.tar.gz |
|---|---|
| Size | 19.8 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
eb8ef30e9ab4b2f2bba15688da212806d27e15b5ac5e12a37833b328b78bdac2
|
|
BLAKE2b-256 checksum How to use checksums |
db156f7badfa34dca4306931fe9d594e79de4bf804b956360afb0240fc3929b3
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.2.0 CPython/3.11.9
|
Release files / rb_commons-0.7.32-py3-none-any.whl
| Download URL | rb_commons-0.7.32-py3-none-any.whl |
|---|---|
| Size | 22.6 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
ca0548f4124e778c6d0738a15dd6f69a947d52d5db307abe860f13b7b10e4de1
|
|
BLAKE2b-256 checksum How to use checksums |
d39de21231329f26ad0f1f6807dcf1e82796f7ce573f9b2d98a07e27efd286ff
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.2.0 CPython/3.11.9
|
