Implement watch_file and watch_repository

TODO:
- [ ] Implement `file_watcher` and `repository_watcher`
- [ ] Apply exception handler
- [ ] Update docstring

Author: ikhoon

centraldogma/base_client.py

@@ -40,7 +40,9 @@ def request(self, method: str, path: str, **kwargs) -> Union[Response]:
         return self._httpx_request(method, path, **kwargs)
 
     def _set_request_headers(self, method: str, **kwargs) -> Dict:
-        kwargs["headers"] = self.patch_headers if method == "patch" else self.headers
+        default_headers = self.patch_headers if method == "patch" else self.headers
+        headers = kwargs.get("headers")
+        kwargs["headers"] = dict(list(default_headers.items()) + list(headers.items())) if headers else default_headers
         return kwargs
 
     def _httpx_request(self, method: str, url: str, **kwargs) -> Response:

centraldogma/content_service.py

@@ -11,28 +11,39 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
+from dataclasses import asdict
+from enum import Enum
+from http import HTTPStatus
+from typing import List, Optional, TypeVar, Any
+
+from urllib.parse import quote
+
+from httpx import Response
+
 from centraldogma.base_client import BaseClient
 from centraldogma.data.change import Change
 from centraldogma.data.commit import Commit
 from centraldogma.data.content import Content
+from centraldogma.data.entry import Entry, EntryType
 from centraldogma.data.push_result import PushResult
-from dataclasses import asdict
-from enum import Enum
-from http import HTTPStatus
-from typing import List, Optional
+from centraldogma.data.revision import Revision
+from centraldogma.exceptions import CentralDogmaException
+from centraldogma.query import Query, QueryType
+
+T = TypeVar('T')
 
 
 class ContentService:
     def __init__(self, client: BaseClient):
         self.client = client
 
     def get_files(
-        self,
-        project_name: str,
-        repo_name: str,
-        path_pattern: Optional[str],
-        revision: Optional[int],
-        include_content: bool = False,
+            self,
+            project_name: str,
+            repo_name: str,
+            path_pattern: Optional[str],
+            revision: Optional[int],
+            include_content: bool = False,
     ) -> List[Content]:
         params = {"revision": revision} if revision else None
         path = f"/projects/{project_name}/repos/{repo_name}/"
