matrix.py

import time
from json import JSONDecodeError
from typing import Any, Dict, Optional
from urllib.parse import urljoin

import requests

from aleksis.core.util.core_helpers import get_site_preferences


class MatrixException(Exception):
    pass


def build_url(path):
    return urljoin(
        urljoin(get_site_preferences()["matrix__homeserver"], "_matrix/client/v3/"), path
    )


def get_headers():
    return {
        "Authorization": "Bearer " + get_site_preferences()["matrix__access_token"],
    }


def do_matrix_request(method: str, url: str, body: Optional[dict] = None) -> Dict[str, Any]:
    """Do a HTTP request to the Matrix Client Server API."""
    while True:
        res = requests.request(method=method, url=build_url(url), headers=get_headers(), json=body)
        if res.status_code != requests.codes.ok:
            try:
                data = res.json()
            except JSONDecodeError:
                raise MatrixException(res.text)

            # If rate limit exceeded, wait and retry
            if data.get("errcode", "") == "M_LIMIT_EXCEEDED":
                time.sleep(data["retry_after_ms"] / 1000)
            else:
                raise MatrixException(data)
        else:
            break

    return res.json()