Inter-Service RPC

from singularity.rpc import rpc

class OrdersService:
  def __init__(self, acquire):
    self.acquire = acquire

  async def post(self, data: dict):
    """Create an order and charge the user via RPC."""
    payment = await rpc("orders").payments.post_charge(
      user_id=data["user_id"],
      amount=data["total"],
    )
    return {"order": "confirmed", "payment": payment}

class PaymentsService:
  """Payment processing service."""

  rpc_exposed = True
  blacklist = ["public", "analytics"]

  # Declare every non-core method you want reachable, over HTTP or RPC.
  http_exposed = ["post=charge", "post=refund"]

  def __init__(self, acquire):
    self.acquire = acquire

  async def post_charge(self, user_id: str, amount: float):
    """Charge a user's payment method."""
    return {"charged": amount, "user": user_id, "transaction_id": "txn_abc123"}

  async def post_refund(self, payment_id: str):
    """Refund a payment."""
    return {"refunded": True, "payment": payment_id}

  async def get(self):
    """List recent payments — also exposed as GET /api/v1/payments."""
    return {"payments": []}

  def _reconcile(self):
    """Internal helper — not an endpoint, so not reachable over RPC."""

await rpc("orders").payments.post_charge(user_id="u1", amount=9.99)

class UsersService:
  rpc_exposed = True
  http_exposed = ["get=by_email"]
  visibility = {"get_by_email": "protected"}  # RPC only — no HTTP route

  async def get_by_email(self, email: str):
    ...

class PaymentsService:
  rpc_exposed = True
  blacklist = ["public", "analytics"]
  http_exposed = ["post=charge"]

  async def post_charge(self, user_id: str, amount: float):
    return {"charged": amount}

# This works whether payments is local or remote
result = await rpc("orders").payments.post_charge(user_id="abc", amount=9.99)

RPC_HEARTBEAT_INTERVAL=30   # seconds between refreshes
RPC_SPEC_TTL=60             # TTL on published keys — keep it above the interval

uv run python app.py --dev --no-heartbeat

# On the calling deployment
RPC_API_KEY=shared-secret

# On the receiving deployment
RPC_REQUIRE_API_KEY=true
RPC_API_KEYS=shared-secret,rotating-secret   # comma-separated; defaults to RPC_API_KEY

curl http://localhost:8000/_rpc/spec | python -m json.tool

{
  "rpc_version": "1.0",
  "base_url": "http://localhost:8000",
  "services": {
    "payments": {
      "blacklist": ["public", "analytics"],
      "methods": {
        "post_charge": {
          "endpoint": "http://localhost:8000/api/v1/payments/charge",
          "http_method": "POST",
          "description": "Charge a user's payment method.",
          "parameters": {
            "user_id": { "type": "str", "required": true },
            "amount": { "type": "float", "required": true }
          },
          "returns": "Any"
        },
        "get": {
          "endpoint": "http://localhost:8000/api/v1/payments",
          "http_method": "GET",
          "description": "List recent payments.",
          "parameters": {},
          "returns": "Any"
        }
      }
    }
  }
}

from singularity import Microservice
from settings import Settings

settings = Settings()

class BillingMS(Microservice):
  host_url = settings.billing_host_url  # e.g., "http://billing-service:8001"
  api_key = settings.billing_api_key    # sent as X-API-Key header
  timeout = 10                          # request timeout in seconds
  prefix = "/api"                       # API prefix on the remote repo

class InvoiceService:
  billing = BillingMS()

  def __init__(self, acquire):
    self.acquire = acquire

  async def post(self, data: dict):
    """Generate an invoice via the billing microservice."""
    # Calls POST http://billing-service:8001/api/invoice
    result = await self.billing.invoice.post(order_id=data["order_id"])
    return {"invoice": result}

class Microservice:
  host_url: str = ""
  api_key: str = ""
  timeout: float = 30.0
  prefix: str = "/api"
  service_name: str = ""

  def __get__(self, instance, owner):
    if instance is None:
      return self
    caller = getattr(instance, "_service_name", "unknown")
    if self.service_name:
      # Single-service shortcut: skip the repo proxy
      return MicroserviceServiceProxy(
        service_name=self.service_name, host_url=self.host_url, api_key=self.api_key,
        timeout=self.timeout, caller=caller, prefix=self.prefix,
      )
    return MicroserviceProxy(
      host_url=self.host_url, api_key=self.api_key,
      timeout=self.timeout, caller=caller, prefix=self.prefix,
    )

from singularity.rpc import rpc

class OrdersService:
  # Declare dependency at class level
  payments = rpc.remote("payments")

  async def post(self, data: dict):
    # Uses the class attribute — caller identity is auto-injected
    result = await self.payments.post_charge(
      user_id=data["user_id"],
      amount=data["total"],
    )
    return {"order": "confirmed", "payment": result}

from singularity.rpc import (
    rpc,
    RPCServiceNotFound,
    RPCAccessDenied,
    RPCMethodNotFound,
    RPCServiceUnavailable,
    RPCRemoteError,
)

class OrdersService:
  async def post(self, data: dict):
    try:
      result = await rpc("orders").payments.post_charge(
        user_id=data["user_id"],
        amount=data["total"],
      )
      return {"order": "confirmed", "payment": result}

    except RPCAccessDenied:
      return {"error": "Orders service is not allowed to access payments"}

    except RPCServiceNotFound:
      return {"error": "Payments service is not registered or unreachable"}

    except RPCMethodNotFound:
      return {"error": "post_charge is not exposed on payments"}

    except RPCServiceUnavailable as e:
      return {"error": f"Payments service is down: {e}"}

    except RPCRemoteError as e:
      # The call reached payments, which rejected it.
      return {"error": f"Payments returned {e.status_code}", "detail": e.payload}

# services/v1/payments/service.py
"""Payment processing service."""

from singularity.core.acquire import Acquire


class PaymentsService:
  """Handles charging and refunding payments."""

  rpc_exposed = True
  blacklist = ["public"]
  http_exposed = ["post=charge", "post=refund"]

  def __init__(self, acquire: Acquire):
    self.acquire = acquire
    self.db = acquire.db_session

  async def get(self):
    """List recent payments."""
    return {"payments": []}

  async def post_charge(self, user_id: str, amount: float):
    """Charge a user's payment method."""
    async with self.db() as session:
      # Record the payment
      payment = Payment(user_id=user_id, amount=amount, status="charged")
      session.add(payment)
      await session.commit()
      return {
        "charged": amount,
        "user": user_id,
        "transaction_id": payment.id,
      }

  async def post_refund(self, payment_id: str):
    """Refund a payment."""
    return {"refunded": True, "payment": payment_id}

# services/v1/orders/service.py
"""Order management service."""

from singularity.core.acquire import Acquire
from singularity.rpc import rpc


class OrdersService:
  """Creates orders and coordinates payment via RPC."""

  rpc_exposed = True

  def __init__(self, acquire: Acquire):
    self.acquire = acquire

  async def get(self):
    """List orders."""
    return {"orders": []}

  async def post(self, data: dict):
    """Create an order and charge the user."""
    payment = await rpc("orders").payments.post_charge(
      user_id=data["user_id"],
      amount=data["total"],
    )
    return {
      "order_id": "ord_new",
      "status": "confirmed",
      "payment": payment,
    }

singularity generate service v1/payments -d "Payment processing" --template crud --rpc --db
singularity generate service v1/orders -d "Order management" --template crud --rpc