Background Tasks

from singularity.tasks import task

@task(name="send_email", notify=["log"])
def send_email(to: str, subject: str, body: str) -> dict:
    """Send a transactional email."""
    # Your email sending logic here
    return {"status": "sent", "to": to}

@task(name="process_order", description="Process an e-commerce order")
def process_order(order_id: str, items: list, priority: int = 0) -> dict:
    # TaskRunner validates that order_id and items are provided at submit time
    return {"order_id": order_id, "total_items": len(items)}

from services.__base.service import Service

class ReportsService(Service):
    def __init__(self, acquire):
        super().__init__(acquire)
        self.tasks = acquire.tasks

    async def post(self, report_type: str):
        task_ref = self.tasks.submit(
            "generate_report",
            report_type=report_type,
            date_range={"start": "2024-01-01", "end": "2024-12-31"},
            _meta={"requester_email": "user@example.com"},
        )
        return {"task_id": task_ref.id, "status": "queued"}

@task(name="generate_invoice", notify=["callback", "log"])
def generate_invoice(order_id: str) -> dict:
    pdf_path = create_invoice_pdf(order_id)
    return {"invoice_path": pdf_path, "order_id": order_id}

@generate_invoice.on_complete
def after_invoice(status: str, result: dict, meta: dict):
    """Chain: email the invoice, then notify via Slack."""
    if status == "success":
        # Submit another task -- creates a chain
        send_email.submit(
            to=meta.get("customer_email"),
            subject=f"Invoice for order {result['order_id']}",
            body=f"Your invoice is ready: {result['invoice_path']}",
        )

from singularity.tasks.notifications import NotificationChannel

class PagerDutyChannel(NotificationChannel):
    """Alert PagerDuty on task failure."""

    def send(
        self,
        task_name: str,
        status: str,
        result=None,
        error=None,
        duration_ms: float = 0,
        meta=None,
        task_config=None,
    ) -> None:
        if status != "failure":
            return
        # Send PagerDuty alert
        requests.post(
            "https://events.pagerduty.com/v2/enqueue",
            json={
                "routing_key": task_config.get("routing_key"),
                "event_action": "trigger",
                "payload": {
                    "summary": f"Task {task_name} failed: {error}",
                    "severity": "critical",
                    "source": "singularity-tasks",
                },
            },
        )

from singularity.tasks.runner import TaskRunner

runner = TaskRunner()
runner.register_notification_channel("pagerduty", PagerDutyChannel())

@task(
    name="critical_sync",
    notify=["log", "pagerduty"],
    notify_config={"pagerduty": {"routing_key": "R0123456789"}},
)
def critical_sync(source: str) -> dict:
    ...

@task(
    name="call_external_api",
    max_retries=5,
    retry_delay=120,    # 2 minutes between retries
    time_limit=300,     # kill after 5 minutes
    notify=["log", "webhook"],
    notify_config={"webhook": alert_config},
)
def call_external_api(endpoint: str, payload: dict) -> dict:
    response = requests.post(endpoint, json=payload, timeout=30)
    response.raise_for_status()
    return response.json()

# List all discovered tasks
tasks list

# Detailed info about a specific task (shows parameters, types, defaults)
tasks info send_email

# Create a new task with boilerplate
singularity generate task send_notification -d "Send push notifications to users"

# Trigger a task manually with keyword arguments
tasks run send_email -k '{"to": "user@example.com", "subject": "Hello", "body": "World"}'

# Trigger with a countdown delay (starts in 60 seconds)
tasks run process_order -k '{"order_id": "ord_123", "items": []}' -c 60

# Check the result of a submitted task
tasks result abc12345-task-id

# tasks/executable/process_payment.py

from singularity.tasks import task
from singularity.tasks.notifications.webhook_channel import WebhookConfig

slack_webhook = WebhookConfig(
    url="https://hooks.slack.com/services/T00/B00/xxxxx",
    headers={"Content-Type": "application/json"},
)

@task(
    name="process_payment",
    description="Charge a customer and update their subscription",
    notify=["log", "callback", "webhook"],
    notify_config={"webhook": slack_webhook},
    max_retries=3,
    retry_delay=60,
    time_limit=120,
)
def process_payment(user_id: str, amount: float, currency: str = "usd") -> dict:
    """Charge the customer via Stripe and activate their subscription."""
    charge = stripe.Charge.create(
        amount=int(amount * 100),
        currency=currency,
        customer=user_id,
    )
    return {"charge_id": charge.id, "amount": amount, "user_id": user_id}


@process_payment.on_complete
def on_payment_done(status: str, result: dict, meta: dict):
    if status == "success":
        # Chain: send a receipt email
        from tasks.executable.send_email import send_email
        send_email.submit(
            to=meta.get("customer_email"),
            subject="Payment received",
            body=f"We charged ${result['amount']} to your card.",
        )

# services/v1/billing/service.py

from services.__base.service import Service

class BillingService(Service):
    def __init__(self, acquire):
        super().__init__(acquire)
        self.tasks = acquire.tasks

    async def post(self, user_id: str, amount: float):
        ref = self.tasks.submit(
            "process_payment",
            user_id=user_id,
            amount=amount,
            _meta={
                "customer_email": "customer@example.com",
                "org_id": "org_abc",
            },
        )
        return {"task_id": ref.id, "status": "processing"}