From e4497791f318f2c3d6e9b9c1787e7eaecb3147d1 Mon Sep 17 00:00:00 2001 From: Slawomir Koszewski Date: Fri, 7 Nov 2025 00:20:54 +0100 Subject: [PATCH] Entity get reegineering. --- sk/devops.py | 239 ++++++++++++++++++++++----------------------------- 1 file changed, 101 insertions(+), 138 deletions(-) diff --git a/sk/devops.py b/sk/devops.py index 538d538..f2a35d8 100644 --- a/sk/devops.py +++ b/sk/devops.py @@ -2,11 +2,16 @@ from __future__ import annotations import requests import urllib.parse from uuid import UUID -from string import Template +import logging DEVOPS_SCOPE = "https://app.vssps.visualstudio.com/.default" DEVOPS_API_VERSION = "7.1" +# Get logger. It should be configured by the main application. +log = logging.getLogger(__name__) +# log.setLevel(logging.DEBUG) +# log.propagate = False + # Define a class decorator def auto_properties(mapping: dict[str,str]): @@ -22,24 +27,19 @@ def auto_properties(mapping: dict[str,str]): pass # Fetch repository details from the API if it is set to None or not existing - self._get() + self.get_auto_properties() 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_args(self, **kwargs): + for name in kwargs: + if name in self.__class__.__auto_properties__: + setattr(self, f"_{name}", kwargs.get(name, None)) 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 @@ -48,101 +48,48 @@ def auto_properties(mapping: dict[str,str]): for name in mapping: setattr(cls, name, make_property(name)) - setattr(cls, "set_auto_properties", set_auto_properties) + setattr(cls, "from_args", from_args) setattr(cls, "from_json", from_json) return cls return decorator -def devops(key: str, get_url: str, list_url: str = None, params: dict = {}): - def decorator(cls): - cls.__entity_key__ = key - cls.__entity_get_url__ = get_url # Use $key in the URL - cls.__entity_list_url__ = list_url # Use $key in the URL - cls.__entity_params__ = params - return cls - return decorator +def get_url(URL: str, token: str, api_version: str, params: dict = {}) -> requests.Response: + """Helper method to make GET requests to DevOps REST API.""" + if not URL or not token or not api_version: + raise ValueError("Organization URL, token, and API version must be set before making requests.") -class DevOps(): - """Base class for DevOps entities.""" + request_parameters = { + "api-version": api_version, + **params + } + log.debug(f"Making GET request", extra={"url": URL, "params": request_parameters, "http_method": "get"}) + r = requests.get(url=URL, params=request_parameters, headers={ + "Authorization": f"Bearer {token}" + }) + r.raise_for_status() # Ensure we raise an error for bad responses + return r.json() # Return parsed JSON response - def __init__(self, org_url: str, token: str, api_version: str = DEVOPS_API_VERSION): +class Organization(): + def __init__(self, org_url: str, token: str | None = None, api_version: str = DEVOPS_API_VERSION): self._org_url = org_url.rstrip("/") + "/" # Ensure trailing slash self._token = token self._api_version = api_version + log.debug(f"Created new Organization object", extra={"entity_class": "Organization"}) - 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(self, key: str): - if not hasattr(self.__class__, "__entity_key__") or not hasattr(self.__class__, "__entity_get_url__"): - raise NotImplementedError("Called _get on a class that has not been decorated with @devops.") - setattr(self, f"_{self.__class__.__entity_key__}", key) # Set the entity key - # Build the URL - url = Template(self.__class__.__entity_get_url__).substitute(key=key) - # Build parameters with key substituted - params = {} - if hasattr(self.__class__, "__entity_params__"): - params = {k: Template(v).substitute(key=key) for k, v in self.__class__.__entity_params__.items()} - - # Fetch the object data from the URL - r = self._get_url_path(url, params=params) - - # Populate attributes - self.from_json(r.json()) - - def _get_entity(self, key_name: str, get_url: str, params: dict = {}): - """ - 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): - super().__init__(org_url, token, api_version) + def get_path(self, path: str, params: dict = {}) -> dict: + return get_url( + URL=urllib.parse.urljoin(self._org_url, path), + token=self._token, + api_version=self._api_version, + params=params + ) @property def projects(self): if not hasattr(self, "_projects"): - self._projects = self._entities( - entity_class=Project, - key_name="id", - list_url="_apis/projects") + # Create Project objects + self._projects = [Project(org=self, **proj) for proj in self.get_path("_apis/projects").get("value", [])] return self._projects def __getitem__(self, key: str) -> Project: @@ -151,43 +98,53 @@ class Organization(DevOps): return project raise KeyError(f"Project with ID or name '{key}' not found.") + def __str__(self): + return f"Organization(url=\"{self._org_url}\")" + @auto_properties({ + "id": "id", "name": "name", "url": "url", "description": "description" }) -@devops("id", "_apis/projects/$key", "_apis/projects") -class Project(DevOps): - def _get(self): - self._get_entity( - key_name="id", - get_url=f"_apis/projects/{self._id}" - ) +class Project(): + def __init__(self, org: Organization, **kwargs): + self._org = org + self.from_args(**kwargs) - def __init__(self, org: Organization, id: str, **kwargs): - super().__init__(org._org_url, org._token, org._api_version) + if not hasattr(self, "_id") or self._id is None: + raise ValueError("Project ID must be provided.") try: - self._id = str(UUID(id)) + self._id = str(UUID(self._id)) except ValueError: - raise ValueError(f"Invalid project ID: {id}") + raise ValueError(f"Invalid project ID: {self._id}") - self.set_auto_properties(**kwargs) + log.debug(f"Created new Project object", extra={"entity_class": "Project"}) + + def get_auto_properties(self): + r = get_url( + URL=f"{self._org._org_url}_apis/projects/{self._id}", + token=self._org._token, + api_version=self._org._api_version + ) + self.from_json(r) def __str__(self): - return f"Project(name={self._name}, id={self._id})" + return f"Project(name=\"{self.name}\", id={self.id})" @property def id(self): return self._id + @property + def organization(self): + return self._org + @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") + self._repositories = [Repository(project=self, **repo) for repo in self._org.get_path(f"{self._id}/_apis/git/repositories").get("value", [])] return self._repositories def __getitem__(self, key: str) -> Repository: @@ -197,6 +154,7 @@ class Project(DevOps): raise KeyError(f"Repository with ID or name '{key}' not found.") @auto_properties({ + "id": "id", "name": "name", "url": "url", "default_branch": "defaultBranch", @@ -206,29 +164,30 @@ class Project(DevOps): "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) +class Repository(): + def __init__(self, project: Project, **kwargs): 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 + if "id" not in kwargs and "name" not in kwargs: + raise ValueError("Either repository ID or name must be provided.") + + if "id" in kwargs: + try: + UUID(kwargs.get("id")) # Check if it's a valid UUID + except ValueError: + raise ValueError("Invalid repository ID, must be a valid UUID.") # set other properties if provided - self.set_auto_properties(**kwargs) + self.from_args(**kwargs) + log.debug(f"Created new {self.__class__.__name__} object", extra={"entity_class": self.__class__.__name__}) + + def get_auto_properties(self): + id = self._id if hasattr(self, "_id") else self._name + if id is None or id == "": + raise ValueError("Repository ID or name must be set to fetch properties.") + + r = self._project.organization.get_path(path=f"{self._project.id}/_apis/git/repositories/{id}") + self.from_json(r) @property def id(self): @@ -268,23 +227,24 @@ class Repository(DevOps): "is_folder": "isFolder", "url": "url" }) -class Item(DevOps): +class Item(): 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 + log.debug(f"Created new {self.__class__.__name__} object", extra={"entity_class": self.__class__.__name__}) 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" - } - ) + pass + # 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): @@ -305,3 +265,6 @@ class Item(DevOps): ) else: raise ValueError("Items can only be listed for folder items.") + + def __str__(self): + return f"Item(path=\"{self._path}\")"