Compare commits
3 Commits
9a46317e6c
...
e190653961
| Author | SHA1 | Date |
|---|---|---|
|
|
e190653961 | |
|
|
a4efe1fd08 | |
|
|
aa93e8c712 |
|
|
@ -7,6 +7,5 @@ requires-python = ">=3.12"
|
|||
dependencies = [
|
||||
"pydantic-settings~=2.12",
|
||||
"pynvim~=0.6.0",
|
||||
"ruamel.yaml~=0.18.17",
|
||||
"requests>=2.32.5",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -1,8 +1,43 @@
|
|||
from json import loads
|
||||
from typing import Any
|
||||
from pydantic import AliasChoices, AliasPath, BaseModel, Field, field_validator
|
||||
|
||||
from pydantic import AliasPath, BaseModel, Field, field_validator
|
||||
from pydantic_settings import BaseSettings
|
||||
from requests import get, put
|
||||
from ruamel.yaml import Node, Representer, ScalarNode
|
||||
|
||||
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")
|
||||
|
||||
|
||||
class ClickupTask(BaseModel):
|
||||
|
|
@ -77,18 +112,18 @@ class ClickupSession(BaseSettings):
|
|||
base_url: str = "https://api.clickup.com/api/v2"
|
||||
|
||||
def _get(self, endpoint: str, **query_params: Any) -> dict[str, Any]:
|
||||
with get(
|
||||
raw_data = get(
|
||||
self.base_url + endpoint,
|
||||
query_params,
|
||||
headers={
|
||||
"accept": "application/json",
|
||||
"Authorization": self.auth_key,
|
||||
},
|
||||
) as resp:
|
||||
return resp.json()
|
||||
)
|
||||
return loads(raw_data)
|
||||
|
||||
def _put(self, endpoint: str, **body_params: Any) -> dict[str, Any]:
|
||||
with put(
|
||||
raw_data = put(
|
||||
self.base_url + endpoint,
|
||||
body_params,
|
||||
headers={
|
||||
|
|
@ -96,8 +131,8 @@ class ClickupSession(BaseSettings):
|
|||
"content-type": "application/json",
|
||||
"Authorization": self.auth_key,
|
||||
},
|
||||
) as resp:
|
||||
return resp.json()
|
||||
)
|
||||
return loads(raw_data)
|
||||
|
||||
def get_tasks(self) -> list[ClickupTask]:
|
||||
data = self._get(
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ from pynvim import Nvim, command, plugin
|
|||
from pynvim.api import Buffer
|
||||
|
||||
from .clickup import ClickupSession
|
||||
from .types import JSONData
|
||||
from .hints import JSONData
|
||||
from .yaml import load, dump
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,12 +1,11 @@
|
|||
from io import StringIO
|
||||
from ruamel.yaml import YAML
|
||||
from json import loads
|
||||
from re import sub
|
||||
from textwrap import dedent
|
||||
|
||||
from .types import JSONData, JSONDataScalar
|
||||
|
||||
yaml = YAML(typ="safe")
|
||||
from . import hints
|
||||
|
||||
|
||||
def dump_scalar(entry: JSONDataScalar) -> str:
|
||||
def dump_scalar(entry: hints.JSONDataScalar) -> str:
|
||||
match entry:
|
||||
case None:
|
||||
return "null\n"
|
||||
|
|
@ -18,7 +17,7 @@ def dump_scalar(entry: JSONDataScalar) -> str:
|
|||
return f"{entry:s}\n"
|
||||
|
||||
|
||||
def dump(obj: dict[str, JSONDataScalar | list[JSONDataScalar]]) -> str:
|
||||
def dump(obj: dict[str, hints.JSONDataScalar | list[hints.JSONDataScalar]]) -> str:
|
||||
ret = ""
|
||||
for key, value in obj.items():
|
||||
ret += key
|
||||
|
|
@ -35,6 +34,74 @@ def dump(obj: dict[str, JSONDataScalar | list[JSONDataScalar]]) -> str:
|
|||
return ret
|
||||
|
||||
|
||||
def load(content: str) -> JSONData:
|
||||
with StringIO(content) as sio:
|
||||
return yaml.load(sio) # pyright: ignore[reportAny, reportUnknownMemberType]
|
||||
def load(content: str) -> hints.JSONData:
|
||||
"""parse subset of yaml into ``JSONData``
|
||||
|
||||
>>> from pprint import pprint
|
||||
>>> yaml = '''
|
||||
... key1: str
|
||||
... key2: "str"
|
||||
... key3: 1
|
||||
... key4: 2.3
|
||||
... key5: 2e12
|
||||
... key6: null
|
||||
... key7: true
|
||||
... key8: false
|
||||
... key9:
|
||||
... - 1
|
||||
... - 23.2
|
||||
... key10:
|
||||
... - str
|
||||
... - "str"
|
||||
... '''
|
||||
>>> yaml_parsed = load(yaml)
|
||||
>>> compare = {
|
||||
... 'key1': 'str',
|
||||
... 'key2': 'str',
|
||||
... 'key3': 1,
|
||||
... 'key4': 2.3,
|
||||
... 'key5': 2e12,
|
||||
... 'key6': None,
|
||||
... 'key7': True,
|
||||
... 'key8': False,
|
||||
... 'key9': [1, 23.2],
|
||||
... 'key10': ['str', 'str'],
|
||||
... }
|
||||
|
||||
>>> {k: yaml_parsed.get(k, None) for k in compare if compare[k] != yaml_parsed[k]}
|
||||
{}
|
||||
"""
|
||||
|
||||
ret = {}
|
||||
key = None
|
||||
value = None
|
||||
for line in sub(r":[\s\n]*", ":\n ", dedent(content).strip()).splitlines():
|
||||
if line.endswith(":"):
|
||||
if key:
|
||||
ret[key] = value
|
||||
value = None
|
||||
|
||||
key = line.removesuffix(":")
|
||||
elif line.startswith(" -"):
|
||||
value = value or []
|
||||
try:
|
||||
parsed = loads(line.removeprefix(" -").strip())
|
||||
except:
|
||||
parsed = loads('"' + line.removeprefix(" -").strip() + '"')
|
||||
value.append(parsed)
|
||||
else:
|
||||
try:
|
||||
value = loads(line.strip())
|
||||
except:
|
||||
value = loads('"' + line.strip() + '"')
|
||||
if key:
|
||||
ret[key] = value
|
||||
value = None
|
||||
|
||||
return ret
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import doctest
|
||||
|
||||
doctest.testmod()
|
||||
|
|
|
|||
Loading…
Reference in New Issue