Predicate-Gated Jobs

predicate_gated/
  app.py        # Queue + tasks
  predicates.py # Custom predicate classes
  main.py       # Submit jobs
  worker.py     # Run the worker

"""Custom predicates registered with the queue."""

from __future__ import annotations

from typing import Any

import httpx

from flexiq.predicates import Defer, Predicate, PredicateContext


class TenantQuotaUnder(Predicate):
    """Defer until the tenant has quota remaining."""

    OP = "tenant_quota_under"

    def __init__(self, *, url: str = "", limit: int = 0) -> None:
        self.url = url
        self.limit = limit

    async def evaluate(self, ctx: PredicateContext) -> bool | Defer:
        tenant = ctx.kwargs.get("tenant")
        if tenant is None:
            return False
        async with httpx.AsyncClient(timeout=2.0) as client:
            r = await client.get(f"{self.url}/quota/{tenant}")
        if r.json()["used"] < self.limit:
            return True
        return Defer(seconds=300.0)

    def to_dict(self) -> dict[str, Any]:
        return {"op": self.OP, "url": self.url, "limit": self.limit}

    @classmethod
    def _from_kwargs(cls, kwargs: dict[str, Any]) -> "Predicate":
        return cls(**kwargs)


class TenantIn(Predicate):
    """Hard allowlist — rejects at enqueue time when on_false='cancel'."""

    OP = "tenant_in"

    def __init__(self, *, tenants: list[str]) -> None:
        self.tenants = list(tenants)

    def evaluate(self, ctx: PredicateContext) -> bool:
        return ctx.kwargs.get("tenant") in self.tenants

    def to_dict(self) -> dict[str, Any]:
        return {"op": self.OP, "tenants": self.tenants}

    @classmethod
    def _from_kwargs(cls, kwargs: dict[str, Any]) -> "Predicate":
        return cls(**kwargs)

"""Queue + tasks. All gates are composable, serializable predicates."""

from flexiq import Queue
from flexiq.predicates import (
    feature_flag,
    is_business_hours,
    queue_paused,
)

from .predicates import TenantIn, TenantQuotaUnder

queue = Queue(db_path=".flexiq/predicate_gated.db", workers=4)

# Register the custom predicate ops so they show up in JSON, the
# string DSL, and queue.list_predicates().
queue.register_predicate("tenant_quota_under")(TenantQuotaUnder)
queue.register_predicate("tenant_in")(TenantIn)


# ── Reports: business hours + tenant allowlist ──────────────────────
@queue.task(
    predicate=is_business_hours(tz="US/Pacific")
              & TenantIn(tenants=["acme", "globex"]),
    on_false="defer",
    default_defer_seconds=900.0,
)
def send_daily_report(tenant: str) -> str:
    return f"sent {tenant}"


# ── Billing: gated behind FF_NEW_BILLING ────────────────────────────
@queue.task(
    queue="critical",
    predicate=feature_flag("new_billing") & ~queue_paused(),
    on_false="cancel",
)
def charge_card(tenant: str, cents: int) -> str:
    return f"charged {cents}c to {tenant}"


# ── Indexing: async quota-bound ─────────────────────────────────────
@queue.task(
    predicate=TenantQuotaUnder(url="http://quota.internal", limit=10_000),
)
async def index_document(tenant: str, doc_id: str) -> str:
    return f"indexed {doc_id}"

"""Submit jobs and watch each layer of the predicate system kick in."""

from flexiq.exceptions import PredicateRejectedError

from .app import charge_card, index_document, queue, send_daily_report


def main() -> None:
    # 1. Allowed (acme is in the allowlist).
    send_daily_report.delay(tenant="acme")

    # 2. Deferred: initech is not in the allowlist. send_daily_report uses
    #    on_false="defer", so the enqueue succeeds and the job is held back to
    #    be re-evaluated later — no exception is raised.
    send_daily_report.delay(tenant="initech")

    # 3. Rejected at enqueue: charge_card uses on_false="cancel", and the
    #    new_billing flag is unset, so the predicate denies and enqueue raises.
    try:
        charge_card.delay(tenant="acme", cents=4200)
    except PredicateRejectedError:
        print("billing flag off — charge rejected")

    # 4. Bypass the billing flag by setting it in env; now it enqueues.
    import os
    os.environ["FF_NEW_BILLING"] = "true"
    charge_card.delay(tenant="acme", cents=4200)

    # 5. Async predicate + async task — both awaited transparently.
    index_document.delay(tenant="acme", doc_id="doc-1")

    # ── Inspection: dashboard reads the same shape ─────────────────
    print(queue.list_predicates())
    # {
    #   "app.send_daily_report": {"op": "and", "args": [...]},
    #   "app.charge_card":       {"op": "and", "args": [...]},
    #   "app.index_document":    {"op": "tenant_quota_under",
    #                             "url": "http://quota.internal",
    #                             "limit": 10000},
    # }


if __name__ == "__main__":
    main()

from .app import queue

if __name__ == "__main__":
    queue.run_worker()

from flexiq.predicates import Predicate, format_predicate, parse

p = queue._task_predicates["app.send_daily_report"]

# 1. JSON for storage / API / dashboard
blob = p.to_dict()

# 2. Stable, parseable string for ops-side editing
text = format_predicate(p)
# 'is_business_hours(tz="US/Pacific") & tenant_in(tenants=["acme", "globex"])'

# 3. Rebuild from either form
assert Predicate.from_dict(blob).to_dict() == blob
assert parse(text).to_dict() == blob

from flexiq.events import EventType

def log_event(event_type: EventType, payload: dict) -> None:
    match event_type:
        case EventType.PREDICATE_DEFERRED:
            print(f"[defer] {payload['task_name']} +{payload['defer_seconds']}s "
                  f"phase={payload['phase']}")
        case EventType.PREDICATE_CANCELLED:
            print(f"[cancel] {payload['task_name']}: "
                  f"{payload.get('reason', '')}")
        case EventType.PREDICATE_REJECTED:
            print(f"[reject] {payload['task_name']}: "
                  f"{payload.get('reason', '')}")

for event in (
    EventType.PREDICATE_DEFERRED,
    EventType.PREDICATE_CANCELLED,
    EventType.PREDICATE_REJECTED,
):
    queue._event_bus.on(event, log_event)