Skip to main content

Migration Guide: Frappe → Framework M

A practical guide for Frappe developers transitioning to Framework M. This guide builds on Frappe's foundations with explicit patterns for scale, type safety, and decoupled architecture.

Quick Reference​

Frappe PatternFramework M Pattern
frappe.db.get_value()repo.get() via DI
frappe.session.userrequest.state.user (UserContext)
doc.before_save()Controller.before_save()
frappe.get_doc()repo.get(session, id)
Implicit transactionExplicit UnitOfWork
Global stateDependency Injection

1. Database Access​

Frappe (Global State)​

# Anywhere in code
value = frappe.db.get_value("Item", item_code, "stock_qty")
frappe.db.set_value("Item", item_code, "stock_qty", new_qty)

# Get document
doc = frappe.get_doc("Item", item_code)
doc.stock_qty = 100
doc.save()

Framework M (Explicit Session)​

# Repository injected, session explicit
class ItemService:
    def __init__(self, repo: GenericRepository[Item], uow_factory: UoWFactory):
        self.repo = repo
        self.uow_factory = uow_factory

    async def update_stock(self, item_code: str, new_qty: int):
        async with self.uow_factory() as uow:
            item = await self.repo.get_by_name(uow.session, item_code)
            item.stock_qty = new_qty
            await self.repo.save(uow.session, item)
            await uow.commit()

Why? Explicit session enables:

  • Multiple independent transactions per request
  • Easy testing (mock the session)
  • Clear transaction boundaries

2. Getting Current User​

Frappe​

current_user = frappe.session.user
if frappe.session.user == "Administrator":
    # admin logic

Framework M​

# Access via request state
from litestar import Controller, get, Request
from framework_m_core.interfaces import UserContext

class ItemController(Controller):
    @get("/items")
    async def list_items(self, request: Request) -> list[Item]:
        user: UserContext = request.state.user
        # user.id, user.roles available
        if "Administrator" in user.roles:
            # admin logic
        return await self.service.list_for_user(user.id)
No Leaky Abstractions
  • Framework M intentionally does not re-export or wrap Litestar primitives (like Controller, get, Request).
  • Re-exporting creates a false sense of decoupling while still locking developers into Litestar-specific typings and signatures.
  • To maintain a "Zero-Cliff" design, all core logic (such as models, lifecycle hooks, and background jobs) is kept fully framework-agnostic.
  • Direct imports from litestar are reserved exclusively for custom infrastructure/web adapter classes, making server dependencies explicit and audit-friendly.

3. Document Lifecycle Hooks​

Frappe (on DocType)​

class Item(Document):
    def before_save(self):
        self.validate_qty()

    def after_insert(self):
        self.create_stock_entry()

Framework M (on Controller)​

# doctype.py - schema only
class Item(BaseDocType):
    item_code: str
    stock_qty: int

# controller.py - business logic
class ItemController(BaseController[Item]):
    async def validate(self, context=None):
        if self.doc.stock_qty < 0:
            raise ValueError("Stock cannot be negative")

    async def after_insert(self, context=None):
        await self.create_stock_entry()

Hook Order (Insert):

validate → before_insert → before_save → [DB INSERT] → after_save → after_insert

4. Child Tables​

Frappe (Loop in Template)​

{% for item in doc.items %}
<tr>
  <td>{{ item.item_code }}</td>
  <td>{{ item.qty }}</td>
</tr>
{% endfor %}
# Backend
for item in doc.items:
    item.amount = item.qty * item.rate

Framework M + Tamagui​

// Frontend - Tamagui List
import { YStack, XStack, MText } from "@framework-m/ui";

<YStack space="$2">
  {invoice.items?.map((item: any, index: number) => (
    <XStack key={index} justifyContent="space-between">
      <MText>{item.item_code}</MText>
      <MText>{item.qty}</MText>
      <MText>{item.amount}</MText>
    </XStack>
  ))}
</YStack>
# Backend - Pydantic child table
class InvoiceItem(BaseDocType):
    class Meta:
        is_child_table = True

    item_code: str
    qty: Decimal
    rate: Decimal
    amount: Decimal = Decimal("0")

class Invoice(BaseDocType):
    items: list[InvoiceItem] = Field(default_factory=list)

# Controller - calculate amounts
class InvoiceController(BaseController[Invoice]):
    async def before_save(self, context=None):
        for item in self.doc.items:
            item.amount = item.qty * item.rate
        self.doc.total = sum(i.amount for i in self.doc.items)

5. Avoiding N+1 Queries​

Frappe (Common N+1)​

# N+1 problem: 1 query for invoices + N queries for customers
invoices = frappe.get_all("Invoice", fields=["name", "customer"])
for inv in invoices:
    customer = frappe.get_doc("Customer", inv.customer)  # N queries!
    inv.customer_name = customer.customer_name

Framework M (Eager Loading)​

# Single query with JOIN
async def list_invoices_with_customers(self, session: AsyncSession):
    stmt = (
        select(Invoice)
        .options(joinedload(Invoice.customer))  # Eager load
        .limit(100)
    )
    result = await session.execute(stmt)
    return result.scalars().all()

Or use repository with includes:

# Repository pattern with includes
invoices = await repo.list(
    session,
    filters={"status": "Draft"},
    includes=["customer", "items"],  # Eager load relations
)

6. Transactions​

Frappe (Implicit)​

# Transaction per request, auto-commit on success
def create_order():
    order = frappe.get_doc({"doctype": "Order", ...})
    order.insert()  # Auto-commits at request end

    items = create_items(order.name)  # Same transaction

Framework M (Explicit UoW)​

async def create_order(self, data: OrderCreate):
    async with self.uow_factory() as uow:
        # All operations in same transaction
        order = Order(**data.model_dump())
        await self.order_repo.save(uow.session, order)

        for item_data in data.items:
            item = OrderItem(order_id=order.id, **item_data)
            await self.item_repo.save(uow.session, item)

        await uow.commit()  # Explicit commit
    # Auto-rollback if exception before commit

