Implemented repository watcher

Author: ikhoon

centraldogma/base_client.py

@@ -42,7 +42,11 @@ def request(self, method: str, path: str, **kwargs) -> Union[Response]:
     def _set_request_headers(self, method: str, **kwargs) -> Dict:
         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
+        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

@@ -30,20 +30,20 @@
 from centraldogma.exceptions import CentralDogmaException
 from centraldogma.query import Query, QueryType
 
-T = TypeVar('T')
+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}/"
@@ -78,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),
@@ -95,13 +95,19 @@ def push(
         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]:
+    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 ' ':
+        if path_pattern in " ":
             path_pattern = path_pattern.replace(" ", "%20")
         path += path_pattern
 
@@ -115,8 +121,14 @@ def watch_repository(self, project_name: str, repo_name: str, last_known_revisio
             # 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]]:
+    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]
@@ -143,7 +155,8 @@ def _to_entry(revision: Revision, json: Any, query_type: QueryType) -> Entry:
         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})")
+                    f"invalid entry type. entry type: {received_entry_type} (expected: {query_type})"
+                )
 
             return Entry.json(revision, entry_path, content)
         else:  # query_type == QueryType.IDENTITY
@@ -155,16 +168,16 @@ def _to_entry(revision: Revision, json: Any, query_type: QueryType) -> Entry:
                 return Entry.directory(revision, entry_path)
 
     def _watch(
-            self,
-            last_known_revision: Revision,
-            timeout_millis: int,
-            path: str) -> Response:
+        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}"
+            "prefer": f"wait={normalized_timeout}",
         }
-        return self.client.request("get", path, headers=headers, timeout=normalized_timeout)
+        return self.client.request(
+            "get", path, headers=headers, timeout=normalized_timeout
+        )
 
     def _change_dict(self, data):
         return {

centraldogma/data/entry.py

@@ -27,11 +27,10 @@ class EntryType(Enum):
     DIRECTORY = auto()
 
 
-T = TypeVar('T')
+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)
@@ -46,7 +45,9 @@ def json(revision: Revision, path: str, content: Any) -> Entry[Any]:
     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):
+    def __init__(
+        self, revision: Revision, path: str, entry_type: EntryType, content: T
+    ):
         self.revision = revision
         self.path = path
         self.entry_type = entry_type
@@ -59,7 +60,9 @@ def has_content(self) -> bool:
     @property
     def content(self) -> T:
         if self._content is None:
-            raise EntryNoContentException(f"{self.path} (type: {self.entry_type}, revision: {self.revision.major})")
+            raise EntryNoContentException(
+                f"{self.path} (type: {self.entry_type}, revision: {self.revision.major})"
+            )
 
         return self._content
 
@@ -70,7 +73,7 @@ def content_as_text(self) -> str:
         if self.entry_type == EntryType.TEXT:
             self._content_as_text = self.content
         elif self.entry_type == EntryType.DIRECTORY:
-            self._content_as_text = ''
+            self._content_as_text = ""
         else:
             self._content_as_text = json.dumps(self.content)
 

centraldogma/data/revision.py

@@ -11,9 +11,24 @@
 #  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 dataclasses import dataclass
 
 
 @dataclass
 class Revision:
-    major: int
+    def __init__(self, major):
+        self.major = major
+
+    @staticmethod
+    def init() -> Revision:
+        return _INIT
+
+    @staticmethod
+    def head() -> Revision:
+        return _HEAD
+
+
+_INIT = Revision(1)
+_HEAD = Revision(-1)

centraldogma/dogma.py

@@ -12,7 +12,7 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 import os
-from typing import List, Optional, TypeVar, Generic
+from typing import List, Optional, TypeVar, Generic, Callable
 
 from centraldogma.base_client import BaseClient
 from centraldogma.content_service import ContentService
@@ -30,11 +30,15 @@
 from centraldogma.project_service import ProjectService
 from centraldogma.query import Query
 from centraldogma.repository_service import RepositoryService
+from centraldogma.watcher import Watcher
 
-T = TypeVar('T')
+T = TypeVar("T")
+U = TypeVar("U")
 
+_DEFAULT_WATCH_TIMEOUT_MILLIS = 1 * 60 * 1000  # 1 minute
 
-class Dogma(Generic[T]):
+
+class Dogma:
     DEFAULT_BASE_URL = "http://localhost:36462"
     DEFAULT_TOKEN = "anonymous"
 
@@ -187,27 +191,60 @@ def push(
         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]:
+        self,
+        project_name: str,
+        repo_name: str,
+        last_known_revision: Revision,
+        path_pattern: str,
+        timeout_millis: int = _DEFAULT_WATCH_TIMEOUT_MILLIS,
+    ) -> Optional[Revision]:
         """
         TODO(ikhoon): TBU
         """
-        return self.content_service.watch_repository(project_name, repo_name, last_known_revision,
-                                                     path_pattern, timeout_millis)
+        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]]:
+        self,
+        project_name: str,
+        repo_name: str,
+        last_known_revision: Revision,
+        query: Query[T],
+        timeout_millis: int = _DEFAULT_WATCH_TIMEOUT_MILLIS,
+    ) -> Optional[Entry[T]]:
         """
         TODO(ikhoon): TBU
         :rtype: object
         """
-        return self.content_service.watch_file(project_name, repo_name, last_known_revision, query, timeout_millis)
+        return self.content_service.watch_file(
+            project_name, repo_name, last_known_revision, query, timeout_millis
+        )
+
+    def repository_watcher(
+        self,
+        project_name: str,
+        repo_name: str,
+        path_pattern: str,
+        function: Callable[[Revision], T] = lambda x: x,
+    ) -> Watcher[T]:
+        from centraldogma.repository_watcher import RepositoryWatcher
+
+        watcher = RepositoryWatcher(
+            self, project_name, repo_name, path_pattern, function
+        )
+        watcher.start()
+        return watcher
+
+    def file_watcher(
+        self,
+        project_name: str,
+        repo_name: str,
+        query: Query[T],
+        function: Callable[[T], U] = lambda x: x,
+    ) -> Watcher[U]:
+        from centraldogma.repository_watcher import FileWatcher
+
+        watcher = FileWatcher(self, project_name, repo_name, query, function)
+        watcher.start()
+        return watcher

centraldogma/exceptions.py

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

centraldogma/query.py

@@ -24,11 +24,10 @@ class QueryType(Enum):
     JSON_PATH = auto()
 
 
-T = TypeVar('T')
+T = TypeVar("T")
 
 
 class Query(Generic[T]):
-
     @staticmethod
     def identity(path: str) -> Query[str]:
         return Query(path=path, query_type=QueryType.IDENTITY, expressions=[])