bench get-app https://github.com/vyogotech/erpnext-microservices-lib

[![Listed on Frappe Gems](https://frappegems.com/api/method/frappe_gems.seo.badge?app=vyogotech%2Ferpnext-microservices-lib)](https://frappegems.com/gems/apps/vyogotech/erpnext-microservices-lib)

Frappe Microservice Framework

A Python framework for building secure, isolated Frappe microservices with proper bounded context and multi-tenant support.

Features

• 🔒 Secure by Default: All endpoints require authentication via Central Site
• 🔌 Independent Database: Each microservice has its own database (bounded context principle)
• 👤 User Context Injection: Authenticated user is automatically injected into handlers
• 🚀 Zero Boilerplate: Create CRUD APIs with a single line of code
• 🏢 Multi-Tenant Ready: Built-in support for tenant isolation
• 🛡️ Tenant-Aware Database: Automatic tenant_id filtering prevents cross-tenant data access
• 🪝 Document Hooks: Frappe-style lifecycle hooks without modifying Frappe core
• 📦 Frappe-Native: Seamlessly works with Frappe DocTypes and APIs
• 📚 Auto-Generated Swagger Docs: Interactive API documentation at /apidocs
• 🔑 OAuth2 + SID Authentication: Support for Bearer tokens and session cookies
• 🛡️ SQL Injection Prevention: Mandatory parameterized queries
• ⚙️ Integrated Background Tasks: Built-in RQ support for asynchronous processing

Installation

pip install frappe-microservice

Or install from source:

cd frappe-microservice-lib
pip install -e .

Quick Start

Basic Microservice

from frappe_microservice import create_microservice

# Initialize microservice
app = create_microservice("my-service")

# Create a secure endpoint (authentication required)
@app.secure_route('/hello', methods=['GET'])
def hello(user):
    """User is automatically injected after authentication"""
    return {"message": f"Hello {user}!"}

# Start the service
app.run()

Automatic CRUD with Resource API

from frappe_microservice import create_microservice

app = create_microservice("orders-service")

# Register a DocType for automatic CRUD (zero code!)
app.register_resource("Sales Order")

# This automatically creates:
# GET    /api/resource/sales-order        - List orders
# POST   /api/resource/sales-order        - Create order
# GET    /api/resource/sales-order/{name} - Get order
# PUT    /api/resource/sales-order/{name} - Update order
# DELETE /api/resource/sales-order/{name} - Delete order

app.run()

Custom Business Logic

from frappe_microservice import create_microservice
import frappe

app = create_microservice("signup-service")

@app.secure_route('/signup', methods=['POST'])
def signup_company(user):
    """Create a new company with multi-tenant isolation"""
    from flask import request
    data = request.json

    # Create company with tenant_id for isolation
    company = frappe.get_doc({
        "doctype": "Company",
        "company_name": data['company_name'],
        "tenant_id": generate_tenant_id(),
        "admin_user": user
    })
    company.insert()

    return {
        "success": True,
        "company_id": company.name,
        "tenant_id": company.tenant_id
    }

app.run()

Tenant-Aware Database (Automatic Tenant Isolation)

The TenantAwareDB wrapper automatically adds tenant_id to all queries, preventing accidental cross-tenant data access:

from frappe_microservice import create_microservice

app = create_microservice("my-service")

@app.secure_route('/users', methods=['GET'])
def list_users(user):
    # Get tenant for authenticated user
    tenant_id = get_user_tenant_id(user)
    app.set_tenant_id(tenant_id)

    # ✅ Automatically adds tenant_id filter!
    users = app.tenant_db.get_all(
        'User',
        fields=['name', 'email', 'role']
    )

    return {"data": users}

@app.secure_route('/users/', methods=['GET'])
def get_user(user, user_id):
    tenant_id = get_user_tenant_id(user)
    app.set_tenant_id(tenant_id)

    try:
        # ✅ Automatically verifies tenant ownership!
        user_doc = app.tenant_db.get_doc('User', user_id)
        return user_doc.as_dict()
    except frappe.PermissionError:
        return {"error": "Access denied"}, 403

Why TenantAwareDB?

• Secure by Default: Impossible to forget tenant_id filter
• Automatic Filtering: All getall/getdoc calls are tenant-scoped
• Prevents Leaks: Raises PermissionError if accessing other tenant's data
• Zero Boilerplate: No manual tenant_id filtering needed

See TENANTAWAREDB_EXAMPLE.py for complete examples.

Document Lifecycle Hooks (No Frappe Modifications!)

Register hooks for document lifecycle events without modifying Frappe core. Perfect for microservices!

from frappe_microservice import create_microservice

app = create_microservice("orders-service")

# Global hook - runs for ALL doctypes
@app.tenant_db.before_insert('*')
def ensure_tenant_id(doc):
    """Ensure tenant_id is set on all documents"""
    from flask import g
    if not doc.tenant_id:
        doc.tenant_id = g.tenant_id

# DocType-specific hooks
@app.tenant_db.before_insert('Sales Order')
def set_order_defaults(doc):
    """Set defaults for new orders"""
    if not doc.status:
        doc.status = 'Draft'
    if not doc.order_date:
        doc.order_date = frappe.utils.today()

@app.tenant_db.after_insert('Sales Order')
def send_order_notification(doc):
    """Send notification after order creation"""
    print(f"📧 Order {doc.name} created for {doc.customer}")

@app.tenant_db.before_validate('Sales Order')
def validate_order_amount(doc):
    """Custom validation"""
    if doc.grand_total and doc.grand_total < 0:
        frappe.throw("Order total cannot be negative")

# Use in endpoints - hooks run automatically!
@app.secure_route('/orders', methods=['POST'])
def create_order(user):
    from flask import request

    tenant_id = get_user_tenant_id(user)
    app.set_tenant_id(tenant_id)

    # All hooks run automatically during insert!
    doc = app.tenant_db.insert_doc('Sales Order', request.json)

    return {"success": True, "name": doc.name}

Supported Hook Events: - before_validate - Before validation starts - validate - During validation - before_insert - Before inserting into database - after_insert - After inserting into database - before_update - Before updating document - after_update - After updating document - before_delete - Before deleting document - after_delete - After deleting document

Why Document Hooks? - 🚫 No Frappe Core Changes: Works entirely in microservice layer - 🎯 Microservice-Specific: Each service has its own hooks - 🔧 Full Control: Easy to test and debug - 📝 Clean Code: Separate business logic from endpoints - 🔄 Reusable: Write once, applies to all operations

See DOCUMENTHOOKSEXAMPLES.py for comprehensive examples.

Traditional DocType Controllers

Use familiar Frappe-style controller classes for your DocTypes! No Frappe core modifications needed.

# controllers/sales_order.py
from frappe_microservice.controller import DocumentController

class SalesOrder(DocumentController):
    def validate(self):
        """Validate order data"""
        if not self.customer:
            self.throw("Customer is required")
        self.calculate_total()

    def before_insert(self):
        """Set defaults"""
        if not self.status:
            self.status = 'Draft'
        if not self.order_date:
            self.order_date = frappe.utils.today()

    def after_insert(self):
        """Post-creation tasks"""
        self.send_notification()
        self.update_customer_stats()

    def calculate_total(self):
        """Reusable method"""
        self.grand_total = sum(item.amount for item in self.items)

    def send_notification(self):
        print(f"Order {self.name} created")

# server.py
from frappe_microservice import create_microservice, setup_controllers

app = create_microservice("orders-service")

# Auto-discover and register controllers
setup_controllers(app, controllers_directory="./controllers")

# Controllers run automatically during insert/update!
@app.secure_route('/orders', methods=['POST'])
def create_order(user):
    doc = app.tenant_db.insert_doc('Sales Order', request.json)
    return {"success": True, "name": doc.name}

Features: - 🎯 Traditional Pattern: Familiar Frappe controller style - 📁 One File Per DocType: Clean code organization - 🔄 Auto-Discovery: Automatically loads from directory - 🎭 Lifecycle Methods: validate(), beforeinsert(), afterinsert(), etc. - 🛠️ Custom Methods: Define reusable business logic - 🧪 Easy Testing: Test controllers independently

File Naming Convention: - sales_order.py → Sales Order DocType → SalesOrder class - signup_user.py → Signup User DocType → SignupUser class

See signup-service/ for a complete example with controllers.

Central Site API Client

The CentralSiteClient provides a standardized way for microservices to communicate back to the Central Site using a Frappe-like API. It is available as app.central in any MicroserviceApp.

from frappe_microservice import create_microservice

app = create_microservice("my-service")

@app.secure_route('/sync', methods=['POST'])
def sync_with_central(user):
    # Fetch a document from the Central Site
    tenant_info = app.central.get_doc("Tenant", "tenant-123")

    # Update a record on the Central Site
    app.central.update("Tenant", "tenant-123", {"status": "Active"})

    # Call a whitelisted method
    result = app.central.call("some_whitelisted_method", {"param": "value"})

    return {"status": "synced"}

Background Task Processing (RQ)

The framework includes integrated support for RQ (Redis Queue) to handle long-running tasks like tenant provisioning or email sending.

Activation

Background task processing is disabled by default. Enable it by setting the following environment variables in your container:

Variable	Description
ENABLE_RQ	Set to 1 to auto-start the embedded RQ worker thread
REDIS_URL	Redis connection URL (default: redis://localhost:6379)

Usage in Code

Any MicroserviceApp instance can enqueue tasks directly:

from frappe_microservice import create_microservice

app = create_microservice("my-service")

@app.secure_route('/task', methods=['POST'])
def trigger_task(user):
    # Enqueue a function to run in the background
    # The framework automatically restores Frappe context in the worker!
    app.enqueue_task(long_running_function, arg1, kwarg1="value")
    return {"status": "enqueued"}

def long_running_function(arg1, kwarg1=None):
    import frappe
    # Full Frappe API available here
    frappe.get_doc({...}).insert()
    frappe.db.commit()

Features:

• Zero Configuration: No separate worker process needed; it runs as a daemon thread inside your service container.
• Context Preservation: Automatically restores frappe.local and DB connections in the worker process.
• Controller Support: Auto-discovery of DocType controllers works seamlessly in background tasks.

Configuration Environment Variables:

Variable	Description	Default
CENTRAL_SITE_URL	URL of the Central Site	http://central-site:8000
CENTRAL_SITE_API_KEY	API Key for authentication	-
CENTRAL_SITE_API_SECRET	API Secret for authentication	-
CENTRAL_SITE_USER	Username (alternative to API Key)	-
CENTRAL_SITE_PASSWORD	Password (alternative to API Secret)	-
CENTRAL_SITE_TIMEOUT	Request timeout in seconds	10

The client is lazily initialized, meaning no connection attempts are made until the first time app.central is accessed.

Container Deployment (Library Entrypoint)

The base image runs the library entrypoint so services do not need their own entrypoint.py. The framework discovers your app via environment