Singularity

Convention-based FastAPI microservices framework. Structure your backend like NestJS/Moleculer, but in Python.

Quickstart

pip install singularity-fm
singularity new my-project
cd my-project
singularity generate service users
singularity dev

Open http://localhost:8000/docs — your service is live.

---

What You Get

• Convention-based services — drop a service.py in a directory, it becomes an API
• Hook system — composable before/after/error middleware per method
• Method-level caching — declarative cache with TTL and invalidation
• Action visibility — mark methods as public, protected (RPC only), or private
• Exception filters — map exception types to HTTP status codes
• Inter-service RPC — call services across deployments via Redis discovery
• Background tasks — Celery-based with @task decorator and notification channels
• CLI tooling — scaffold projects, generate services, run dev server
• Testing harness — ServiceTestClient with RPC and microservice mocks

---

Installation

# Core only
pip install singularity-fm

# With database support (SQLAlchemy + asyncpg)
pip install singularity-fm[db]

# With background tasks (Celery + Redis)
pip install singularity-fm[tasks]

# With JWT authentication
pip install singularity-fm[auth]

# Everything
pip install singularity-fm[all]

---

Project Structure

After singularity new my-project:

my-project/
├── app.py              # 4-line entry point
├── settings.py         # Your settings (extends SingularitySettings)
├── .env                # Environment variables
├── services/           # Your services (auto-discovered)
├── tasks/executable/   # Your background tasks
├── models/             # SQLAlchemy models
├── middlewares/         # Custom middleware (auto-discovered)
└── tests/

app.py

from singularity import Singularity
from settings import Settings

app = Singularity(settings=Settings())

settings.py

from singularity import SingularitySettings

class Settings(SingularitySettings):
    stripe_api_key: str = ""

Settings are loaded from environment variables and .env automatically.

---

Services

Create services/users/service.py — that's it. The Manager discovers it and registers routes at /api/users.

from singularity.core.acquire import Acquire

class UsersService:
    """User management."""

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

    async def get(self):
        return {"users": []}

    async def post(self, data: dict):
        return {"created": data}

    async def put(self, data: dict):
        return {"updated": data}

    async def delete(self, user_id: str):
        return {"deleted": user_id}

Generate with CLI

singularity generate service users                        # CRUD service
singularity generate service users --template minimal     # GET only
singularity generate service payments --rpc --db          # With RPC + database
singularity generate service stripe --template webhook    # Webhook receiver

Custom Routes

class TodosService:
    http_exposed = ["get=stats", "post=toggle"]

    async def get(self): ...
    async def get_stats(self): ...     # GET /api/todos/stats
    async def post_toggle(self): ...   # POST /api/todos/toggle

---

Hooks

Composable before/after/error middleware. Declare as a class attribute:

from singularity.core.hooks import HookContext

async def authenticate(ctx: HookContext):
    if not ctx.kwargs.get("token"):
        raise HTTPException(status_code=401)

async def log_response(ctx: HookContext):
    ctx.acquire.logger.info(f"Response: {ctx.result}")

class UsersService:
    hooks = {
        "before": {"all": [authenticate]},
        "after": {"all": [log_response]},
        "error": {"all": [handle_error]},
    }

Before hooks can short-circuit by setting ctx.result. Error hooks can swallow exceptions by setting ctx.result.

---

Method-Level Caching

Declarative caching with TTL and automatic invalidation:

class UsersService:
    cache = {
        "get": {"ttl": 60, "key": "users:list"},
        "post": {"invalidates": ["get"]},
    }

• ttl — cache duration in seconds
• key — explicit cache key (auto-generated if omitted)
• invalidates — list of methods whose cache keys to delete

Uses the async cache backend (in-memory or Redis, auto-detected from settings).

---

Action Visibility

Control which methods are exposed as HTTP routes vs RPC vs internal:

class PaymentsService:
    rpc_exposed = True
    visibility = {
        "charge": "protected",       # RPC only, no HTTP route
        "_reconcile": "private",     # Internal only, not exposed anywhere
    }

Level	HTTP	RPC	Internal
public (default)	Yes	Yes	Yes
protected	No	Yes	Yes
private	No	No	Yes

---

Exception Filters

Map exception types to HTTP responses. Checked before error hooks:

class UsersService:
    exception_filters = {
        ValueError: {"status": 400, "message": "Invalid input"},
        KeyError: {"status": 404, "message": "Not found"},
    }

Matched via isinstance, so subclasses are caught too. Unmatched exceptions fall through to error hooks.

---

Inter-Service RPC

Expose a service for RPC calls:

class PaymentsService:
    rpc_exposed = True
    blacklist = ["untrusted"]  # Block specific callers

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

Call from another service:

from singularity.rpc.proxy import RemoteServiceDescriptor

class OrdersService:
    payments = RemoteServiceDescriptor("payments")

    async def post(self, data: dict):
        result = await self.payments.charge(user_id="abc", amount=9.99)
        return {"order": data, "payment": result}

Local calls are in-process (zero overhead). Remote services are discovered via Redis with automatic heartbeat.

---

Background Tasks

from singularity.tasks import task

@task(name="send_email", notify=["log"])
def send_email(to: str, subject: str) -> dict:
    return {"sent_to": to}

@send_email.on_complete
def on_done(status, result, meta):
    print(f"Email {status}: {result}")

Submit from a service:

self.acquire.tasks.submit("send_email", to="user@example.com", subject="Welcome")

Generate with CLI:

singularity generate task send_welcome

---

Service Events

Auto-broadcast WebSocket events on mutations:

class TodosService:
    service_events = ["created", "updated", "removed"]

    async def post(self, data: dict):
        return {"id": 1, "title": data["title"]}
        # Automatically broadcasts: {"event": "todos.created", "data": {...}}

---

Webhooks

from singularity.core.webhook import BaseWebhook

class StripeWebhookService(BaseWebhook):
    events = {
        "invoice.paid": "handle_invoice_paid",
    }

    def verify(self, headers: dict, raw_body: bytes) -> bool:
        return True  # Implement signature verification

    def get_event_type(self, headers: dict, payload: dict) -> str:
        return payload.get("type", "unknown")

    async def handle_invoice_paid(self, payload: dict, headers: dict):
        return {"status": "processed"}

---

Testing

from singularity.testing import ServiceTestClient

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

    async def get(self):
        return {"users": []}

# Test
async def test_get_users():
    client = ServiceTestClient(UsersService)
    result = await client.get()
    assert result == {"users": []}

Mock RPC dependencies:

client = ServiceTestClient(OrdersService, rpc={
    "payments": {"charge": {"charged": 9.99}},
})
result = await client.post({"item": "book"})

---

CLI Reference

singularity new <project>                    # Scaffold a new project
singularity dev [--port 8000]                # Start dev server with hot reload
singularity generate service <name>          # Generate service boilerplate
singularity generate task <name>             # Generate task boilerplate

---

Architecture

graph TD
    Client([Client]) -->|HTTP| FastAPI
    FastAPI -->|Route| Manager
    Manager -->|Inject Acquire| Services
    Services <-->|RPC| Registry
    Registry -->|Heartbeat| Redis
    Services -->|Submit| TaskRunner
    TaskRunner -->|Queue| Celery
    Celery -->|Consume| Redis
    Services -->|Query| PostgreSQL

Loading