tc-db-base 1.0.2

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
• 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

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

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

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
├── 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