ClickUp

pip install "flyteplugins-clickup[app]"

import flyte
from flyte.extras.webhooks import WebhookAppEnvironment, WebhookEvent, run_once
from flyteplugins.clickup import ClickUpProvider, events

# CLICKUP_WEBHOOK_SECRET is mounted automatically.
#
# `scopes` lists ClickUp list IDs. The provider reads the list ID from both
# list events and task events.
app_env = WebhookAppEnvironment(
    name="clickup-webhooks",
    providers=[ClickUpProvider()],
    scopes=["9000"],
    image=flyte.Image.from_debian_base().with_pip_packages("flyteplugins-clickup[app]"),
    resources=flyte.Resources(cpu=1, memory="512Mi"),
)

@app_env.on_event(events.Task.STATUS_UPDATED)
async def on_status_updated(event: WebhookEvent) -> dict:
    """Launch a run once per status change.

    `qualified_type` is `taskStatusUpdated`, and `action` is None.
    """
    import flyte.remote as remote

    task = remote.Task.get(name="clickup-ops.close_ticket", auto_version="latest")
    result = await run_once.aio(
        task,
        key=event.dedupe_key(),
        task_id=event.resource_id,
    )
    return {"run": result.run.name, "created": result.created}

env = flyte.TaskEnvironment(
    name="clickup-ops",
    image=flyte.Image.from_debian_base().with_pip_packages("httpx"),
    secrets=[flyte.Secret(key="clickup-api-token", as_env_var="CLICKUP_API_TOKEN")],
    resources=flyte.Resources(cpu=1, memory="512Mi"),
)

CLICKUP_API = "https://api.clickup.com/api/v2"

@env.task
async def close_ticket(task_id: str) -> str:
    """Close a ticket unless it's already complete.

    `run_once` prevents duplicate runs, not duplicate writes within a run.
    ClickUp accepts a redundant status write and logs it, so the task checks
    the current status first.
    """
    import os

    import httpx

    headers = {"Authorization": os.environ["CLICKUP_API_TOKEN"]}
    async with httpx.AsyncClient(base_url=CLICKUP_API, headers=headers, timeout=15.0) as client:
        current = (await client.get(f"/task/{task_id}")).raise_for_status().json()
        if current["status"]["status"] == "complete":
            return "already complete"
        response = await client.put(f"/task/{task_id}", json={"status": "complete"})
        response.raise_for_status()
    return "closed"

# A separate environment with no secrets, so the replay runs before any
# secret is created.
replay_env = flyte.TaskEnvironment(
    name="clickup-replay",
    image=flyte.Image.from_debian_base().with_pip_packages("flyteplugins-clickup"),
    resources=flyte.Resources(cpu=1, memory="512Mi"),
)

@replay_env.task
async def replay_sample_delivery() -> dict[str, str]:
    """Verify and parse the sample delivery bundled with the plugin."""
    import flyteplugins.clickup as plugin

    secret = "a-test-signing-secret"
    sign, body = plugin.SAMPLE_DELIVERY
    headers = sign(body, secret)

    assert plugin.verify(body, headers, secret), "a correctly signed delivery must verify"
    assert not plugin.verify(body, headers, "wrong-secret"), "a bad signature must not"

    # Check the header name too. The sample's headers come from the plugin, so
    # the round trip above passes whatever the header is called. ClickUp sends
    # `X-Signature`, not `X-Clickup-Signature`.
    assert list(headers) == ["X-Signature"], f"unexpected signature header: {list(headers)}"
    assert not plugin.verify(body, {"X-Clickup-Signature": headers["X-Signature"]}, secret), (
        "X-Clickup-Signature must not verify"
    )

    event = plugin.parse(headers, body)
    return {
        # `taskCreated`: ClickUp sends a single event name with no action.
        "qualified_type": event.qualified_type,
        "action_is_none": str(event.action is None),
        "scope": event.scope or "",
        "title": event.title or "",
        "dedupe_key": event.dedupe_key(),
        # The header that carries the signature.
        "signature_header": next(iter(headers)),
    }

flyte run --local clickup_tasks.py replay_sample_delivery