tc-db-base 1.0.3

Schema-driven MongoDB database module with auto-generated repositories

tc-db-base - Schema-Driven Database Service

A flexible, schema-driven database module that auto-generates repositories with CRUD operations, search functions, and index management based on a JSON schema.

Installation

# Install from PyPI
pip install tc-db-base

# Or with Flask support
pip install tc-db-base[flask]

Features

• Schema-Driven: Define collections in JSON, get auto-generated methods
• Auto-Generated Finders: find_by_{field}() for unique/searchable fields
• Compound Queries: find_by_x_and_y(), find_by_x_or_y(), find_by_x_not_y()
• Fluent Query Builder: Chain methods like .where().order_by().limit().get()
• Transactions: Flexible transaction support - use only when needed via session parameter
• Real-Time Pub/Sub: MongoDB Change Streams for live notifications
• Validation: Automatic validation against schema
• Soft Delete: Optional soft delete support per collection
• Timestamps: Auto-managed created_at / updated_at
• Index Management: Auto-create indexes from schema
• Search: Built-in search across searchable fields
• Aggregation Shortcuts: sum(), avg(), min(), max(), group_by()
• Bulk Operations: update(), delete(), increment() on query results
• Unit of Work: Group multiple operations into atomic transactions
• Event Emitter: Local pub/sub without MongoDB dependency

Getting Started

1. Create Schema Files

Create your schema in resources/schema/:

your_project/
├── resources/
│   └── schema/
│       ├── dbs.json           # Main config
│       └── dbs/
│           └── app_db.json    # Database schema
└── app.py

See examples/ folder for sample schema files.

2. Define Your Database Schema

resources/schema/dbs.json:

{
  "version": "1.0.0",
  "databases": ["app_db"],
  "settings": {
    "auto_timestamps": true,
    "default_limit": 100
  }
}

resources/schema/dbs/app_db.json:

{
  "db_name": "app_db",
  "collections": {
    "users": {
      "fields": {
        "user_id": {"type": "string", "required": true},
        "email": {"type": "string", "required": true, "format": "email"},
        "name": {"type": "string"},
        "status": {"type": "string", "enum": ["active", "inactive"], "default": "active"},
        "created_at": {"type": "datetime", "auto": "create"},
        "updated_at": {"type": "datetime", "auto": "update"}
      },
      "unique_fields": ["user_id", "email"],
      "searchable_fields": ["user_id", "email", "name", "status"],
      "soft_delete": true
    }
  }
}

3. Use the Auto-Generated Repository

from tc_db_base import init_db, get_repository

# Initialize and connect
db = init_db()

# Get auto-generated repository
users = get_repository('users')

# Create (with auto-validation and timestamps)
user_id = users.create({
    'user_id': 'usr_123',
    'email': '[e-mail]',
    'name': 'John Doe'
})

# Auto-generated finders (from unique_fields)
user = users.find_by_user_id('usr_123')
user = users.find_by_email('[e-mail]')

# Auto-generated finders (from searchable_fields)
active_users = users.find_by_status('active')

# Search across searchable fields
results = users.search('john', limit=10)

# Standard CRUD
user = users.find_by_id(user_id)
users.update_by_id(user_id, {'name': 'John Smith'})
users.delete_by_id(user_id)  # Soft delete if enabled

Schema Definition (dbs.json)

{
  "dbs": {
    "user_db": {
      "users": {
        "description": "User accounts",
        "fields": {
          "user_key": {"type": "string", "required": true},
          "email": {"type": "string", "required": true, "format": "email"},
          "name": {"type": "string"},
          "status": {"type": "string", "enum": ["active", "inactive"], "default": "active"},
          "created_at": {"type": "datetime", "auto": "create"},
          "updated_at": {"type": "datetime", "auto": "update"}
        },
        "unique_fields": ["user_key", "email"],
        "indexes": {
          "user_key_idx": {"fields": ["user_key"], "unique": true},
          "email_idx": {"fields": ["email"], "unique": true}
        },
        "searchable_fields": ["user_key", "email", "name"],
        "soft_delete": true,
        "timestamps": true
      }
    }
  }
}

