Moved devops package to sk module.

sk/devops.py

@@ -0,0 +1,282 @@
+from __future__ import annotations
+import requests
+import urllib.parse
+from uuid import UUID
+from azure.identity import DefaultAzureCredential
+
+DEVOPS_SCOPE = "https://app.vssps.visualstudio.com/.default"
+DEVOPS_API_VERSION = "7.1"
+
+# Define a class decorator
+def auto_properties(mapping: dict[str,str] | None = None):
+
+    def make_property(name: str):
+        private_var = f"_{name}"
+
+        def getter(self):
+            try:
+                i = getattr(self, private_var)
+                if i is not None:
+                    return i
+            except AttributeError:
+                pass
+
+            # Fetch repository details from the API if it is set to None or not existing
+            self._get()
+            return getattr(self, private_var)
+
+        return property(fget=getter)
+
+    def set_auto_properties(self, **kwargs):
+        allowed = set(self.__class__.__auto_properties__)
+        unknown = [k for k in kwargs if k not in allowed]
+        if unknown:
+            raise ValueError(f"Unknown properties for {self.__class__.__name__}: {', '.join(unknown)}")
+        for k, v in kwargs.items():
+            setattr(self, f"_{k}", v)
+        return self
+
+    def from_json(self, json_data: dict):
+        for name in self.__class__.__auto_properties__:
+            setattr(self, f"_{name}", json_data.get(self.__class__.__auto_properties__[name], None))
+        return self
+
+    def decorator(cls):
+        cls.__auto_properties__ = mapping # Make a copy of the mapping
+
+        # Create properties dynamically
+        for name in mapping:
+            setattr(cls, name, make_property(name))
+
+        setattr(cls, "set_auto_properties", set_auto_properties)
+        setattr(cls, "from_json", from_json)
+
+        return cls
+    return decorator
+
+class DevOps():
+    """Base class for DevOps entities."""
+
+    def __init__(self, org_url: str, token: str, api_version: str = DEVOPS_API_VERSION):
+        self._org_url = org_url.rstrip("/") + "/" # Ensure trailing slash
+        self._token = token
+        self._api_version = api_version
+
+    def _get_url_path(self, path: str, params: dict = {}) -> requests.Response:
+        if not self._org_url or not self._token or not self._api_version:
+            raise ValueError("Organization URL, token, and API version must be set before making requests.")
+
+        request_parameters = {
+            "api-version": self._api_version,
+            **params
+        }
+        encoded_path = urllib.parse.quote(path.lstrip("/")) # Ensure single slash between base and path
+        url = self._org_url + encoded_path
+        r = requests.get(url=url, params=request_parameters, headers={
+            "Authorization": f"Bearer {self._token}"
+        })
+        r.raise_for_status() # Ensure we raise an error for bad responses
+        return r
+
+    def _get_entity(self, key_name: str, get_url: str, params: dict = {}) -> object:
+        """
+        Each entity class can use this method to populate its attributes, by defining
+        its own _get method that calls this one with the key name,
+        and the URL.
+        """
+        r = self._get_url_path(get_url, params=params) # Fetch the object data from the URL
+        setattr(self, f"_{key_name}", r.json().get(key_name, None)) # Set the key attribute
+        self.from_json(r.json()) # Populate other attributes from JSON
+
+    def _entity(self, entity_class: type, key_name: str, entity_data: dict) -> object:
+        """A generic method to create an entity from JSON data."""
+        args = { key_name: entity_data.get(key_name) }
+        e = entity_class(self, **args)
+        e.from_json(entity_data)
+        return e
+
+    def _entities(self, entity_class: type, key_name: str, list_url: str, params: dict = {}) -> list[object]:
+        """A generic method to retrieve a list of entities."""
+        r = self._get_url_path(list_url, params=params)
+        entities_data = r.json().get("value", [])
+
+        entities_list = []
+        for entity in entities_data:
+            entities_list.append(self._entity(entity_class, key_name, entity))
+
+        return entities_list
+
+class Organization(DevOps):
+    def __init__(self, org_url: str, token: str | None = None, api_version: str = DEVOPS_API_VERSION):
+        if token is None:
+            token = DefaultAzureCredential().get_token(DEVOPS_SCOPE).token
+        super().__init__(org_url, token, api_version)
+
+    @property
+    def projects(self):
+        if not hasattr(self, "_projects"):
+            self._projects = self._entities(
+                entity_class=Project,
+                key_name="id",
+                list_url="_apis/projects")
+        return self._projects
+
+    def __getitem__(self, key: str) -> Project:
+        for project in self.projects:
+            if project.id == key or project.name == key:
+                return project
+        raise KeyError(f"Project with ID or name '{key}' not found.")
+
+@auto_properties({
+    "name":        "name",
+    "url":         "url",
+    "description": "description"
+    })
+class Project(DevOps):
+    def _get(self):
+        self._get_entity(
+            key_name="id",
+            get_url=f"_apis/projects/{self._id}"
+        )
+
+    def __init__(self, org: Organization, id: str, **kwargs):
+        super().__init__(org._org_url, org._token, org._api_version)
+
+        try:
+            self._id = str(UUID(id))
+        except ValueError:
+            raise ValueError(f"Invalid project ID: {id}")
+
+        self.set_auto_properties(**kwargs)
+
+    def __str__(self):
+        return f"Project(name={self._name}, id={self._id})"
+
+    @property
+    def id(self):
+        return self._id
+
+    @property
+    def repositories(self):
+        if not hasattr(self, "_repositories"):
+            self._repositories = self._entities(
+                entity_class=Repository,
+                key_name="id",
+                list_url=f"{self._id}/_apis/git/repositories")
+        return self._repositories
+
+    def __getitem__(self, key: str) -> Repository:
+        for repo in self.repositories:
+            if repo.id == key or repo.name == key:
+                return repo
+        raise KeyError(f"Repository with ID or name '{key}' not found.")
+
+@auto_properties({
+    "name": "name",
+    "url": "url",
+    "default_branch": "defaultBranch",
+    "is_disabled": "isDisabled",
+    "is_in_maintenance": "isInMaintenance",
+    "remote_url": "remoteUrl",
+    "ssh_url": "sshUrl",
+    "web_url": "webUrl"
+    })
+class Repository(DevOps):
+    def _get(self):
+        self._get_entity(
+            key_name="id",
+            get_url=f"{self._project.id}/_apis/git/repositories/{self._id}"
+        )
+
+    def __init__(self, project: Project, id: str, **kwargs):
+        super().__init__(project._org_url, project._token, project._api_version)
+        self._project = project
+        self._id = id
+
+        try:
+            UUID(id) # Check if it's a valid UUID
+        except ValueError:
+            # Called with a repository name, fetch by name
+            self._get()
+            if kwargs:
+                raise ValueError("Automatic properties cannot be set when retrieving by name.")
+            return
+
+        # set other properties if provided
+        self.set_auto_properties(**kwargs)
+
+    @property
+    def id(self):
+        return self._id
+
+    @property
+    def project(self):
+        return self._project
+
+    def __str__(self):
+        return f"Repository(name={self.name}, id={self._id})"
+
+    @property
+    def items(self):
+        if not hasattr(self, "_items"):
+            self._items = self._entities(
+                entity_class=Item,
+                key_name="path",
+                list_url=f"{self._project.id}/_apis/git/repositories/{self._id}/items",
+                params={
+                    "scopePath": "/",
+                    "recursionLevel": "oneLevel"
+                }
+            )
+        return self._items
+
+    def __getitem__(self, key: str) -> Item:
+        for item in self.items:
+            if item.path == key:
+                return item
+        raise KeyError(f"Item with path '{key}' not found.")
+
+@auto_properties({
+    "object_id":       "objectId",
+    "git_object_type": "gitObjectType",
+    "commit_id":       "commitId",
+    "is_folder":       "isFolder",
+    "url":             "url"
+    })
+class Item(DevOps):
+    def __init__(self, repository: Repository, path: str, **kwargs):
+        super().__init__(repository._org_url, repository._token, repository._api_version)
+        self._repository = repository
+        self._path = path
+        self.set_auto_properties(**kwargs) # set properties defined in decorator
+
+    def _get(self):
+        self._get_entity(
+            key_name="path",
+            get_url=f"{self._repository._project.id}/_apis/git/repositories/{self._repository.id}/items",
+            params={
+                "path": self._path,
+                "$format": "json",
+                "recursionLevel": "none"
+            }
+        )
+
+    @property
+    def path(self):
+        return self._path
+
+    @property
+    def children(self):
+        """List items under this item if it is a folder."""
+        if self.git_object_type == "tree":
+            return self._entities(
+                entity_class=Item,
+                key_name="path",
+                list_url=f"{self._repository._project.id}/_apis/git/repositories/{self._repository.id}/items",
+                params={
+                    "scopePath": self._path,
+                    "recursionLevel": "oneLevel"
+                }
+            )
+        else:
+            raise ValueError("Items can only be listed for folder items.")