7. Background Jobs​

Frappe​

frappe.enqueue(
    "myapp.tasks.send_email",
    queue="short",
    email=email,
    subject=subject,
)

Framework M (JobQueueProtocol)​

Framework M uses a JobQueueProtocol interface, allowing you to plug in any background job system (Celery, Arq, BullMQ). The default adapter uses Taskiq.

from framework_m_core.interfaces import JobQueueProtocol

# Injected dependency
class EmailService:
    def __init__(self, queue: JobQueueProtocol):
        self.queue = queue

    async def send_welcome(self, email: str):
        # Enqueue task (implementation agnostic)
        await self.queue.enqueue(
            "send_email_task",
            email=email,
            subject="Welcome"
        )

# Defining a background job (Engine-agnostic)
from framework_m import job

@job(name="send_email_task")
async def send_email_task(email: str, subject: str):
    await smtp.send(email, subject)

8. Events / Hooks​

Frappe (doc_events)​

# hooks.py
doc_events = {
    "Invoice": {
        "on_submit": "myapp.invoice.on_submit",
    }
}

Framework M (Event Bus)​

from framework_m_core.interfaces import EventBusProtocol, Event

# Publish event
event = Event(type="invoice.submitted", data={"id": str(invoice.id)})
await event_bus.publish("invoice.submitted", event)

# Subscribe (in another service)
async def handle_invoice_submitted(event: Event) -> None:
    await update_ledger(event.data["id"])

# Register handler on startup
await event_bus.subscribe("invoice.submitted", handle_invoice_submitted)

9. Permissions​

Frappe​

if frappe.has_permission("Invoice", "write", doc.name):
    # allowed

Framework M​

# Via PermissionProtocol (injected)
from framework_m_core.interfaces import PermissionProtocol, PolicyEvaluateRequest, PermissionAction
from framework_m_core.interfaces import UserContext

class InvoiceService:
    def __init__(self, permissions: PermissionProtocol):
        self.permissions = permissions

    async def update(self, user: UserContext, invoice_id: str, data: dict):
        result = await self.permissions.evaluate(PolicyEvaluateRequest(
            principal=user.id,
            action=PermissionAction.WRITE,
            resource="Invoice",
            resource_id=invoice_id,
            principal_attributes={"roles": user.roles},
        ))
        if not result.authorized:
            raise PermissionDenied(result.reason)

Or declarative in Meta:

class Invoice(BaseDocType):
    class Meta:
        permissions = {
            "read": ["Employee", "Manager"],
            "write": ["Manager"],
            "submit": ["Manager"],
        }

10. API Endpoints​

Frappe (whitelisted)​

@frappe.whitelist()
def get_stock_balance(item_code):
    return frappe.db.get_value("Item", item_code, "stock_qty")

Framework M (Litestar routes)​

from litestar import Controller, get

class StockController(Controller):
    path = "/api/stock"

    @get("/{item_code:str}/balance")
    async def get_balance(self, item_code: str, repo: ItemRepository) -> dict:
        item = await repo.get_by_code(item_code)
        return {"balance": item.stock_qty}

Or use RPC-style for Frappe compatibility:

# In my_app/api.py
from framework_m_core.decorators import rpc

@rpc()  # Registers as "my_app.api.get_stock_balance"
async def get_stock_balance(item_code: str, repo: ItemRepository) -> dict:
    item = await repo.get_by_code(item_code)
    return {"balance": item.stock_qty}

# Exposed as: POST /api/v1/rpc/fn/my_app.api.get_stock_balance
# Body: {"item_code": "ITEM-001"}

Two RPC patterns available:

  • Controller methods: POST /api/v1/rpc/{DocType}/{method} - uses @whitelist on controller
  • Standalone functions: POST /api/v1/rpc/fn/{dotted.path} - uses @rpc decorator

11. Queries and Aggregations​

Frappe​

# Get all with filters
items = frappe.get_all("Item", filters={"item_group": "Electronics"}, fields=["name", "price"])

# Query Builder
from frappe.query_builder import DocType
Item = DocType("Item")
query = frappe.qb.from_(Item).select(Item.name, Item.price).where(Item.stock_qty > 0)

# Raw SQL
frappe.db.sql("SELECT name, price FROM tabItem WHERE stock_qty > %s", (0,), as_dict=True)

Framework M​

from framework_m_core.interfaces import FilterSpec, FilterOperator

# Repository list with filters
items = await repo.list(
    session,
    filters=[FilterSpec(field="item_group", operator=FilterOperator.EQ, value="Electronics")],
    limit=100,
)

# SQLAlchemy Core (for complex queries)
from sqlalchemy import select, func
stmt = select(Item.name, Item.price).where(Item.stock_qty > 0)
result = await session.execute(stmt)

# Aggregations
stmt = select(func.sum(Item.stock_qty)).where(Item.item_group == "Electronics")
total = (await session.execute(stmt)).scalar()

12. Print Formats​

Frappe​

# Jinja template in DocType
frappe.get_print("Invoice", invoice_name, print_format="Standard")

Framework M​

# PrintProtocol for document rendering
from framework_m_core.interfaces import PrintProtocol, PrintFormat

class InvoiceService:
    def __init__(self, printer: PrintProtocol, repo: RepositoryProtocol[Invoice]):
        self.printer = printer
        self.repo = repo

    async def generate_pdf(self, invoice_id: str) -> bytes:
        invoice = await self.repo.get(UUID(invoice_id))
        return await self.printer.render(
            doc=invoice,
            template="standard",
            format=PrintFormat.PDF,
        )

    async def list_templates(self) -> list[str]:
        return await self.printer.get_templates("Invoice")

# Adapters: JinjaPrintAdapter (HTML), GotenbergAdapter (PDF via Gotenberg)

