import json from dataclasses import fields from typing import Any import urllib3 from errors import RequestError def dict_cls(d: dict, cls: Any) -> Any: """ Create a dataclass from a dictionary. """ field_names = set(f.name for f in fields(cls)) filtered_dict = {k: v for k, v in d.items() if k in field_names} return cls(**filtered_dict) def log_except(fn): """ Log unhandled exceptions to a logger instead of `stderr`. """ def wrapper(self, *args, **kwargs): try: return fn(self, *args, **kwargs) except Exception: self.logger.exception(f"Exception in '{fn.__name__}':") raise return wrapper def request(fn): """ Either return json data or raise a `RequestError` if the request was unsuccessful. """ def wrapper(*args, **kwargs): try: resp = fn(*args, **kwargs) except urllib3.exceptions.HTTPError as e: raise RequestError(None, f"Failed to connect: {e}") from None if resp.status < 200 or resp.status >= 300: raise RequestError( resp.status, f"Failed to get response from '{resp.geturl()}':\n{resp.data}", ) return {} if resp.status == 204 else json.loads(resp.data) return wrapper def except_deleted(fn): """ Ignore the `RequestError` on 404s, the content might have been removed. """ def wrapper(*args, **kwargs): try: return fn(*args, **kwargs) except RequestError as e: if e.status != 404: raise return wrapper def hash_str(string: str) -> int: """ Create the hash for a string """ hash = 5381 for ch in string: hash = ((hash << 5) + hash) + ord(ch) return hash & 0xFFFFFFFF