/frappe-service <service_name> [--doctype <doctype>] [--operations <op1,op2>]

/frappe-service OrderProcessing --doctype "Sales Order"
/frappe-service InventoryManagement --operations allocate,release,transfer
/frappe-service PaymentGateway

"""
<Service Name> Service

<Detailed description of what this service handles>

Responsibilities:
    - <Responsibility 1>
    - <Responsibility 2>
    - <Responsibility 3>

Usage:
    from <app>.<module>.services.<service_name> import <ServiceName>Service

    service = <ServiceName>Service()
    result = service.process_order(order_data)
"""

import frappe
from frappe import _
from frappe.utils import now, today, flt, cint, cstr
from typing import TYPE_CHECKING, Optional, Any, Callable
from contextlib import contextmanager
from functools import wraps

from <app>.<module>.services.base import BaseService
from <app>.<module>.repositories.<doctype>_repository import <DocType>Repository

if TYPE_CHECKING:
    from frappe.model.document import Document

This service handles:
    - <Operation 1>
    - <Operation 2>
    - <Operation 3>

Architecture:
    Controller/API → Service → Repository → Database

Example:
    service = <ServiceName>Service()
    order = service.create_order(customer="CUST-001", items=[...])
    service.submit_order(order.name)
"""

def __init__(self, user: Optional[str] = None):
    super().__init__(user)
    self.repo = <DocType>Repository()
    # Initialize other repositories as needed
    # self.item_repo = ItemRepository()
    # self.customer_repo = CustomerRepository()

# ──────────────────────────────────────────────────────────────────────────
# Public Operations (Business Logic)
# ──────────────────────────────────────────────────────────────────────────

@require_permission("<DocType>", "create")
@with_transaction
@log_operation("create_<doctype>")
def create(self, data: dict) -> dict:
    """
    Create a new <DocType>.

    Args:
        data: Document data containing:
            - title (str): Required title
            - date (str): Date in YYYY-MM-DD format
            - description (str): Optional description

    Returns:
        Created document summary

    Raises:
        frappe.ValidationError: If validation fails
        frappe.PermissionError: If user lacks permission

    Example:
        service.create({
            "title": "New Order",
            "date": "2024-01-15"
        })
    """
    # 1. Validate input
    self._validate_create_data(data)

    # 2. Apply business rules
    data = self._apply_defaults(data)
    data = self._apply_business_rules(data)

    # 3. Create via repository
    doc = self.repo.create(data)

    # 4. Post-creation actions
    self._on_create(doc)

    # 5. Return summary
    return doc.get_summary()

@require_permission("<DocType>", "write")
@with_transaction
def update(self, name: str, data: dict) -> dict:
    """
    Update existing <DocType>.

    Args:
        name: Document name
        data: Fields to update

    Returns:
        Updated document summary
    """
    doc = self.repo.get_or_throw(name, for_update=True)

    # Validate update is allowed
    self._validate_can_update(doc)

    # Apply update
    doc.update(data)
    doc.save()

    return doc.get_summary()

@require_permission("<DocType>", "submit")
@with_transaction
@log_operation("submit_<doctype>")
def submit(self, name: str) -> dict:
    """
    Submit document for processing.

    This triggers:
        1. Pre-submission validation
        2. Document submission
        3. Post-submission actions (e.g., stock updates, GL entries)

    Args:
        name: Document name

    Returns:
        Submitted document summary

    Raises:
        frappe.ValidationError: If submission requirements not met
    """
    doc = self.repo.get_or_throw(name, for_update=True)

    # Pre-submission checks
    self._validate_submission(doc)

    # Submit
    doc.submit()

    # Post-submission processing
    self._on_submit(doc)

    return doc.get_summary()

@require_permission("<DocType>", "cancel")
@with_transaction
@log_operation("cancel_<doctype>")
def cancel(self, name: str, reason: Optional[str] = None) -> dict:
    """
    Cancel submitted document.

    Args:
        name: Document name
        reason: Cancellation reason (recommended)

    Returns:
        Cancelled document summary
    """
    doc = self.repo.get_or_throw(name, for_update=True)

    # Validate cancellation
    self._validate_cancellation(doc)

    # Store reason
    if reason:
        doc.db_set("cancellation_reason", reason, update_modified=False)

    # Cancel
    doc.cancel()

    # Post-cancellation processing
    self._on_cancel(doc)

    return doc.get_summary()

# ──────────────────────────────────────────────────────────────────────────
# Complex Business Operations
# ──────────────────────────────────────────────────────────────────────────

@with_transaction
def process_workflow(
    self,
    name: str,
    action: str,
    comment: Optional[str] = None
) -> dict:
    """
    Process workflow action on document.

    Args:
        name: Document name
        action: Workflow action (e.g., "Approve", "Reject")
        comment: Optional comment for the action

    Returns:
        Updated document with new workflow state
    """
    doc = self.repo.get_or_throw(name, for_update=True)

    # Validate action is allowed
    allowed_actions = self._get_allowed_workflow_actions(doc)
    if action not in allowed_actions:
        frappe.throw(
            _("Action '{0}' not allowed. Allowed: {1}").format(
                action, ", ".join(allowed_actions)
            )
        )

    # Apply workflow action
    from frappe.model.workflow import apply_workflow
    apply_workflow(doc, action)

    # Add comment
    if comment:
        doc.add_comment("Workflow", f"{action}: {comment}")

    return doc.get_summary()

def calculate_totals(self, name: str) -> dict:
    """
    Calculate and update document totals.

    Args:
        name: Document name

    Returns:
        Calculated totals
    """
    doc = self.repo.get_or_throw(name)

    subtotal = sum(
        flt(item.qty) * flt(item.rate)
        for item in doc.get("items", [])
    )

    tax_amount = flt(subtotal) * flt(doc.tax_rate or 0) / 100
    grand_total = flt(subtotal) + flt(tax_amount)

    return {
        "subtotal": subtotal,
        "tax_amount": tax_amount,
        "grand_total": grand_total
    }

def bulk_operation(
    self,
    names: list[str],
    operation: str,
    **kwargs
) -> dict:
    """
    Perform bulk operation on multiple documents.

    Args:
        names: List of document names
        operation: Operation to perform (update_status, submit, cancel)
        **kwargs: Operation-specific arguments

    Returns:
        Results summary
    """
    results = {"success": [], "failed": []}

    for name in names:
        try:
            if operation == "update_status":
                self.update(name, {"status": kwargs.get("status")})
            elif operation == "submit":
                self.submit(name)
            elif operation == "cancel":
                self.cancel(name, kwargs.get("reason"))

            results["success"].append(name)
        except Exception as e:
            results["failed"].append({
                "name": name,
                "error": str(e)
            })

    return results

# ──────────────────────────────────────────────────────────────────────────
# Query Methods
# ──────────────────────────────────────────────────────────────────────────

def get_pending_items(self, limit: int = 50) -> list[dict]:
    """Get items pending action."""
    return self.repo.get_list(
        filters={"status": "Pending", "docstatus": 0},
        fields=["name", "title", "date", "owner", "creation"],
        order_by="creation asc",
        limit=limit
    )

def get_statistics(self, period: str = "month") -> dict:
    """
    Get statistics for dashboard.

    Args:
        period: Time period (day, week, month, year)

    Returns:
        Statistics dict
    """
    from frappe.utils import add_days, add_months, get_first_day

    today_date = today()

    if period == "day":
        from_date = today_date
    elif period == "week":
        from_date = add_days(today_date, -7)
    elif period == "month":
        from_date = get_first_day(today_date)
    else:  # year
        from_date = add_months(get_first_day(today_date), -12)

    return {
        "total": self.repo.get_count(),
        "period_total": self.repo.get_count(
            {"creation": [">=", from_date]}
        ),
        "by_status": self._get_counts_by_status(),
        "period": period,
        "from_date": from_date
    }

# ──────────────────────────────────────────────────────────────────────────
# Private Methods (Internal Logic)
# ──────────────────────────────────────────────────────────────────────────

def _validate_create_data(self, data: dict) -> None:
    """Validate data for document creation."""
    self.validate_mandatory(data, ["title"])

    # Custom validations
    if data.get("date") and data["date"] < today():
        frappe.throw(_("Date cannot be in the past"))

def _validate_can_update(self, doc: "Document") -> None:
    """Validate document can be updated."""
    if doc.docstatus == 2:
        frappe.throw(_("Cannot update cancelled document"))

    if doc.status == "Completed":
        frappe.throw(_("Cannot update completed document"))

def _validate_submission(self, doc: "Document") -> None:
    """Validate all requirements for submission."""
    if doc.docstatus != 0:
        frappe.throw(_("Document is not in draft state"))

    # Add more validations as needed
    # if not doc.get("items"):
    #     frappe.throw(_("Cannot submit without items"))

def _validate_cancellation(self, doc: "Document") -> None:
    """Validate document can be cancelled."""
    if doc.docstatus != 1:
        frappe.throw(_("Only submitted documents can be cancelled"))

    # Check for linked documents
    # linked = self._get_linked_submitted_docs(doc.name)
    # if linked:
    #     frappe.throw(_("Cannot cancel. Linked documents exist: {0}").format(linked))

def _apply_defaults(self, data: dict) -> dict:
    """Apply default values to data."""
    if not data.get("date"):
        data["date"] = today()

    if not data.get("status"):
        data["status"] = "Draft"

    return data

def _apply_business_rules(self, data: dict) -> dict:
    """Apply business rules to data."""
    # Example: Set posting date to today if not specified
    # Example: Calculate derived fields
    return data

def _on_create(self, doc: "Document") -> None:
    """Post-creation hook for additional processing."""
    # Send notification
    # frappe.publish_realtime("new_document", {"name": doc.name})
    pass

def _on_submit(self, doc: "Document") -> None:
    """Post-submission processing."""
    # Create linked records (GL entries, stock ledger, etc.)
    # Update inventory
    # Send notifications
    pass

def _on_cancel(self, doc: "Document") -> None:
    """Post-cancellation processing."""
    # Reverse linked records
    # Update inventory
    pass

def _get_allowed_workflow_actions(self, doc: "Document") -> list[str]:
    """Get allowed workflow actions for document."""
    from frappe.model.workflow import get_transitions
    return [t.action for t in get_transitions(doc)]

def _get_counts_by_status(self) -> dict:
    """Get document counts grouped by status."""
    result = frappe.db.sql("""
        SELECT status, COUNT(*) as count
        FROM `tab<DocType>`
        WHERE docstatus < 2
        GROUP BY status
    """, as_dict=True)

    return {row.status: row.count for row in result}

Use this instead of direct instantiation for easier testing/mocking.

Args:
    user: Optional user context

Returns:
    Service instance
"""
return <ServiceName>Service(user=user)