13. Workflow & Document States​

Framework M provides a pluggable WorkflowProtocol with independent adapters. Workflows are defined as DocTypes, keeping business state machines decoupled from controller logic.

Frappe​

# Workflow defined in UI, transitions via API
frappe.get_doc("Invoice", name).set_workflow_action("Approve")

Framework M (WorkflowProtocol)​

# Workflow defined as DocType
# Workflow: "InvoiceApproval"
#   - States: Draft, Pending Approval, Approved, Rejected
#   - Transitions: Submit (Draft → Pending), Approve (Pending → Approved), etc.
#   - allowed_roles per transition

# DocType references the workflow
class Invoice(BaseDocType):
    class Meta:
        workflow = "InvoiceApproval"

# Transitions via injected WorkflowProtocol
from framework_m_core.interfaces import WorkflowProtocol, TransitionRequest
from framework_m_core.interfaces import UserContext

class InvoiceService:
    def __init__(self, workflow: WorkflowProtocol):
        self.workflow = workflow

    async def approve(self, invoice_id: str, user: UserContext):
        # Checks permissions, validates transition, updates state, emits event
        await self.workflow.transition(TransitionRequest(
            doctype="Invoice",
            doc_id=invoice_id,
            action="Approve",
            user=user,
        ))

# Available adapters:
# - InternalWorkflowAdapter: Simple state machine, stores state in DB
# - TemporalWorkflowAdapter: Distributed, long-running workflows via Temporal.io

Why Protocol-based?

  • Swap implementations without changing business logic
  • Internal adapter for simple use cases
  • Temporal for complex, distributed workflows
  • Workflow definitions are data, not code

14. Naming Series​

Framework M uses declarative naming patterns via Meta.name_pattern. It employs optimistic name generation with retry on conflict, avoiding row-level locking on counter tables.

Frappe​

# In DocType, autoname = "naming_series:"
doc.naming_series = "INV-.YYYY.-"

Framework M (Declarative Naming)​

class Invoice(BaseDocType):
    class Meta:
        # Declarative pattern - framework handles generation
        name_pattern = "INV-.YYYY.-.####"  # → INV-2026-0001

# Pattern tokens:
#   .YYYY  → Year (2026)
#   .MM    → Month (01-12)
#   .DD    → Day (01-31)
#   .####  → Sequential number (optimistic, retry on conflict)
#   {field} → Field value (e.g., {customer_code})

# For high-volume DocTypes, use database-native sequences:
class HighVolumeInvoice(BaseDocType):
    class Meta:
        name_pattern = "sequence:invoice_seq"  # Uses DB sequence

# Framework detects database and uses appropriate mechanism:
#   PostgreSQL: CREATE SEQUENCE + nextval()
#   MySQL/MariaDB: AUTO_INCREMENT helper table
#   SQLite: sqlite_sequence table
#   Oracle: CREATE SEQUENCE
#   MSSQL: CREATE SEQUENCE (2012+) or IDENTITY

Why this approach?

  • id (UUID) is the primary key—zero contention on insert
  • name is optional human-readable identifier
  • No SELECT ... FOR UPDATE lock on counters (Frappe anti-pattern)
  • Optimistic generation: insert → generate name → retry if conflict

15. Fixtures & Data Import​

Frappe​

// fixtures in hooks.py
fixtures = ["Custom Field", {"dt": "Role", "filters": [["name", "in", ["Manager"]]]}]

Framework M​

# CLI command for fixtures
@app.command
async def load_fixtures(path: str = "fixtures/"):
    """Load JSON/YAML fixtures into database."""
    async with uow_factory() as uow:
        for file in Path(path).glob("*.json"):
            data = json.loads(file.read_text())
            for record in data:
                await repo.save(uow.session, Model(**record))
        await uow.commit()

16. Scheduled Jobs​

Frappe​

# hooks.py
scheduler_events = {
    "daily": ["myapp.tasks.daily_cleanup"],
    "cron": {"0 9 * * *": ["myapp.tasks.morning_report"]},
}

Framework M (JobQueueProtocol Scheduling)​

Schedules are managed via the JobQueueProtocol or adapter-specific decorators.

# Programmatic scheduling via Protocol
await queue.schedule("morning_report", cron="0 9 * * *")

# Or Decorator-based (Taskiq Adapter)
from taskiq import cron

@broker.task
@cron("0 9 * * *")
async def morning_report():
    await report_service.generate_daily()

@broker.task
@cron("0 0 * * *")
async def daily_cleanup():
    await cleanup_service.run()

17. Cache​

Frappe​

# Get/set cache with TTL
frappe.cache().set_value("key", value, expires_in_sec=3600)
value = frappe.cache().get_value("key")

# Delete pattern
frappe.cache().delete_keys("user:*")

Framework M​

from framework_m_core.interfaces import CacheProtocol

class ItemService:
    def __init__(self, cache: CacheProtocol, repo: RepositoryProtocol[Item]):
        self.cache = cache
        self.repo = repo

    async def get_item(self, item_id: str) -> Item:
        # Try cache first
        cached = await self.cache.get(f"item:{item_id}")
        if cached:
            return Item.model_validate(cached)

        # Fetch from DB and cache
        item = await self.repo.get(UUID(item_id))
        await self.cache.set(f"item:{item_id}", item.model_dump(), ttl=3600)
        return item

    async def invalidate_item(self, item_id: str):
        await self.cache.delete(f"item:{item_id}")

    async def invalidate_all_items(self):
        deleted = await self.cache.delete_pattern("item:*")
        return deleted

# Adapters: RedisCacheAdapter, MemoryCacheAdapter (dev)

18. Audit Trail / Version History​

Frappe​

# Auto-tracks changes in Version doctype
doc.get_doc_before_save()  # Compare changes

# Add comment
doc.add_comment("Updated stock levels")

