discourse_api.py

import time

import requests


class ApiClient:
	def __init__(self, url: str):
		self.url = url

	def get(self, url: str, params=None, retries=3):
		if "://" not in url:
			url = self.url + url.lstrip("/")

		will_retry = retries > 0 or retries == -1
		do_retry = lambda: self.get(url, params=params, retries=retries - 1)

		r = requests.get(url, params=params)
		if r.status_code == 422:
			# 422 is a "you're doing this too much"
			# But getting a topic by ID with a print=true can result in
			# a 422 if the topic is large enough.
			# This is a hack to avoid that.
			if will_retry:
				if params and params.get("print") == "true":
					print(f"HTTP 422 ({url}, {params}): Retrying without print=true")
					params.pop("print")
				elif will_retry:
					print(f"HTTP 422 ({url}): Retrying in 3 seconds…")
					time.sleep(3)
				return do_retry()
			else:
				raise RuntimeError(f"Got HTTP 422 for {url}. Not retrying.")

		try:
			return r.json()
		except Exception as e:
			print(f"Exception at {url} while decoding JSON: {e} | will retry: {will_retry}")
			if will_retry:
				return do_retry()
			else:
				raise RuntimeError(f"Got JSON decoding error for {url}. Not retrying.")

	##
	# API methods

	def get_site(self):
		return self.get("/site.json")

	def get_categories(self):
		return self.get("/categories.json")

	def get_categories_and_top(self):
		return self.get("/categories_and_top")

	def get_group(self, name: str):
		return self.get(f"/groups/{name}.json")

	def get_group_posts(self, name: str):
		return self.get(f"/groups/{name}/posts.json")

	def get_group_topics(self, name: str):
		return self.get(f"/topics/groups/{name}.json")

	def get_post(self, topic_id: int, post_number: int):
		return self.get(f"/posts/by_number/{topic_id}/{post_number}.json")

	def get_topic(self, id: int, print=False):
		# print=True returns more details (and up to 1000 posts)
		params = {"print": "true"} if print else {}
		return self.get(f"/t/{id}.json", params=params)

	def get_user(self, username: str):
		return self.get(f"/u/{username}.json")