Auto-Generated Methods

For each collection, these methods are auto-generated:

CRUD Operations

repo.create(data)                    # Insert with validation
repo.create_many(documents)          # Bulk insert
repo.find_by_id(id)                  # Find by _id
repo.find_one(query)                 # Find single
repo.find_many(query, skip, limit)   # Find multiple
repo.update_one(query, data)         # Update single
repo.update_by_id(id, data)          # Update by ID
repo.delete_one(query)               # Delete (soft if enabled)
repo.delete_by_id(id)                # Delete by ID
repo.hard_delete(query)              # Permanent delete
repo.restore(query)                  # Restore soft-deleted
repo.count(query)                    # Count documents
repo.exists(query)                   # Check existence

Dynamic Finders (auto-generated from schema)

# For unique_fields: ["user_key", "email"]
repo.find_by_user_key(value)         # Returns single doc
repo.find_by_email(value)            # Returns single doc

# For searchable_fields: ["account_key", "name"]
repo.find_by_account_key(value)      # Returns list
repo.find_by_name(value)             # Returns list

# Compound queries (AND, OR, NOT)
repo.find_by_user_key_and_account_key(uk, ak)     # AND
repo.find_by_user_key_or_account_key(uk, ak)      # OR
repo.find_by_account_key_not_user_key(ak, uk)     # NOT

Fluent Query Builder

# Complex queries with fluent API
results = (repo.query()
    .where('status', 'active')
    .where_gt('age', 18)
    .where_in('role', ['admin', 'manager'])
    .where_between('created_at', start, end)
    .where_like('name', '%john%')
    .or_where('is_vip', True)
    .order_by('created_at', 'desc')
    .skip(10)
    .limit(20)
    .select(['name', 'email'])
    .get())

# Single document
user = repo.query().where('email', '[e-mail]').first()

# Existence check
exists = repo.query().where('email', '[e-mail]').exists()

# Count
count = repo.query().where('status', 'active').count()

# Aggregations
total = repo.query().where('status', 'active').sum('amount')
avg = repo.query().where('status', 'active').avg('age')

# Bulk operations
repo.query().where('status', 'pending').update({'status': 'active'})
repo.query().where('status', 'deleted').delete()

# Array operations
repo.query().where('user_key', 'u123').push('tags', 'new_tag')
repo.query().where('user_key', 'u123').increment('login_count')

# Pagination
results = repo.query().where('status', 'active').page(2, per_page=20).get()

# Nested conditions
results = (repo.query()
    .where('status', 'active')
    .where_or(lambda q: q.where('role', 'admin').where('role', 'manager'))
    .get())

Search & Aggregation

repo.search(text, fields, limit)     # Search searchable_fields
repo.search_by_fields(field=value)   # Multi-field search
repo.aggregate(pipeline)             # Run aggregation
repo.group_by(field, match)          # Group by field
repo.count_by(field)                 # Count grouped
repo.distinct(field, query)          # Distinct values

Index Management

repo.ensure_indexes()                # Create indexes from schema
repo.get_indexes()                   # List current indexes

Query Builder Reference

Where Conditions