# frappe.log_error for exceptions
frappe.log_error("Something went wrong")

Framework M​

from framework_m_core.interfaces import AuditLogProtocol, AuditEntry
from framework_m_core.interfaces import UserContext

class InvoiceService:
    def __init__(self, audit: AuditLogProtocol, repo: RepositoryProtocol[Invoice]):
        self.audit = audit
        self.repo = repo

    async def update_invoice(self, invoice_id: str, data: dict, user: UserContext):
        old = await self.repo.get(UUID(invoice_id))
        # ... update logic ...
        new = await self.repo.save(updated_invoice)

        # Log the change with diff
        await self.audit.log(
            user_id=user.id,
            action="update",
            doctype="Invoice",
            document_id=invoice_id,
            changes={"status": {"old": old.status, "new": new.status}},
        )

    async def get_history(self, invoice_id: str) -> list[AuditEntry]:
        return await self.audit.query(
            filters={"doctype": "Invoice", "document_id": invoice_id},
            limit=50,
        )

# Adapters:
# - DatabaseAuditAdapter: Writes to ActivityLog table (Indie)
# - FileAuditAdapter: JSONL file for Splunk/Filebeat
# - ElasticAuditAdapter: Elasticsearch (Enterprise)

The SearchProtocol provides a unified interface for full-text search, with support for adapters like Meilisearch and Elasticsearch.

Frappe​

# Link field search
frappe.get_list("Customer", filters={"name": ["like", "%search%"]})

# Global search
frappe.search_link("Customer", "john")

Framework M​

from framework_m_core.interfaces import SearchProtocol, SearchResult

class CustomerService:
    def __init__(self, search: SearchProtocol, repo: RepositoryProtocol[Customer]):
        self.search = search
        self.repo = repo

    async def search_customers(self, query: str) -> SearchResult:
        return await self.search.search(
            doctype="Customer",
            query=query,
            limit=20,
        )

    async def index_customer(self, customer: Customer):
        # Called after save to keep search index updated
        await self.search.index(
            doctype="Customer",
            doc_id=str(customer.id),
            doc=customer.model_dump(),
        )

# Adapters: MeilisearchAdapter, ElasticsearchAdapter, WhooshAdapter (dev)

20. Real-time Notifications​

The NotificationProtocol supports multi-channel notifications (In-app, Email, SMS). Real-time updates are handled via the SocketProtocol (WebSocket adapter).

Frappe​

# Push to browser
frappe.publish_realtime("invoice_updated", {"id": doc.name}, user=user)

# Create notification
frappe.get_doc({
    "doctype": "Notification Log",
    "for_user": user,
    "subject": "Invoice approved",
}).insert()

Framework M​

from framework_m_core.interfaces import NotificationProtocol
from framework_m_core.interfaces import UserContext

class InvoiceService:
    def __init__(self, notifications: NotificationProtocol):
        self.notifications = notifications

    async def approve_invoice(self, invoice_id: str, user: UserContext):
        # ... approval logic ...

        # In-app notification (stored in DB, shown in UI)
        await self.notifications.send_in_app(
            user_id=invoice.owner,
            title="Invoice Approved",
            body=f"Invoice {invoice.name} has been approved",
            link=f"/app/Invoice/{invoice_id}",
        )

        # Or email notification
        await self.notifications.send_email(
            to=[invoice.owner_email],
            subject="Invoice Approved",
            body=f"Your invoice {invoice.name} has been approved.",
            template="invoice_approved",
            context={"invoice": invoice.model_dump()},
        )

# WebSocket endpoint: WS /api/v1/notifications/stream
# Real-time push to connected clients via SocketProtocol

21. Virtual DocTypes (External Data Sources)​

Frappe​

# Virtual DocType - data from external API, not database
class ExternalProduct(Document):
    def db_insert(self):
        # Custom insert logic
        api.create_product(self.as_dict())

    def load_from_db(self):
        data = api.get_product(self.name)
        self.update(data)

Framework M​

# Implement RepositoryProtocol for any data source
class ExternalProductRepository(RepositoryProtocol[Product]):
    def __init__(self, api_client: APIClient):
        self.api = api_client

    async def get(self, session, id: UUID) -> Product | None:
        data = await self.api.get_product(str(id))
        return Product(**data) if data else None

    async def save(self, session, entity: Product) -> Product:
        await self.api.upsert_product(entity.model_dump())
        return entity

# Register override in container
container.repository_factory.register_override("Product", ExternalProductRepository)

22. Custom Fields & Dynamic Meta​

Frappe​

# Get doctype meta with custom fields
meta = frappe.get_meta("Invoice")
for field in meta.fields:
    print(field.fieldname, field.fieldtype)

# Check if field exists
if meta.has_field("custom_discount"):
    pass

Framework M​

# Meta endpoint returns full schema including custom fields
# GET /api/v1/meta/Invoice

# In code - MetaRegistry
from framework_m_core.registry import MetaRegistry

registry = MetaRegistry.get_instance()
meta = registry.get_doctype("Invoice")

for field in meta.fields:
    print(field.name, field.type)

# Custom fields via DocType overrides
class ExtendedInvoice(Invoice):
    custom_discount: Decimal = Field(default=Decimal("0"))

registry.register_override("Invoice", ExtendedInvoice)

23. Currency & Number Formatting​

Frappe​

frappe.format_value(1234567.89, {"fieldtype": "Currency"})  # ₹12,34,567.89
frappe.format_value(0.15, {"fieldtype": "Percent"})  # 15%

Framework M​

# Formatting via Python's babel library (locale-aware)
from babel.numbers import format_currency, format_percent

class InvoiceService:
    def format_total(self, amount: Decimal, locale: str = "en_IN") -> str:
        return format_currency(amount, "INR", locale=locale)
        # → ₹12,34,567.89

    def format_rate(self, rate: Decimal, locale: str = "en_IN") -> str:
        return format_percent(rate, locale=locale)
        # → 15%

