rename

eta/clickup.py

@@ -0,0 +1,68 @@
+from dataclasses import dataclass, field
+from json import loads
+from pydantic import BaseModel
+from requests import HTTPError, get
+
+from .common import EnvVar, JSONDataMap, JSONData, JSONDataList, JSONDataScalar
+
+
+class ClickupTask(BaseModel):
+    id: str
+    name: str
+
+
+@dataclass
+class ClickupSession:
+    auth_key: str = field(
+        default_factory=EnvVar(
+            "CLICKUP_AUTH",
+            "clickup auth token is required to be set",
+        )
+    )
+    workspace_id: str = field(
+        default_factory=EnvVar(
+            "CLICKUP_WORKSPACE_ID",
+            "clickup workspace id is required to be set",
+        )
+    )
+    user_id: str = field(
+        default_factory=EnvVar(
+            "CLICKUP_USER_ID",
+            "clickup user id is required to be set",
+        )
+    )
+    base_url: str = "https://api.clickup.com/api/v2"
+
+    def _get(self, endpoint: str, **query_params: str) -> JSONDataMap:
+        with get(
+            self.base_url + endpoint,
+            query_params,
+            headers={
+                "accept": "application/json",
+                "Authorization": self.auth_key,
+            },
+        ) as resp:
+            return resp.json()
+
+    def get_tasks(self, **filters: str) -> list[ClickupTask]:
+        data = self._get(
+            f"/team/{self.workspace_id}/task",
+            **{
+                "subtasks": "true",
+                "include_markdown_description": "true",
+                "assignees[]": self.user_id,
+            }
+            | filters,
+        ).get("tasks", [])
+        if isinstance(data, list):
+            return [ClickupTask.model_validate(t) for t in data if isinstance(t, dict)]
+
+        return []
+
+    def get_task(self, task_id: str) -> ClickupTask:
+        return ClickupTask.model_validate(
+            self._get(
+                f"/task/{task_id}",
+                include_markdown_description="true",
+            ),
+        )

eta/common.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+from dataclasses import dataclass
+from os import environ
+from typing import TypeAlias
+
+
+JSONDataScalar: TypeAlias = str | None | float | bool
+JSONDataList: TypeAlias = list["JSONDataScalar | JSONDataMap | JSONDataList"]
+JSONDataMap: TypeAlias = dict[str, "JSONDataScalar | JSONDataList | JSONDataMap"]
+JSONData: TypeAlias = "JSONDataMap | JSONDataList"
+
+
+@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

eta/main.py

@@ -0,0 +1,90 @@
+from lsprotocol.types import (
+    TEXT_DOCUMENT_COMPLETION,
+    TEXT_DOCUMENT_DOCUMENT_SYMBOL,
+    CompletionItem,
+    CompletionItemKind,
+    CompletionItemLabelDetails,
+    CompletionOptions,
+    CompletionParams,
+    DocumentSymbol,
+    DocumentSymbolParams,
+    NotebookDocumentSyncOptions,
+    Position,
+    Range,
+    SymbolKind,
+    TextDocumentSyncKind,
+    WorkDoneProgressBegin,
+    WorkDoneProgressEnd,
+    WorkDoneProgressReport,
+)
+from pygls.lsp.server import LanguageServer
+
+from .clickup import ClickupSession, ClickupTask
+
+
+class CustomServer(LanguageServer):
+    cache: dict[str, ClickupTask]
+    cu_session: ClickupSession
+
+    def __init__(
+        self,
+        name: str,
+        version: str,
+        text_document_sync_kind: TextDocumentSyncKind = TextDocumentSyncKind.Incremental,
+        notebook_document_sync: NotebookDocumentSyncOptions | None = None,
+    ) -> None:
+        super().__init__(name, version, text_document_sync_kind, notebook_document_sync)
+        self.cache = {}
+        self.cu_session = ClickupSession()
+        self.update_task_cache()
+
+    def update_task_cache(self) -> None:
+        self.protocol.progress.begin(
+            "startup", WorkDoneProgressBegin("Fetching Cache ...", percentage=0, cancellable=True)
+        )
+        self.cache = {}
+        tasks = self.cu_session.get_ta()
+        for ti, t in enumerate(tasks):
+            self.cache[t.id] = t
+            self.protocol.progress.report(
+                "startup",
+                WorkDoneProgressReport(
+                    message="Fetched Cache", percentage=int(100 * (1 + ti) / len(tasks))
+                ),
+            )
+
+        self.protocol.progress.end("startup", WorkDoneProgressEnd(message="Done Caching"))
+
+
+server = CustomServer("mrpy-server", "0.1.0")
+
+
+@server.feature(TEXT_DOCUMENT_DOCUMENT_SYMBOL)
+async def list_ids(params: DocumentSymbolParams) -> list[DocumentSymbol]:
+    return [
+        DocumentSymbol(
+            t.id,
+            SymbolKind.Enum,
+            Range(Position(i, 0), Position(i, 0)),
+            Range(Position(i, 0), Position(i, 0)),
+            detail=t.name,
+        )
+        for i, t in enumerate(server.cache.values())
+    ]
+
+
+@server.feature(TEXT_DOCUMENT_COMPLETION)
+async def complete_cu_ids(params: CompletionParams) -> list[CompletionItem]:
+    return [
+        CompletionItem(
+            t.name,
+            CompletionItemLabelDetails(detail=f" #{t.id}"),
+            kind=CompletionItemKind.Constant,
+            insert_text=f"[{t.name} #{t.id}](https://app.clickup.com/t/{t.id})",
+        )
+        for t in server.cache.values()
+    ]
+
+
+def main() -> None:
+    server.start_io()