Method	Description	Example
where(field, value)	Equals	.where('status', 'active')
where_eq(field, value)	Equals	.where_eq('age', 25)
where_ne(field, value)	Not equals	.where_ne('status', 'deleted')
where_gt(field, value)	Greater than	.where_gt('age', 18)
where_gte(field, value)	Greater or equal	.where_gte('score', 100)
where_lt(field, value)	Less than	.where_lt('price', 50)
where_lte(field, value)	Less or equal	.where_lte('quantity', 10)
where_in(field, list)	In list	.where_in('status', ['a', 'b'])
where_not_in(field, list)	Not in list	.where_not_in('role', ['guest'])
where_like(field, pattern)	SQL LIKE	.where_like('name', '%john%')
where_regex(field, pattern)	Regex match	.where_regex('email', '.*@gmail')
where_between(field, a, b)	Between values	.where_between('age', 18, 65)
where_null(field)	Is null	.where_null('deleted_at')
where_not_null(field)	Not null	.where_not_null('email')
where_exists(field)	Field exists	.where_exists('profile')

Execution Methods

Method	Description
get() / all()	Get all results
first()	Get first result
first_or_fail()	Get first or raise
last()	Get last result
count()	Count results
exists()	Check if exists
distinct(field)	Get distinct values
pluck(field)	Get field values only

Modification Methods

Method	Description
update(data)	Update matching docs
delete()	Delete matching docs
increment(field, n)	Increment field
decrement(field, n)	Decrement field
push(field, value)	Push to array
pull(field, value)	Remove from array

Transactions

Flexible transaction support - use only when needed by passing session parameter.

Basic Usage (Context Manager)

users = get_repository('users')
accounts = get_repository('accounts')

# Transactions only when you need them
with users.transaction() as session:
    user_id = users.create({'name': 'John', 'email': '[e-mail]'}, session=session)
    accounts.create({'user_id': user_id, 'balance': 0}, session=session)
    # Auto-commits on success
    # Auto-rollback on any exception

Callback Style

def create_user_with_account(session):
    user_id = users.create({'name': 'John'}, session=session)
    accounts.create({'user_id': user_id}, session=session)
    return user_id

result = users.with_transaction(create_user_with_account)

Using @transactional Decorator

from tc_db_base import transactional

@transactional
def create_order(customer_id, items, txn=None):
    order_id = orders.create({
        'customer_id': customer_id,
        'items': items,
        'status': 'pending'
    }, session=txn.session)
    
    # Update inventory
    for item in items:
        inventory.update_one(
            {'product_id': item['product_id']},
            {'quantity': item['quantity']},
            session=txn.session
        )
    
    return order_id

# Call normally - transaction is automatic
order_id = create_order('c123', [{'product_id': 'p1', 'quantity': 2}])

Unit of Work Pattern

from tc_db_base import UnitOfWork

uow = UnitOfWork()

# Register operations (not executed yet)
uow.register_new('users', {'name': 'John', 'email': '[e-mail]'})
uow.register_new('accounts', {'user_id': 'temp', 'balance': 0})
uow.register_update('settings', {'user_id': 'u123'}, {'theme': 'dark'})
uow.register_delete('sessions', {'user_id': 'u123'})

# Execute all in single transaction
results = uow.commit()
print(f"Inserted: {results['inserted_ids']}")
print(f"Modified: {results['modified_count']}")
print(f"Deleted: {results['deleted_count']}")

Standalone Transaction Manager

from tc_db_base import transaction

# Works across multiple repositories
with transaction() as txn:
    users.create({...}, session=txn.session)
    orders.create({...}, session=txn.session)
    payments.create({...}, session=txn.session)

Key Points

• Flexible: Normal operations don't use transactions (faster)
• Opt-in: Pass session=session only when needed
• Auto-rollback: Any exception triggers rollback
• Cross-collection: Same session works across different repositories

Real-Time Pub/Sub

Built-in pub/sub using MongoDB Change Streams for real-time notifications.

Basic Usage

from tc_db_base import get_pubsub, ChangeType, ChangeEvent

pubsub = get_pubsub()

# Subscribe to all changes
def on_user_change(event: ChangeEvent):
    print(f"User {event.document_id} was {event.change_type.value}")
    print(f"Document: {event.document}")

sub_id = pubsub.subscribe('users', on_user_change)

