Reddit API in Python

# The only dependency. There is no redditapis package on PyPI.
pip install requests

# Keep the key out of your source tree.
export REDDITAPIS_KEY="rk_live_your_key_here"

import os

import requests

BASE_URL = "https://api.redditapis.com"

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

# One connection pool, one header, reused by every call below.
res = session.get(
    f"{BASE_URL}/api/reddit/posts",
    params={"subreddit": "webdev", "sort": "top", "t": "week", "limit": 25},
    timeout=30,
)
res.raise_for_status()

for post in res.json()["posts"]:
    print(post["upvotes"], post["title"], post["permalink"])

# Site-wide post search. Drop the subreddit param and it searches all of Reddit.
# min_score and is_self are applied to the returned page, and the response
# carries a meta block saying how many rows the filters removed.
res = session.get(
    f"{BASE_URL}/api/reddit/search",
    params={
        "q": "postgres connection pooling",
        "subreddit": "devops",
        "sort": "relevance",
        "t": "month",
        "limit": 50,
        "min_score": 20,
        "is_self": "true",
    },
    timeout=30,
)
res.raise_for_status()

posts = res.json()["posts"]
for p in posts:
    print(p["subreddit"], p["author"], p["title"], p["url"])

def iter_posts(subreddit, max_pages=5):
    """Walk a subreddit listing with the cursor the API hands back.

    Pass the `after` value straight back on the next request. Stop when it
    comes back null, which means there is no next page to ask for.

    That is NOT the same as having every post: Reddit often stops serving a
    busy listing early. The final response carries `listing_status`, and only
    "complete" means nothing is missing.
    """
    cursor = None
    for _ in range(max_pages):
        params = {"subreddit": subreddit, "sort": "new", "limit": 100}
        if cursor:
            params["after"] = cursor
        res = session.get(f"{BASE_URL}/api/reddit/posts", params=params, timeout=30)
        res.raise_for_status()
        body = res.json()
        yield from body["posts"]
        cursor = body.get("after")
        if not cursor:
            return


count = sum(1 for _ in iter_posts("rust"))
print(count, "posts across at most 5 pages, 5 reads, $0.010")

# The new-comment stream for one subreddit. This is the read that keyword
# monitoring is built on: poll it, match on body, keep the ids you have seen.
res = session.get(
    f"{BASE_URL}/api/reddit/sub/devops/comments",
    params={"limit": 100},
    timeout=30,
)
res.raise_for_status()

comments = res.json()["comments"]
for c in comments:
    if "kubernetes" in c["body"].lower():
        print(c["author"], c["subreddit"], c["upvotes"], c["url"])

# Deep comment search reads the comment TREES behind the matching threads, so
# it returns the actual comment bodies rather than the parent posts. One call
# fans out into a search plus several tree reads, which is why it is priced at
# $0.02 instead of $0.002.
res = session.get(
    f"{BASE_URL}/api/reddit/search/comments/deep",
    params={"q": "vector database", "limit": 10, "max_comments": 50},
    timeout=60,
)
res.raise_for_status()

body = res.json()
for c in body["comments"]:
    print(c["score"], c["author"], c["permalink"])
    print("  thread:", c["parent_post"]["title"])

print(body["meta"]["comments_matched"], "matched,", body["meta"]["comments_returned"], "returned")

import time


class RedditapisError(RuntimeError):
    def __init__(self, status, detail):
        self.status = status
        self.detail = detail
        super().__init__(f"{status} {detail}")


# 402 and 403 are terminal: no amount of retrying buys credits or fixes a bad
# key. 429 and 503 are transient and safe to retry.
TERMINAL = {400, 401, 402, 403, 404}
RETRYABLE = {429, 500, 502, 503, 504}


def call(path, params=None, attempts=4, timeout=30):
    delay = 1.0
    last = None
    for _ in range(attempts):
        res = session.get(f"{BASE_URL}{path}", params=params, timeout=timeout)
        if res.ok:
            return res.json()
        detail = res.json().get("error", res.text[:200])
        if res.status_code in TERMINAL:
            raise RedditapisError(res.status_code, detail)
        if res.status_code not in RETRYABLE:
            raise RedditapisError(res.status_code, detail)
        last = RedditapisError(res.status_code, detail)
        # The 429 carries Retry-After in seconds. Honour it before backing off.
        time.sleep(float(res.headers.get("Retry-After", delay)))
        delay *= 2
    raise last

# GET /account/me is free and never consumes credits. Poll it from a scheduled
# job so a long run never dies halfway through on an empty balance.
res = session.get(f"{BASE_URL}/account/me", timeout=30)
res.raise_for_status()

account = res.json()
print(account["credits_remaining"], "left")
print(account["credits_used"], "spent over", account["total_requests"], "requests")

# Every billed response also carries the balance in a header, so a worker can
# watch it without a second request.
posts = session.get(
    f"{BASE_URL}/api/reddit/posts",
    params={"subreddit": "python", "limit": 5},
    timeout=30,
)
print(posts.headers["X-Credits-Remaining"])