# Frontend: use Intl API
# new Intl.NumberFormat("en-IN", { style: "currency", currency: "INR" }).format(1234567.89)

[!NOTE] I18nProtocol covers text translation only. Number/currency formatting uses Python's babel library directly or a separate FormatterProtocol (planned).

24. File Attachments​

Frappe​

# Attach file to document
file_doc = frappe.get_doc({
    "doctype": "File",
    "file_url": "/files/invoice.pdf",
    "attached_to_doctype": "Invoice",
    "attached_to_name": "INV-001",
})
file_doc.insert()

# Get attachments
files = frappe.get_all("File", filters={"attached_to_name": "INV-001"})

Framework M​

# StorageProtocol for file operations
from framework_m_core.interfaces import StorageProtocol

class InvoiceService:
    def __init__(self, storage: StorageProtocol, repo: RepositoryProtocol[Attachment]):
        self.storage = storage
        self.repo = repo

    async def attach_file(self, invoice_id: str, file: UploadFile) -> str:
        # Upload to S3/local/etc.
        content = await file.read()
        path = await self.storage.save_file(
            path=f"invoices/{invoice_id}/{file.filename}",
            content=content,
            content_type=file.content_type,
        )

        # Store attachment reference
        attachment = Attachment(
            doctype="Invoice",
            doc_id=invoice_id,
            file_url=path,
            filename=file.filename,
        )
        await self.repo.save(attachment)
        return path

25. Email​

Frappe​

frappe.sendmail(
    recipients=["user@example.com"],
    subject="Invoice Ready",
    template="invoice_notification",
    args={"invoice": invoice.as_dict()},
)

Framework M​

# EmailSenderProtocol for sending emails
from framework_m_core.interfaces import EmailSenderProtocol, EmailMessage
from jinja2 import Environment

class InvoiceService:
    def __init__(self, email_sender: EmailSenderProtocol, jinja: Environment):
        self.email_sender = email_sender
        self.jinja = jinja

    async def send_notification(self, invoice: Invoice):
        template = self.jinja.get_template("invoice_notification.html")
        html = template.render(invoice=invoice.model_dump())

        message = EmailMessage(
            to=["user@example.com"],
            subject="Invoice Ready",
            html_body=html,
        )
        result = await self.email_sender.send(message)
        return result.success

# Adapters: SMTPEmailSender, LogEmailSender (dev), etc.

26. Linked Documents & Dynamic Links​

Frappe​

# Get all linked documents
linked = frappe.get_all("Dynamic Link", filters={
    "link_doctype": "Customer",
    "link_name": "CUST-001",
})

# Dynamic Link field
class Contact(Document):
    links = []  # Child table with link_doctype, link_name

Framework M​

    # Explicit links (using UUID + metadata)
    customer_id: UUID | None = Field(default=None, json_schema_extra={"link": "Customer"})
    supplier_id: UUID | None = Field(default=None, json_schema_extra={"link": "Supplier"})

    # Or polymorphic links via child table
    links: list[DynamicLink] = Field(default_factory=list)

class DynamicLink(BaseDocType):
    class Meta:
        is_child_table = True

    link_doctype: str
    link_id: UUID

# Query linked documents
async def get_customer_contacts(self, customer_id: str):
    return await self.repo.list(
        session,
        filters=[FilterSpec("customer", FilterOperator.EQ, customer_id)],
        includes=["customer"],  # Eager load
    )

27. Reports (Query Reports / Script Reports)​

Frappe​

# Script Report
def execute(filters):
    columns = [{"fieldname": "name", "label": "Name", "fieldtype": "Data"}]
    data = frappe.db.sql("""SELECT name FROM tabItem WHERE ...""", as_dict=True)
    return columns, data

Framework M​

# Report as a Service with typed response
class ItemReportService:
    def __init__(self, repo: GenericRepository[Item]):
        self.repo = repo

    async def execute(self, filters: ItemReportFilters) -> ReportResult:
        items = await self.repo.list(
            session,
            filters=filters.to_filter_specs(),
        )
        return ReportResult(
            columns=[
                Column(name="name", label="Name", type="string"),
                Column(name="stock_qty", label="Stock", type="number"),
            ],
            data=[item.model_dump() for item in items],
        )

# Endpoint
@get("/api/reports/item-stock")
async def item_stock_report(
    filters: ItemReportFilters,
    report_service: ItemReportService,
) -> ReportResult:
    return await report_service.execute(filters)

28. Single DocTypes (Settings)​

Frappe​

# Single DocType - only one record exists
settings = frappe.get_single("System Settings")
settings.enable_feature = True
settings.save()

Framework M​

# Single DocType via Meta flag
class SystemSettings(BaseDocType):
    class Meta:
        is_single = True  # Only one record

    enable_feature: bool = False
    default_currency: str = "INR"

# Service for settings
class SettingsService:
    def __init__(self, repo: GenericRepository[SystemSettings]):
        self.repo = repo
        self._cache: SystemSettings | None = None

    async def get(self, session) -> SystemSettings:
        if self._cache is None:
            self._cache = await self.repo.get_single(session)
        return self._cache

    async def update(self, session, data: dict) -> SystemSettings:
        settings = await self.get(session)
        for key, value in data.items():
            setattr(settings, key, value)
        await self.repo.save(session, settings)
        self._cache = None  # Invalidate cache
        return settings

29. Multi-tenancy​

Frappe​

# Multi-site via bench
frappe.local.site  # Current site
# Each site = separate database

# Multi-company within site
filters = {"company": frappe.defaults.get_user_default("Company")}

Framework M​

# TenantContext injected via middleware
class ItemService:
    async def list_items(self, tenant: TenantContext, session):
        # Automatic tenant filtering via RLS or explicit filter
        return await self.repo.list(
            session,
            filters=[FilterSpec("tenant_id", FilterOperator.EQ, tenant.id)],
        )