undefined

"""
External Integration Service

Handles communication with external APIs with retry logic,
error handling, and response normalization.
"""

import frappe
from frappe import _
from typing import Optional, Any
import requests
from tenacity import retry, stop_after_attempt, wait_exponential


class <Integration>Service:
    """
    Service for integrating with <External Service>.

    Configuration:
        - API Key: System Settings > <Integration> API Key
        - Base URL: System Settings > <Integration> Base URL
    """

    def __init__(self):
        self.api_key = frappe.db.get_single_value("System Settings", "<integration>_api_key")
        self.base_url = frappe.db.get_single_value("System Settings", "<integration>_base_url")

        if not self.api_key:
            frappe.throw(_("<Integration> API key not configured"))

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def _make_request(
        self,
        method: str,
        endpoint: str,
        data: Optional[dict] = None
    ) -> dict:
        """
        Make HTTP request with retry logic.

        Args:
            method: HTTP method (GET, POST, PUT, DELETE)
            endpoint: API endpoint
            data: Request payload

        Returns:
            Response data

        Raises:
            frappe.ValidationError: On API error
        """
        url = f"{self.base_url}/{endpoint}"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        try:
            response = requests.request(
                method=method,
                url=url,
                json=data,
                headers=headers,
                timeout=30
            )
            response.raise_for_status()
            return response.json()

        except requests.exceptions.Timeout:
            frappe.throw(_("Request timed out. Please try again."))

        except requests.exceptions.HTTPError as e:
            error_msg = self._parse_error_response(e.response)
            frappe.throw(_("API Error: {0}").format(error_msg))

        except requests.exceptions.RequestException as e:
            frappe.throw(_("Connection error: {0}").format(str(e)))

    def _parse_error_response(self, response) -> str:
        """Parse error message from API response."""
        try:
            data = response.json()
            return data.get("message") or data.get("error") or response.text
        except Exception:
            return response.text

    # Public API methods
    def create_external_record(self, data: dict) -> dict:
        """Create record in external system."""
        return self._make_request("POST", "records", data)

    def get_external_record(self, external_id: str) -> dict:
        """Get record from external system."""
        return self._make_request("GET", f"records/{external_id}")

    def sync_records(self) -> dict:
        """Sync records with external system."""
        # Implementation
        pass

undefined

┌─────────────────────┐
│   Controller/API    │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│   <ServiceName>     │ ← Business Logic
│      Service        │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│   <DocType>         │ ← Data Access
│   Repository        │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│     Database        │
└─────────────────────┘

undefined

undefined

from <app>.<module>.services.<service_name> import <ServiceName>Service

service = <ServiceName>Service()

undefined