103 lines
2.2 KiB
Python
103 lines
2.2 KiB
Python
import asyncio
|
|
import json
|
|
from dataclasses import fields
|
|
from typing import Any
|
|
|
|
import urllib3
|
|
|
|
from errors import RequestError
|
|
|
|
|
|
def dict_cls(dict_var: 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 dict_var.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 wrap_async(fn):
|
|
"""
|
|
Call an asynchronous function from a synchronous one.
|
|
"""
|
|
|
|
def wrapper(self, *args, **kwargs):
|
|
if not self.loop:
|
|
raise RuntimeError("loop is None.")
|
|
|
|
return asyncio.run_coroutine_threadsafe(
|
|
fn(self, *args, **kwargs), loop=self.loop
|
|
).result()
|
|
|
|
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 message might have been
|
|
deleted by someone else already.
|
|
"""
|
|
|
|
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 (poorly).
|
|
"""
|
|
|
|
hashed = 0
|
|
results = map(ord, string)
|
|
|
|
for result in results:
|
|
hashed += result
|
|
|
|
return hashed
|