diff --git a/rplugin/python3/mrpy/clickup.py b/rplugin/python3/mrpy/clickup.py index cae45bd..a1794d8 100644 --- a/rplugin/python3/mrpy/clickup.py +++ b/rplugin/python3/mrpy/clickup.py @@ -4,42 +4,9 @@ from json import loads from os import environ from typing import Any, Callable, Self, TypedDict -from urllib.request import Request, urlopen - from .hints import JSONDataMap - - -def get( - url: str, - query_params: dict[str, str] | None = None, - headers: dict[str, str] | None = None, -) -> str: - with urlopen( - Request( - url + ("?" + "&".join(f"{k}={v}" for k, v in query_params.items())) - if query_params - else "", - headers=headers or {}, - method="GET", - ), - ) as resp: - return resp.read().decode("utf-8") - - -def put( - url: str, - data: dict[str, str] | None = None, - headers: dict[str, str] | None = None, -) -> str: - with urlopen( - Request( - url, - str(data).encode(), - headers=headers or {}, - method="PUT", - ), - ) as resp: - return resp.read().decode("utf-8") +from .requests import get, put +from .env import EnvVar class TaskRespDict(TypedDict): @@ -134,33 +101,22 @@ class ClickupTask: return ret -def get_env_var(var_name: str, err_msg: str = "") -> Callable[[], str]: - def wrapper(var: str = var_name) -> str: - try: - return environ[var] - except KeyError as e: - e.add_note(err_msg) - raise - - return wrapper - - @dataclass class ClickupSession: auth_key: str = field( - default_factory=get_env_var( + default_factory=EnvVar( "CLICKUP_AUTH", "clickup auth token is required to be set", ) ) workspace_id: str = field( - default_factory=get_env_var( + default_factory=EnvVar( "CLICKUP_WORKSPACE_ID", "clickup workspace id is required to be set", ) ) user_id: str = field( - default_factory=get_env_var( + default_factory=EnvVar( "CLICKUP_USER_ID", "clickup user id is required to be set", ) diff --git a/rplugin/python3/mrpy/env.py b/rplugin/python3/mrpy/env.py new file mode 100644 index 0000000..57ebd9a --- /dev/null +++ b/rplugin/python3/mrpy/env.py @@ -0,0 +1,24 @@ +from dataclasses import dataclass +from os import environ + + +@dataclass +class EnvVar: + """ + Environment Variable fetcher for use in dataclass ``field(default_factory=...)`` + + >>> @dataclass + >>> class SomeDataclass: + ... field_name: str = field(default_factory=EnvVar("SOME_VAR_NAME", "err msg")) + + """ + + var_name: str + err_msg: str = "" + + def __call__(self) -> str: + try: + return environ[self.var_name] + except KeyError as e: + e.add_note(self.err_msg) + raise diff --git a/rplugin/python3/mrpy/requests.py b/rplugin/python3/mrpy/requests.py new file mode 100644 index 0000000..fda6efc --- /dev/null +++ b/rplugin/python3/mrpy/requests.py @@ -0,0 +1,34 @@ +from urllib.request import Request, urlopen + + +def get( + url: str, + query_params: dict[str, str] | None = None, + headers: dict[str, str] | None = None, +) -> str: + with urlopen( + Request( + url + ("?" + "&".join(f"{k}={v}" for k, v in query_params.items())) + if query_params + else "", + headers=headers or {}, + method="GET", + ), + ) as resp: + return resp.read().decode("utf-8") + + +def put( + url: str, + data: dict[str, str] | None = None, + headers: dict[str, str] | None = None, +) -> str: + with urlopen( + Request( + url, + str(data).encode(), + headers=headers or {}, + method="PUT", + ), + ) as resp: + return resp.read().decode("utf-8")