# Tenancy modes (configured via adapter):
# - ImplicitTenantAdapter: Single tenant (indie deployments)
# - HeaderTenantAdapter: Multi-tenant via X-Tenant-ID header
# - DatabasePerTenantAdapter: Separate DB per tenant

# Frontend: tenant from JWT or API response
# Backend: automatic RLS filtering when enabled

Part 2: Frontend Migration (Frappe UI → Framework M)​

Quick Reference​

Frappe UIFramework M / React Native / React
frappe.ui.form.FormFormView.tsx file-based override
frappe.ListviewListView.tsx file-based override
frappe.call()useCall() (Idempotent)
cur_frmReact state / props
frappe.route_optionsRoute Traversals / React Router
frappe.msgprint()Toast / notification component
frappe.confirm()Modal / dialog component
frappe.ui.DialoguseModal()

30. Forms​

Frappe​

frappe.ui.form.on("Invoice", {
    customer: function(frm) {
        // Called when customer field changes
        frappe.call({
            method: "myapp.api.get_customer_address",
            args: { customer: frm.doc.customer },
            callback: (r) => frm.set_value("address", r.message)
        });
    },
    refresh: function(frm) {
        frm.add_custom_button("Send Email", () => { ... });
    }
});

Framework M (AutoForm / Uniforms)​

Customizing the form logic is done via a file-based override. Use AutoForm for schema-driven generation or build a fully bespoke UI using Tamagui primitives.

// src/overrides/Invoice/FormView.tsx
import { useEffect } from "react";
import { useForm } from "@refinedev/react-hook-form";
import { AutoForm, frameworkMDataProvider } from "@framework-m/desk";
import { YStack, Card, Heading, MText } from "@framework-m/ui";

export default function CustomInvoiceForm({ doctype, id, children }: FormViewProps) {
  const {
    refineCore: { onFinish, queryResult },
    watch,
    setValue
  } = useForm();

  const customer = watch("customer");

  // React to field changes using standard React hooks
  useEffect(() => {
    if (customer) {
      // Use dataProvider.custom for RPC/Method calls
      frameworkMDataProvider.custom({
        url: `/rpc/fn/myapp.api.get_customer_address`,
        method: "POST",
        payload: { customer }
      }).then(res => {
        setValue("address", res.data.address);
      });
    }
  }, [customer]);

  return (
    <YStack gap="$md" padding="$md">
      <Card elevated padding="$sm" backgroundColor="$primaryMuted">
        <Heading size="$sm">Invoice Customization</Heading>
      </Card>

      {/* AutoForm handles schema-to-UI mapping */}
      <AutoForm
        schema={queryResult?.data?.data?.schema}
        onSubmit={onFinish}
      />
    </YStack>
  );
}

31. List Views​

Frappe​

frappe.listview_settings["Invoice"] = {
    add_fields: ["status", "grand_total"],
    get_indicator: (doc) => [doc.status, doc.status === "Paid" ? "green" : "orange"],
    onload: (listview) => {
        listview.page.add_action_item("Bulk Approve", () => { ... });
    }
};

Framework M (AutoTable)​

Override the list view for a DocType using AutoTable.

// src/overrides/Invoice/ListView.tsx
import { AutoTable } from "@framework-m/desk";
import { YStack, XStack, Button, Heading } from "@framework-m/ui";

export default function CustomInvoiceList({ doctype }: ListViewProps) {
  const handleBulkApprove = () => {
    // Custom bulk approval logic
  };

  return (
    <YStack gap="$md" flex={1} padding="$md">
      <XStack justifyContent="space-between" alignItems="center">
        <Heading>Invoices</Heading>
        <Button onPress={handleBulkApprove}>Bulk Approve</Button>
      </XStack>

      {/* AutoTable handles server-side sorting, filtering, and pagination */}
      <AutoTable resource="Invoice" />
    </YStack>
  );
}

32. RPC Calls​

Frappe​

frappe.call({
  method: "myapp.api.get_stock_balance",
  args: { item_code: "ITEM-001" },
  callback: r => console.log(r.message),
});

Framework M (useCall Hook)​

Use the useCall hook from @framework-m/desk for RPC and custom endpoints. It automatically handles authentication headers, base URL resolution, and mutation safety (Idempotency).

import { useCall } from "@framework-m/desk";
import { Button } from "@framework-m/ui";

export function StockBalanceButton({ itemCode }) {
  const { execute, isLoading, data } = useCall();

  const handleCheck = async () => {
    // Automatically injects Idempotency-Key header to prevent duplicate side-effects
    await execute({
      doctype: "Stock",
      method: "get_balance",
      payload: { item_code: itemCode }
    });
  };

  return (
    <Button onPress={handleCheck} disabled={isLoading}>
      {isLoading ? "Checking..." : `Check Balance`}
    </Button>
  );
}

[!TIP] Why useCall? In Frappe, clicking a frappe.call button twice during a network delay might trigger the backend code twice. useCall generates an Idempotency-Key under the hood. If the network drops and the browser retries the request, the Framework M backend intercepts the duplicate key and safely returns the cached response without running the Python logic again.

33. Dialogs and Modals​

Frappe​

frappe.prompt(
  [{ fieldname: "reason", label: "Reason", fieldtype: "Small Text" }],
  values => {
    frappe.call({ method: "cancel_invoice", args: { ...values } });
  },
  "Cancel Invoice",
);

Framework M (Feedback Components)​

Use ConfirmDialog from @framework-m/desk for standard confirmation flows, or build a custom modal using @framework-m/ui layout primitives.

import { ConfirmDialog, frameworkMDataProvider } from "@framework-m/desk";
import { Button } from "@framework-m/ui";
import { useState } from "react";

