abs-nosql-repository-core
Shared base repository and service utilities for Beanie (NoSQL)-backed FastAPI apps.
Overview
This package provides reusable base classes and utilities to simplify building repository and service layers for MongoDB/Beanie-based FastAPI applications. It includes:
- BaseRepository: Generic CRUD and query logic for Beanie/MongoDB documents.
- BaseCollectionRepository: CRUD and schema management for MongoDB collections.
- BaseService: Service layer abstraction for business logic using repositories.
- BaseCollectionService: Service layer for collection-level operations.
- BaseDocument: Standard document model for Beanie.
- BaseSchema: Standard Pydantic schema for API/data validation.
Installation
Add to your pyproject.toml:
[dependencies]
abs-nosql-repository-core = ">=0.9.7,<0.10.0"
Or install via pip :
pip install abs-nosql-repository-core
Dependencies
- Python >=3.13
- fastapi
- beanie
- abs-exception-core
- abs-utils
Usage
1. Define Your Document Model
from beanie import Document
from abs_nosql_repository_core.document import BaseDocument
class MyEntity(BaseDocument):
name: str
description: str
2. Define Your Repository
from abs_nosql_repository_core.repository import BaseRepository
from myapp.model import MyEntity
class MyEntityRepository(BaseRepository):
def __init__(self):
super().__init__(document=MyEntity)
Or for collection-level operations:
from abs_nosql_repository_core.repository import BaseCollectionRepository
from motor.motor_asyncio import AsyncIOMotorDatabase
class MyCollectionRepository(BaseCollectionRepository):
def __init__(self, db: AsyncIOMotorDatabase):
super().__init__(db)
3. Define Your Service
from abs_nosql_repository_core.service import BaseService
from myapp.repository import MyEntityRepository
class MyEntityService(BaseService):
def __init__(self, repository: MyEntityRepository):
super().__init__(repository)
Or for collection-level services:
from abs_nosql_repository_core.service import BaseCollectionService
from myapp.repository import MyCollectionRepository
class MyCollectionService(BaseCollectionService):
def __init__(self, repository: MyCollectionRepository):
super().__init__(repository)
4. Use in Your FastAPI App
- Initialize your repositories and services in your dependency injection container or app startup.
- Use the service methods in your API endpoints.
5. Example: Entity Manager Service Integration
Repository Example:
from abs_nosql_repository_core.repository import BaseRepository
from src.model import Entities
class EntityManageRepository(BaseRepository):
def __init__(self):
super().__init__(document=Entities)
# Add custom methods as needed
Service Example:
from abs_nosql_repository_core.service import BaseService
from src.repository.entity_manage_repository import EntityManageRepository
class EntityManageService(BaseService):
def __init__(self, repository: EntityManageRepository):
super().__init__(repository)
# Add business logic methods as needed
Usage in FastAPI Endpoint:
from fastapi import APIRouter, Depends
from src.service.entity_manage_service import EntityManageService
router = APIRouter()
@router.post("/entities/")
async def create_entity(entity_data: EntitySchema, service: EntityManageService = Depends()):
return await service.create_entity(entity_data)
API Reference
Repository Layer
BaseRepository(document=None, db=None, ...)— For document-based CRUD and queries.BaseCollectionRepository(db)— For collection-level operations (create, drop, add field, etc).
Service Layer
BaseService(repository)— Standard CRUD and query service.BaseCollectionService(repository)— Collection management service.
Document & Schema
BaseDocument— Extend for your Beanie models.BaseSchema— Extend for your Pydantic schemas.
Common Methods
create,bulk_create,get_by_attr,get_all,update,delete,bulk_update,bulk_delete, etc.
Advanced Features
- Filtering, sorting, pagination, and aggregation via
ListFilter,FilterSchema, etc. - Reference fields and joins via
AggregationStage. - Embedding and event hooks (see source for details).
Integrating Azure Service Bus and Socket.IO with Repositories
The BaseRepository supports integration with Azure Service Bus and Socket.IO for event-driven and real-time features. This allows you to emit events (such as record creation, update, or deletion) to external systems or clients.
Why Integrate?
- Azure Service Bus: Enables asynchronous event processing, microservice communication, and integration with cloud workflows.
- Socket.IO: Enables real-time notifications to connected clients (e.g., web dashboards) when data changes.
How to Use
1. Initialize the Utilities
from abs_utils.azure_service_bus.azure_service_bus import AzureServiceBus
from abs_utils.socket_io.server import SocketIOService
# Azure Service Bus
azure_service_bus = AzureServiceBus(
connection_string="<your-azure-service-bus-connection-string>",
queue_name="<your-queue-name>"
)
# Socket.IO
sio_service = SocketIOService(cors_origins="*")
2. Inject into Your Repository
from abs_nosql_repository_core.repository import BaseRepository
from myapp.model import MyEntity
class MyEntityRepository(BaseRepository):
def __init__(self, sio_service=None, azure_service_bus=None):
super().__init__(
document=MyEntity,
sio_service=sio_service, # Optional: for real-time events
azure_service_bus=azure_service_bus # Optional: for async event bus
)
3. Event Emission in CRUD Operations
When you call create, update, or delete on your repository, events will be automatically sent to Azure Service Bus and/or broadcasted via Socket.IO if the respective services are provided.
- Azure Service Bus: Sends structured event payloads for downstream processing.
- Socket.IO: Emits events to rooms or clients for real-time updates.
4. Example Usage in a Service
from myapp.repository import MyEntityRepository
from abs_nosql_repository_core.service import BaseService
class MyEntityService(BaseService):
def __init__(self, repository: MyEntityRepository):
super().__init__(repository)
# Usage
service = MyEntityService(
repository=MyEntityRepository(
sio_service=sio_service,
azure_service_bus=azure_service_bus
)
)
# Create an entity (will emit events)
await service.create({"name": "Test Entity"})
5. Dependency Injection Example (FastAPI)
If you use a DI container (e.g., dependency_injector):
from dependency_injector import containers, providers
from abs_utils.azure_service_bus.azure_service_bus import AzureServiceBus
from abs_utils.socket_io.server import SocketIOService
from myapp.repository import MyEntityRepository
class Container(containers.DeclarativeContainer):
azure_service_bus = providers.Singleton(
AzureServiceBus,
connection_string="<your-conn-str>",
queue_name="<your-queue>"
)
sio_service = providers.Singleton(SocketIOService, cors_origins="*")
my_entity_repository = providers.Factory(
MyEntityRepository,
sio_service=sio_service,
azure_service_bus=azure_service_bus
)
6. Custom Event Handling
You can extend your repository to emit custom events or handle event payloads as needed by overriding methods or using the provided services directly.
License
MIT
1 maintainer
Metadata
1 maintainer
Release files for abs-nosql-repository-core 0.12.1
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| abs_nosql_repository_core-0.12.1.tar.gz | 21.7 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| abs_nosql_repository_core-0.12.1-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 47.3 kB
Release files / abs_nosql_repository_core-0.12.1.tar.gz
| Download URL | abs_nosql_repository_core-0.12.1.tar.gz |
|---|---|
| Size | 21.7 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
7652022c887d949f640f45f28ec7ce1dda67e8bcf9ca0bf5668fbe5a7d600e0e
|
|
BLAKE2b-256 checksum How to use checksums |
42e3f7f9ae47713527ce340d47bdf72516eb3e09508b86e643fe1b6b9d2c55e0
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.1.3 CPython/3.13.5 Darwin/24.5.0
|
Release files / abs_nosql_repository_core-0.12.1-py3-none-any.whl
| Download URL | abs_nosql_repository_core-0.12.1-py3-none-any.whl |
|---|---|
| Size | 25.6 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
d4cd411764ad68ab994f4f9f06b28747c31b1c308d713f98e809039974bec228
|
|
BLAKE2b-256 checksum How to use checksums |
a723d3519a350f986eb95de2dc206fa3e3b723d92fcba849761e036a57ac3f2a
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.1.3 CPython/3.13.5 Darwin/24.5.0
|
