service_layer_rules

Service Layer Rules

• Scope: Applies to backend/services/*.py.
• Goal: Implement core business logic and orchestrate complex workflows. Coordinate repositories/SDKs. Keep HTTP concerns out of this layer.
• Exceptions: Raise domain/service exceptions declared in backend/consts/exceptions.py. If a new case is needed, add a new class there, then raise it here. Do not translate to HTTP here.
• Environment variables: Do not access os.getenv() directly. Read configuration from consts.const (see environment_variable rule) or accept parameters.

Reference: backend/consts/exceptions.py

Correct example (service orchestrates business logic and raises domain exceptions)

# backend/services/agent_service.py
from typing import Any, Dict

from consts.exceptions import LimitExceededError, AgentRunException, MemoryPreparationException
# from consts.const import APPID, TOKEN  # Example: read config via consts, not os.getenv


def run_agent(task_payload: Dict[str, Any]) -> Dict[str, Any]:
    """Run agent core workflow and return domain result dict.
    Raises domain exceptions on failure; no HTTP concerns here.
    """
    if _is_rate_limited(task_payload):
        raise LimitExceededError("Too many requests for this tenant.")

    try:
        memory = _prepare_memory(task_payload)
    except Exception as exc:
        # Wrap low-level error in a domain exception for the app layer to translate
        raise MemoryPreparationException("Failed to prepare memory.") from exc

    try:
        result = _execute_core_logic(task_payload, memory)
    except Exception as exc:
        raise AgentRunException("Agent execution failed.") from exc

    # Return a plain Python object, not a Response
    return {"status": "ok", "data": result}


def _is_rate_limited(_: Dict[str, Any]) -> bool:
    return False


def _prepare_memory(_: Dict[str, Any]) -> Dict[str, Any]:
    return {"memo": "prepared"}


def _execute_core_logic(_: Dict[str, Any], __: Dict[str, Any]) -> Dict[str, Any]:
    return {"answer": "42"}