export const CancelInvoiceButton = ({ invoiceId }) => {
  const [showConfirm, setShowConfirm] = useState(false);

  const handleCancel = async () => {
    await frameworkMDataProvider.custom({
      url: "/rpc/fn/myapp.invoice.cancel",
      method: "POST",
      payload: { invoiceId }
    });
    setShowConfirm(false);
  };

  return (
    <>
      <Button onPress={() => setShowConfirm(true)}>Cancel Invoice</Button>

      <ConfirmDialog
        open={showConfirm}
        onOpenChange={setShowConfirm}
        title="Cancel Invoice"
        message="Are you sure you want to permanently cancel this invoice?"
        onConfirm={handleCancel}
        confirmText="Confirm Cancellation"
        cancelText="Back"
      />
    </>
  );
};

34. Real-time Updates​

Frappe​

frappe.realtime.on("invoice_updated", data => {
  cur_frm.reload_doc();
});

Framework M (liveProvider)​

Framework M uses a standard Refine liveProvider backed by WebSockets.

import { useEffect, useState } from "react";
import { useShell } from "@framework-m/desk";
import { YStack, MText } from "@framework-m/ui";

export function RealtimeInvoiceStatus() {
  const { realtime } = useShell();
  const [status, setStatus] = useState("Draft");

  useEffect(() => {
    // Protocol-first: use shell's realtime adapter
    const stop = realtime.connect((event) => {
      if (event.room === "Invoice" && event.type === "updated") {
        setStatus(event.payload.status);
      }
    });

    return () => stop();
  }, [realtime]);

  return (
    <YStack p="$4" backgroundColor="$background">
      <MText variant="label">Current Status:</MText>
      <MText fontWeight="bold" color="$blue10">{status}</MText>
    </YStack>
  );
}

Minimal Patterns: Rapid Development in M​

[!WARNING] Use Reasonably: These patterns bypass some standard architectural layers (Services, structured UoW) for speed. Ideal for prototypes, scripts, and simple CRUD. For complex business logic, prefer the full Service/Controller/Repository layers.

Philosophy: M's DI system works with function parameters, not just classes. You don't need service classes for simple cases, use DI directly.

The Spectrum of Complexity​

Frappe-Like (Quick)                              Enterprise (Structured)
│                                                           │
▼                                                           ▼
Minimal DocType ─► RPC Function ─► Simple Service ─► Full Service Layer
  (data only)       (with DI)        (single class)   (repos, UoW, events)

Start minimal. Add structure when complexity demands it.

Pattern 1: Minimal DocType (No Controller)​

Controllers are optional. For simple data storage, this is all you need:

# feedback.py - entire file
from framework_m import DocType, Field

class Feedback(DocType):
    """User feedback - just storage, no business logic."""
    subject: str
    message: str
    rating: int = Field(ge=1, le=5, description="1-5 stars")
    email: str | None = None

That's it. You get:

  • ✅ Database table created automatically
  • ✅ REST API: POST /api/v1/Feedback, GET /api/v1/Feedback, etc.
  • ✅ Validation (Pydantic enforces rating is 1-5)
  • ✅ JSON Schema for frontend forms

Frappe equivalent lines: ~15 (with naming, mandatory setup) M lines: 8

Pattern 2: RPC Function with DI (Frappe-style APIs)​

For custom endpoints, use @rpc with injected dependencies directly in function params:

# stock_api.py
from typing import Any
from framework_m_core.di import inject, Provide
from framework_m_core.decorators import rpc
from framework_m import Container

@rpc()
@inject
async def get_stock_balance(
    item_code: str,
    warehouse: str,
    # Repository would typically be accessed via unit_of_work or a custom provider
    uow: Any = Provide[Container.unit_of_work],
) -> dict:
    """Get stock balance for item at warehouse."""
    async with uow:
       # Accessing the session for the given stock-related query
       result = await uow.session.execute(...)
       # ...
    return {"item_code": item_code, "warehouse": warehouse, "qty": total}

Called as: POST /api/v1/rpc/fn/stock_api.get_stock_balance (Dotted Python path)

{ "item_code": "ITEM-001", "warehouse": "Main" }

No class needed. The @inject decorator handles DI in the function.

Pattern 3: Quick Validation (Controller Light)​

Need just validation? Add a minimal controller:

# feedback.py
from framework_m import DocType, Field, Controller

class Feedback(DocType):
    subject: str
    message: str
    rating: int = Field(ge=1, le=5)
    email: str | None = None

class FeedbackController(Controller[Feedback]):
    """Only validate - no other hooks needed."""

    async def validate(self, context=None) -> None:
        if len(self.doc.message) < 10:
            raise ValueError("Message must be at least 10 characters")

Only implement what you need. Empty hooks are inherited from BaseController.

Pattern 4: Direct Database Access (The "Hack")​

For quick scripts or migrations, you can use the session directly:

from sqlalchemy import text
from framework_m_core.decorators import rpc
from framework_m_core.di import inject, Provide
from framework_m import Container
from framework_m_core.unit_of_work import UnitOfWork

@rpc() # Path auto-generated as "stock_reports.sales_summary"
@inject
async def sales_summary(
    year: int,
    uow: UnitOfWork = Provide[Container.unit_of_work],
) -> list[dict]:
    """Raw SQL for complex reporting - this is the 'hack' path."""
    async with uow:
        result = await uow.session.execute(text("""
            SELECT
                customer,
                SUM(grand_total) as total,
                COUNT(*) as count
            FROM invoice
            WHERE EXTRACT(YEAR FROM posting_date) = :year
            AND docstatus = 1
            GROUP BY customer
            ORDER BY total DESC
            LIMIT 10
        """), {"year": year})
        return [dict(row._mapping) for row in result]
WARNING

Raw SQL bypasses permissions, validation, and audit. Use for read-only reports or one-off scripts. For production business logic, use Repositories.

Pattern 5: Multiple RPCs, One File​

