Code Examples

Working integration examples for common use cases. All examples use https://pharlo.io as the base URL and read credentials from environment variables.

See Authentication guide for how to get and manage your API key.

Publish a video

Create a publishing assignment and post it immediately.

⚠️

YouTube default privacy is private. Set "privacy": "public" explicitly if you want the video publicly visible. See Platform Capabilities for the full payload schema per platform.

curl -X POST https://pharlo.io/api/v1/assignments \
  -H "Authorization: Bearer $DELIVERY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "connectionId": "7f3e1a2b-4c5d-6e7f-8a9b-0c1d2e3f4a5b",
    "type": "video",
    "mediaUrl": "https://cdn.example.com/demo.mp4",
    "payload": {
      "title": "Product Demo",
      "description": "See what is new",
      "privacy": "public"
    }
  }'

Schedule a video

Set scheduledAt to a future time with an explicit timezone offset. The assignment is created immediately but published at the scheduled time.

⚠️

Always include a timezone offset such as +03:00. A bare ISO 8601 string without an offset (e.g. 2026-05-02T18:00:00) is rejected with a 400 validation error. See Assignments reference for details.

curl -X POST https://pharlo.io/api/v1/assignments \
  -H "Authorization: Bearer $DELIVERY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "connectionId": "7f3e1a2b-4c5d-6e7f-8a9b-0c1d2e3f4a5b",
    "type": "video",
    "mediaUrl": "https://cdn.example.com/demo.mp4",
    "scheduledAt": "2026-05-02T18:00:00+03:00",
    "payload": {
      "title": "Product Demo",
      "privacy": "public"
    }
  }'

Poll for assignment completion

Assignments are processed asynchronously. Poll GET /assignments/{id} until the status reaches a terminal state.

Terminal statuses: published | failed | cancelled

For production systems, prefer webhooks over polling — they push status changes to your server in real time and don't require open connections. Polling is useful for scripts, CLIs, and simple integrations.

import os
import time
import requests
 
TERMINAL = {"published", "failed", "cancelled"}
 
def wait_for_completion(assignment_id: str, timeout_s: int = 300) -> dict:
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        resp = requests.get(
            f"https://pharlo.io/api/v1/assignments/{assignment_id}",
            headers={"Authorization": f"Bearer {os.environ['DELIVERY_API_KEY']}"},
        )
        resp.raise_for_status()
        data = resp.json()
        if data["status"] in TERMINAL:
            return data
        time.sleep(15)
    raise TimeoutError(f"Assignment {assignment_id} did not complete within {timeout_s}s")
 
result = wait_for_completion("a1b2c3d4-e5f6-7890-abcd-ef1234567890")
if result["status"] == "published":
    print("Live at:", result.get("platformUrl"))
else:
    print("Failed:", result.get("errors"))

Retry a failed assignment

Find failed assignments and retry them in one pass.

import os
import requests
 
BASE = "https://pharlo.io"
HEADERS = {"Authorization": f"Bearer {os.environ['DELIVERY_API_KEY']}"}
 
# List failed assignments
resp = requests.get(f"{BASE}/api/v1/assignments", headers=HEADERS, params={"status": "failed"})
failed = resp.json().get("items", [])
print(f"Found {len(failed)} failed assignment(s)")
 
# Retry each one
for item in failed:
    r = requests.post(f"{BASE}/api/v1/assignments/{item['id']}/retry", headers=HEADERS)
    r.raise_for_status()
    print(f"Retried {item['id']} — new status: {r.json()['status']}")

Webhook receiver — HMAC verification

The API sends a X-Webhook-Signature header with each delivery — a SHA-256 HMAC of the raw request body using your webhook secret. Always verify it before processing the payload.

⚠️

Compute the HMAC over the raw bytes of the request body — before any JSON parsing. Parsing first and re-serializing can change whitespace and produce a different hash. Use timingSafeEqual / hmac.compare_digest to prevent timing attacks. See Webhooks guide for the full delivery format.

const express = require("express");
const crypto = require("crypto");
 
const app = express();
const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;
 
app.post(
  "/webhook",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const signature = req.headers["x-webhook-signature"];
    const expected = crypto
      .createHmac("sha256", WEBHOOK_SECRET)
      .update(req.body)
      .digest("hex");
 
    if (!crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected))) {
      return res.status(401).send("Invalid signature");
    }
 
    const event = JSON.parse(req.body);
    console.log(`Assignment ${event.assignmentId} → ${event.status}`);
 
    if (event.status === "published") {
      console.log("Live at:", event.platformUrl);
    }
    if (event.status === "failed") {
      console.error("Errors:", event.errors);
    }
 
    res.sendStatus(200);
  }
);
 
app.listen(3000);

OAuth channel connect flow

Connect a new social media channel by creating an OAuth ticket, redirecting the user, then verifying the result. See Channels guide and OAuth reference for full details.

import os
import time
import requests
 
BASE    = "https://pharlo.io"
HEADERS = {"Authorization": f"Bearer {os.environ['DELIVERY_API_KEY']}"}
 
# Step 1: create a ticket and get the OAuth URL
resp = requests.post(
    f"{BASE}/api/v1/oauth/tickets",
    headers=HEADERS,
    json={"platform": "youtube", "redirectUrl": "https://your-app.com/channels/connected"},
)
resp.raise_for_status()
ticket = resp.json()
print("Open this URL to connect the channel:")
print(ticket["authUrl"])
ticket_id = ticket["ticketId"]
 
# Step 2: wait for the user to complete OAuth (poll every 5 s, max 60 s)
for _ in range(12):
    time.sleep(5)
    status_resp = requests.get(f"{BASE}/api/v1/oauth/tickets/{ticket_id}", headers=HEADERS)
    status = status_resp.json().get("status")
    if status == "success":
        print("Channel connected!")
        break
    if status in ("failed", "expired"):
        print(f"OAuth {status} — start over")
        break
    print(f"Waiting... ({status})")
 
# Step 3: confirm the new channel appears
connections = requests.get(f"{BASE}/api/v1/connections", headers=HEADERS).json()
print("Connected channels:", [c["name"] for c in connections.get("items", [])])