# Subscribe to specific operations
sub_id = pubsub.subscribe(
    'users',
    on_user_change,
    change_types=[ChangeType.INSERT, ChangeType.UPDATE]
)

# Subscribe with filter (only active users)
sub_id = pubsub.subscribe(
    'users',
    on_user_change,
    filter_query={'status': 'active'}
)

# Start listening
pubsub.start()

# Unsubscribe
pubsub.unsubscribe(sub_id)

# Stop
pubsub.stop()

Decorator Syntax

@pubsub.on('users', [ChangeType.INSERT])
def handle_new_user(event: ChangeEvent):
    print(f"New user created: {event.document}")

@pubsub.on_insert('orders')
def handle_new_order(event: ChangeEvent):
    # Send notification, update inventory, etc.
    pass

@pubsub.on_update('users')
def handle_user_update(event: ChangeEvent):
    # event.update_description contains changed fields
    pass

@pubsub.on_delete('sessions')
def handle_session_end(event: ChangeEvent):
    pass

Change Event Properties

event.change_type      # ChangeType enum (INSERT, UPDATE, DELETE, etc.)
event.collection       # Collection name
event.document_id      # Document _id as string
event.document         # Full document (on insert/update with lookup)
event.update_description  # {updatedFields: {...}, removedFields: [...]}
event.timestamp        # When the change occurred
event.raw_event        # Original MongoDB change event

Local Event Emitter

For pub/sub without MongoDB (in-memory):

from tc_db_base import get_emitter

emitter = get_emitter()

# Subscribe
emitter.on('user:created', lambda data: print(f"User: {data}"))
emitter.once('app:ready', lambda: print("App is ready!"))

# Publish
emitter.emit('user:created', {'name': 'John'})

# Unsubscribe
emitter.off('user:created')

Field Types

Type	Python Type	Description
string	str	Text values
number	int/float	Numeric values
boolean	bool	True/False
array	list	Lists
object	dict	Nested objects
datetime	datetime	Timestamps

Field Options

Option	Description
required	Field is mandatory
default	Default value
enum	Allowed values list
format	Validation format (email, url, phone)
sensitive	Excluded from query results
auto	Auto-set: "create" or "update"

REST API (Standalone Server)

# Using module
python -m tc_db_base.server --port 5002

# Or using installed command
tc-db-server --port 5002

Endpoints

GET  /health                         # Health check
GET  /schema                         # Full schema
GET  /schema/{collection}            # Collection schema

GET  /api/v1/{collection}            # List documents
GET  /api/v1/{collection}/{id}       # Get by ID
POST /api/v1/{collection}            # Create
PUT  /api/v1/{collection}/{id}       # Update
DELETE /api/v1/{collection}/{id}     # Delete

GET  /api/v1/{collection}/search?q=text  # Search
POST /api/v1/{collection}/aggregate      # Aggregation
GET  /api/v1/{collection}/count-by/{field}  # Count by field

Project Structure

tc_db_base/
├── __init__.py          # Package exports
├── client.py            # MongoDB client
├── service.py           # DatabaseService
├── repository.py        # DynamicRepository
├── query_builder.py     # Fluent query builder
├── pubsub.py            # Real-time pub/sub
├── server.py            # REST API server
└── schema/
    ├── __init__.py      # SchemaLoader
    ├── validator.py     # SchemaValidator
    └── dbs.json         # Database schema

Configuration

Uses task_circuit_base for configuration (if available), otherwise falls back to environment variables.

Schema files are loaded from (in priority order):

1. SCHEMA_PATH environment variable
2. {cwd}/resources/schema/
3. Parent directories /resources/schema/
4. Package default tc_db_base/schema/

# Via environment variables
MONGO_URI=mongodb://localhost:27017
SCHEMA_PATH=/path/to/custom/schema

# Via resources/config.yaml (if using task_circuit_base)
database:
  mongo_uri: mongodb://localhost:27017
  connection:
    max_pool_size: 100