Group related functions without classes:

# inventory_api.py
from framework_m_core.decorators import rpc
from framework_m_core.di import inject, Provide
from framework_m import Container

@rpc()
@inject
async def transfer_stock(
    item: str,
    from_wh: str,
    to_wh: str,
    qty: float,
    # Service/Repo would be injected from an AppContainer or custom provider
    stock_svc: Any = Provide["stock_service"],
) -> dict:
    result = await stock_svc.transfer(item, from_wh, to_wh, qty)
    return {"success": True, "entry_id": str(result.id)}

@rpc()
@inject
async def adjust_stock(
    item: str,
    warehouse: str,
    qty: float,
    reason: str,
    stock_svc: Any = Provide["stock_service"],
) -> dict:
    result = await stock_svc.adjust(item, warehouse, qty, reason)
    return {"success": True, "entry_id": str(result.id)}

@rpc()
@inject
async def get_stock_levels(
    warehouse: str | None = None,
    # Repo from an AppContainer
    repo: Any = Provide["item_repo"],
) -> list[dict]:
    filters = [("warehouse", "=", warehouse)] if warehouse else []
    items = await repo.list(filters=filters, limit=1000)
    return [{"item": i.item_code, "qty": i.qty} for i in items.items]
```   return [{"item": i.item_code, "qty": i.qty} for i in items.items]

Three endpoints, no classes, full DI support.

Pattern 6: Background Job (Minimal)​

# jobs/cleanup.py
from typing import Any
from framework_m_core.di import inject, Provide
from framework_m_core.interfaces.job_queue import JobQueueProtocol
from framework_m import Container

# Handler is a PLAIN async function - no direct infra imports (taskiq/arq)
async def cleanup_old_logs(days: int = 30):
    """Business logic for cleanup."""
    # Injected via DI in the calling service
    print(f"Cleaning logs older than {days} days...")
    return {"success": True}

# Calling the job via the Protocol (from a service or controller)
@inject
async def trigger_cleanup(
    days: int,
    queue: JobQueueProtocol = Provide[Container.job_queue],
):
    # Enqueue by dotted path name - infra-agnostic
    await queue.enqueue("my_app.jobs.cleanup_old_logs", days=days)

Pattern 7: Explicit Translations (No Global State)​

In Frappe, you use a global translation helper _() anywhere in your backend. In Framework M, we avoid global helpers or ContextVars because they are coupled to the HTTP thread context and fail in non-web runtimes (CLI, background workers, and Temporal activities). Instead, we explicitly inject the I18nProtocol and pass the target locale:

Frappe (Implicit Global)​

# Anywhere in backend code
msg = _("Quantity must be greater than zero")

Framework M (Explicit Protocol)​

from framework_m_core.interfaces import I18nProtocol
from framework_m_core.di import inject, Provide
from framework_m_core.container import Container

class StockService:
    @inject
    def __init__(self, i18n: I18nProtocol = Provide[Container.i18n]):
        self.i18n = i18n
        # Local alias for clean DX
        self.t = i18n.translate

    async def validate_qty(self, qty: int, user: UserContext):
        if qty <= 0:
            # Explicit, mockable, and portable across CLI, Workers, and Temporal
            raise ValueError(await self.t(
                "Quantity must be greater than zero", 
                locale=user.locale
            ))

RPC Function Example (Explicit DI & Locale Parameter)​

For standalone endpoints decorated with @rpc, inject the translation service and explicitly pass the client context/locale:

from typing import Any
from framework_m_core.decorators import rpc
from framework_m_core.di import inject, Provide
from framework_m_core.interfaces import I18nProtocol, t
from framework_m_core.container import Container
from framework_m_core.models import UserContext
from sqlalchemy.ext.asyncio import AsyncSession

@rpc()
@inject
async def get_translated_welcome(
    user: UserContext,
    session: AsyncSession,
    i18n: I18nProtocol = Provide[Container.i18n],
) -> dict[str, Any]:
    # Translate explicit message with namespace parameter
    welcome_msg = await i18n.translate(
        t("Welcome"),
        locale=user.locale,
        session=session,
        namespace="desk"
    )
    return {"message": welcome_msg}

When to Add Structure​

SituationPattern
Simple data storageMinimal DocType (no controller)
Custom API endpoint@rpc function with DI
Simple validationController with only validate()
Complex business logicService class
Multi-step transactionsService + UnitOfWork
External integrationsFull Protocol/Adapter

Comparison: Lines of Code​

TaskFrappeM (Minimal)M (Full)
Simple DocType1588
DocType + validation201515
Custom API endpoint81225
Stock balance query51020
Background job101015

M is 30-50% more verbose for minimal cases, but the gap shrinks as complexity grows.

Summary: Key Mindset Shifts​

FrappeFramework MWhy
Global frappe.dbInjected repoTestability, no hidden state
Implicit transactionsExplicit UoWClear boundaries, multiple txns
Hooks on DocTypeHooks on ControllerSeparation of data and logic
frappe.call()Typed RPC (@rpc)IDE support, validation
frappe.enqueue()JobQueueProtocolPluggable backends (Indie/Ent)
Global helper _()Explicit i18n DIPortable across CLI, Jobs, and Temporal
frappe.realtimeuseShell().realtimeUnified shell protocol
cur_frm (global)React state/hooksComponent isolation
Jinja templatesTamagui PrimitivesReact Native compatibility
Server-rendered formsSchema-driven UIBetter UX, offline support

Next Steps​

  1. Backend: Create a new DocType and implement a Controller inheriting from Controller[T].
  2. Logic: Use the Repository pattern and UnitOfWork for data access and transactions.
  3. Background: Define plain async functions and enqueue them via JobQueueProtocol.
  4. Frontend: Use file-based overrides in src/overrides to customize standard views.
  5. UI: Build custom interfaces using @framework-m/ui (Tamagui) primitives.

See also: