Post to social media from Python

export POSTIFY_API_KEY="postify_live_XXXXXXXXXXXXXXXX"

import os
import time

import requests

BASE_URL = "https://app.usepostify.com"
MAX_ATTEMPTS = 4
MAX_WAIT_SECONDS = 60

session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['POSTIFY_API_KEY']}"


class PostifyError(Exception):
    """A non-2xx response, read from its problem+json body."""

    def __init__(self, status, problem, request_id=None):
        self.status = status
        self.code = problem.get("code")  # stable and machine-readable
        self.detail = problem.get("detail") or problem.get("title")
        self.errors = problem.get("errors", [])  # validation_failed only
        self.request_id = problem.get("request_id") or request_id
        super().__init__(
            f"{status} {self.code}: {self.detail} (request_id={self.request_id})"
        )


def read_problem(resp):
    if "problem+json" in resp.headers.get("Content-Type", ""):
        try:
            return resp.json()
        except ValueError:
            pass
    # Something in front of the API answered: there is no `code` to branch on.
    return {"title": f"HTTP {resp.status_code}", "detail": resp.text[:200]}


def seconds_to_wait(resp, error, default):
    """Honour Retry-After (delay-seconds form); give up rather than wait minutes."""
    raw = resp.headers.get("Retry-After", "").strip()
    seconds = int(raw) if raw.isdigit() else default
    if seconds > MAX_WAIT_SECONDS:
        raise error
    return seconds


def request(method, path, *, json=None, idempotency_key=None):
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    # A POST is only safe to repeat when it carries an Idempotency-Key.
    can_repeat = method == "GET" or idempotency_key is not None
    for attempt in range(MAX_ATTEMPTS):
        final = attempt == MAX_ATTEMPTS - 1
        try:
            resp = session.request(
                method, f"{BASE_URL}{path}", json=json, headers=headers, timeout=30
            )
        except (requests.ConnectionError, requests.Timeout):
            if final or not can_repeat:
                raise
            time.sleep(2**attempt)  # the same key goes out again
            continue
        if resp.ok:
            return resp.json()

        error = PostifyError(
            resp.status_code, read_problem(resp), resp.headers.get("Request-Id")
        )
        if final:
            raise error
        if error.code == "rate_limited":
            # Per-key burst limit: clears within seconds, Retry-After says when.
            wait = seconds_to_wait(resp, error, default=2**attempt)
        elif error.code == "idempotency_in_progress":
            # The first attempt with this key is still running: wait, same key.
            wait = seconds_to_wait(resp, error, default=2)
        elif error.code == "internal_error" and can_repeat:
            wait = 2**attempt  # it may or may not have landed: the key decides
        else:
            # quota_exhausted, validation_failed, auth and scope errors:
            # retrying cannot change the answer.
            raise error
        time.sleep(wait)

def list_channels():
    return request("GET", "/v1/channels")["data"]


def live_channels():
    return [c for c in list_channels() if c["status"] == "live"]

{
  "id": "cmc1x2y3z0001abcd",
  "platform": "linkedin",
  "handle": "@acme-co",
  "status": "live"
}

import uuid
from datetime import datetime, timedelta, timezone


def schedule_post(channel_id, body, when, idempotency_key=None):
    payload = {
        "variants": [{"channel_id": channel_id, "body": body}],
        # ISO-8601 in UTC. Mutually exclusive with `draft` and `publish_now`.
        "scheduled_at": when.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
    }
    # One key per logical post: request() re-sends it on every retry. If your
    # own job queue can retry this call, create the key with the job and store
    # it, so a rerun reuses it instead of minting a new one.
    key = idempotency_key or str(uuid.uuid4())
    return request("POST", "/v1/posts", json=payload, idempotency_key=key)


if __name__ == "__main__":
    try:
        channels = live_channels()
        if not channels:
            raise SystemExit("No live channel. Connect one in the Postify app.")
        post = schedule_post(
            channels[0]["id"],
            "Hello from Python.",
            datetime.now(timezone.utc) + timedelta(hours=1),
        )
        print(post["id"], post["status"], post["scheduled_at"])
    except PostifyError as err:
        if err.code == "validation_failed":
            for field in err.errors:
                print(f"{field['pointer']}: {field['message']}")
        elif err.code == "quota_exhausted":
            print("The monthly API allowance is used up; wait for the reset.")
        elif err.code in ("authentication_required", "invalid_api_key"):
            print("Check POSTIFY_API_KEY.")
        elif err.code == "insufficient_scope":
            print("Mint a key with the channels:read and posts:write scopes.")
        else:
            print(err)
        raise SystemExit(1)