@@ -67,11 +78,11 @@ def get_file(
         return Content.from_dict(resp.json())
 
     def push(
-        self,
-        project_name: str,
-        repo_name: str,
-        commit: Commit,
-        changes: List[Change],
+            self,
+            project_name: str,
+            repo_name: str,
+            commit: Commit,
+            changes: List[Change],
     ) -> PushResult:
         params = {
             "commitMessage": asdict(commit),
@@ -81,7 +92,79 @@ def push(
         }
         path = f"/projects/{project_name}/repos/{repo_name}/contents"
         resp = self.client.request("post", path, json=params)
-        return PushResult.from_dict(resp.json())
+        json: object = resp.json()
+        return PushResult.from_dict(json)
+
+    def watch_repository(self, project_name: str, repo_name: str, last_known_revision: Revision, path_pattern: str,
+                         timeout_millis: int) -> Optional[Revision]:
+        path = f"/projects/{project_name}/repos/{repo_name}/contents"
+        if path_pattern[0] != "/":
+            path += "/**/"
+
+        if path_pattern in ' ':
+            path_pattern = path_pattern.replace(" ", "%20")
+        path += path_pattern
+
+        response = self._watch(last_known_revision, timeout_millis, path)
+        if response.status_code == HTTPStatus.OK:
+            json = response.json()
+            return Revision(json["revision"])
+        elif response.status_code == HTTPStatus.NOT_MODIFIED:
+            return None
+        else:
+            # TODO(ikhoon): Handle excepitons after https://github.com/line/centraldogma-python/pull/11/ is merged.
+            pass
+
+    def watch_file(self, project_name: str, repo_name: str, last_known_revision: Revision, query: Query[T],
+                   timeout_millis) -> Optional[Entry[T]]:
+        path = f"/projects/{project_name}/repos/{repo_name}/contents/{query.path}"
+        if query.query_type == QueryType.JSON_PATH:
+            queries = [f"jsonpath={quote(expr)}" for expr in query.expressions]
+            path = f"{path}?{'&'.join(queries)}"
+
+        response = self._watch(last_known_revision, timeout_millis, path)
+        if response.status_code == HTTPStatus.OK:
+            json = response.json()
+            revision = Revision(json["revision"])
+            return self._to_entry(revision, json["entry"], query.query_type)
+        elif response.status_code == HTTPStatus.NOT_MODIFIED:
+            return None
+        else:
+            # TODO(ikhoon): Handle excepitons after https://github.com/line/centraldogma-python/pull/11/ is merged.
+            pass
+
+    @staticmethod
+    def _to_entry(revision: Revision, json: Any, query_type: QueryType) -> Entry:
+        entry_path = json["path"]
+        received_entry_type = EntryType[json["type"]]
+        content = json["content"]
+        if query_type == QueryType.IDENTITY_TEXT:
+            return Entry.text(revision, entry_path, content)
+        elif query_type == QueryType.IDENTITY or query_type == QueryType.JSON_PATH:
+            if received_entry_type != EntryType.JSON:
+                raise CentralDogmaException(
+                    f"invalid entry type. entry type: {received_entry_type} (expected: {query_type})")
+
+            return Entry.json(revision, entry_path, content)
+        else:  # query_type == QueryType.IDENTITY
+            if received_entry_type == EntryType.JSON:
+                return Entry.json(revision, entry_path, content)
+            elif received_entry_type == EntryType.TEXT:
+                return Entry.text(revision, entry_path, content)
+            else:  # received_entry_type == EntryType.DIRECTORY
+                return Entry.directory(revision, entry_path)
+
+    def _watch(
+            self,
+            last_known_revision: Revision,
+            timeout_millis: int,
+            path: str) -> Response:
+        normalized_timeout = (timeout_millis + 999) // 1000
+        headers = {
+            "if-none-match": f"{last_known_revision.major}",
+            "prefer": f"wait={normalized_timeout}"
+        }
+        return self.client.request("get", path, headers=headers, timeout=normalized_timeout)
 
     def _change_dict(self, data):
         return {

centraldogma/data/change.py

@@ -31,4 +31,4 @@ class ChangeType(Enum):
 class Change:
     path: str
     type: ChangeType
-    content: Optional[Union[map, str]] = None
+    content: Optional[Union[dict, str]] = None

centraldogma/data/entry.py

@@ -0,0 +1,77 @@
+#  Copyright 2021 LINE Corporation
+#
+#  LINE Corporation licenses this file to you under the Apache License,
+#  version 2.0 (the "License"); you may not use this file except in compliance
+#  with the License. You may obtain a copy of the License at:
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#  License for the specific language governing permissions and limitations
+#  under the License.
+from __future__ import annotations
+
+import json
+from enum import Enum, auto
+from typing import TypeVar, Generic, Any
+
+from centraldogma.data.revision import Revision
+from centraldogma.exceptions import EntryNoContentException
+
+
+class EntryType(Enum):
+    JSON = auto()
+    TEXT = auto()
+    DIRECTORY = auto()
+
+
+T = TypeVar('T')
+
+
+class Entry(Generic[T]):
+
+    @staticmethod
+    def text(revision: Revision, path: str, content: str) -> Entry[str]:
+        return Entry(revision, path, EntryType.TEXT, content)
+
+    @staticmethod
+    def json(revision: Revision, path: str, content: Any) -> Entry[Any]:
+        if type(content) is str:
+            content = json.loads(content)
+        return Entry(revision, path, EntryType.JSON, content)
+
+    @staticmethod
+    def directory(revision: Revision, path: str) -> Entry[None]:
+        return Entry(revision, path, EntryType.DIRECTORY, None)
+
+    def __init__(self, revision: Revision, path: str, entry_type: EntryType, content: T):
+        self.revision = revision
+        self.path = path
+        self.entry_type = entry_type
+        self._content = content
+        self._content_as_text = None
+
+    def has_content(self) -> bool:
+        return self.content is not None
+
+    @property
+    def content(self) -> T:
+        if self._content is None:
+            raise EntryNoContentException(f"{self.path} (type: {self.entry_type}, revision: {self.revision.major})")
+
+        return self._content
+
+    def content_as_text(self) -> str:
+        if self._content_as_text is not None:
+            return self._content_as_text
+
+        if self.entry_type == EntryType.TEXT:
+            self._content_as_text = self.content
+        elif self.entry_type == EntryType.DIRECTORY:
+            self._content_as_text = ''
+        else:
+            self._content_as_text = json.dumps(self.content)
+
+        return self._content_as_text

centraldogma/data/revision.py

@@ -0,0 +1,19 @@
+#  Copyright 2021 LINE Corporation
+#
+#  LINE Corporation licenses this file to you under the Apache License,
+#  version 2.0 (the "License"); you may not use this file except in compliance
+#  with the License. You may obtain a copy of the License at:
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#  License for the specific language governing permissions and limitations
+#  under the License.
+from dataclasses import dataclass
+
+
+@dataclass
+class Revision:
+    major: int

centraldogma/dogma.py

@@ -11,6 +11,9 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
+import os
+from typing import List, Optional, TypeVar, Generic
+
 from centraldogma.base_client import BaseClient
 from centraldogma.content_service import ContentService
 from centraldogma.data import (
@@ -22,13 +25,16 @@
     PushResult,
     Repository,
 )
+from centraldogma.data.entry import Entry
+from centraldogma.data.revision import Revision
 from centraldogma.project_service import ProjectService
+from centraldogma.query import Query
 from centraldogma.repository_service import RepositoryService
-from typing import List, Optional
-import os
 
+T = TypeVar('T')
 
-class Dogma:
+
+class Dogma(Generic[T]):
     DEFAULT_BASE_URL = "http://localhost:36462"
     DEFAULT_TOKEN = "anonymous"
 
@@ -179,3 +185,29 @@ def push(
             the content is supposed to be the new name.
         """
         return self.content_service.push(project_name, repo_name, commit, changes)
+
+    def watch_repository(
+            self,
+            project_name: str,
+            repo_name: str,
+            last_known_revision: Revision,
+            path_pattern: str,
+            timeout_millis: int) -> Optional[Revision]:
+        """
+        TODO(ikhoon): TBU
+        """
+        return self.content_service.watch_repository(project_name, repo_name, last_known_revision,
+                                                     path_pattern, timeout_millis)
+
+    def watch_file(
+            self,
+            project_name: str,
+            repo_name: str,
+            last_known_revision: Revision,
+            query: Query[T],
+            timeout_millis: int) -> Optional[Entry[T]]:
+        """
+        TODO(ikhoon): TBU
+        :rtype: object
+        """
+        return self.content_service.watch_file(project_name, repo_name, last_known_revision, query, timeout_millis)

centraldogma/exceptions.py

@@ -45,3 +45,8 @@ class UnauthorizedException(CentralDogmaException):
 
 class UnknownException(CentralDogmaException):
     pass
+
+
+class EntryNoContentException(CentralDogmaException):
+    """A CentralDogmaException that is raised when attempted to retrieve the content from a directory entry."""
+    pass

centraldogma/query.py

@@ -0,0 +1,51 @@
+#  Copyright 2021 LINE Corporation
+#
+#  LINE Corporation licenses this file to you under the Apache License,
+#  version 2.0 (the "License"); you may not use this file except in compliance
+#  with the License. You may obtain a copy of the License at:
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#  License for the specific language governing permissions and limitations
+#  under the License.
+from __future__ import annotations
+
+from enum import Enum, auto
+from typing import TypeVar, Generic, Any, List
+
+
+class QueryType(Enum):
+    IDENTITY = auto()
+    IDENTITY_TEXT = auto()
+    IDENTITY_JSON = auto()
+    JSON_PATH = auto()
+
+
+T = TypeVar('T')
+
+
+class Query(Generic[T]):
+
+    @staticmethod
+    def identity(path: str) -> Query[str]:
+        return Query(path=path, query_type=QueryType.IDENTITY, expressions=[])
+
+    @staticmethod
+    def text(path: str) -> Query[str]:
+        return Query(path=path, query_type=QueryType.IDENTITY_TEXT, expressions=[])
+
+    @staticmethod
+    def json(path: str) -> Query[Any]:
+        return Query(path=path, query_type=QueryType.IDENTITY_JSON, expressions=[])
+
+    @staticmethod
+    def json_path(path: str, json_paths: List[str]) -> Query[Any]:
+        return Query(path=path, query_type=QueryType.JSON_PATH, expressions=json_paths)
+
+    def __init__(self, path: str, query_type: QueryType, expressions: List[str]):
+        self.path = path
+        self.query_type = query_type
+        self.expressions = expressions