import requests import urllib.parse from uuid import UUID 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(getattr(self, "_id")) 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): self._org_url = None self._token = None self._api_version = None def get(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 _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(list_url, params=params) entities_data = r.json().get("value", []) entities_list = [] for entity in entities_data: args = { key_name: entity.get(key_name) } e = entity_class(self, **args) e.from_json(entity) entities_list.append(e) return entities_list class Organization(DevOps): 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 @classmethod def organization(org_url: str, token: str, api_version: str = DEVOPS_API_VERSION): return Organization(org_url, token, api_version) @property def projects(self): return self._entities( entity_class=Project, key_name="id", list_url="_apis/projects") @auto_properties({ "name": "name", "url": "url", "description": "description" }) class Project(DevOps): def _get(self, id: str): r = self.get(f"_apis/projects/{id}") self.from_json(r.json()) def __init__(self, parent: Organization, id: str, **kwargs): self._org_url = parent._org_url self._token = parent._token self._api_version = parent._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})" def get_path(self, path: str, params: dict = {}): return self.get(f"{self._id}/{path.lstrip('/')}", params=params) @property def id(self): return self._id @property def repositories(self): return self._entities( entity_class=Repository, key_name="id", list_url=f"{self._id}/_apis/git/repositories") @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, repo_name: str): r = self._project.get_path(f"_apis/git/repositories/{urllib.parse.quote(repo_name)}") self._id = r.json().get("id", None) self.from_json(r.json()) def __init__(self,_project: Project, id_or_name: str, **kwargs): self._project = _project self._org_url = _project._org_url self._token = _project._token self._api_version = _project._api_version try: self._id = str(UUID(id_or_name)) except ValueError: # Id not available, use API to get the repository object r = self.get(f"{self._project.id}/_apis/git/repositories/{urllib.parse.quote(id_or_name)}") self._id = r.json().get("id", None) self.from_json(r.json()) # Successfully retrieved the repository by name, # throw an error if automatic properties were set and # id_or_name was not set to repository id if kwargs: raise ValueError("Automatic properties cannot be set when retrieving by name.") return # set other properties if provided self.set_auto_properties(**kwargs) def __str__(self): return f"Repository(name={self.name}, id={self._id})" def get_path(self, path: str, params: dict = {}): return self._project.get_path(f"_apis/git/repositories/{self._id}/{path.lstrip('/')}", params=params) def _get_items(self, scope_path: str = "/", recursion_level: str = "oneLevel"): r = self.get_path(f"items", params={"scopePath": scope_path, "recursionLevel": recursion_level}) return r.json() @property def items(self): # GET https://dev.azure.com/{organization}/{project}/_apis/git/repositories/{repositoryId}/items?api-version=7.1 return self._entities( entity_class=Item, key_name="path", list_url=f"{self._project.id}/_apis/git/repositories/{self._id}/items", params={ "scopePath": "/", "recursionLevel": "oneLevel" } ) @auto_properties({ "object_id": "objectId", "git_object_type": "gitObjectType", "commit_id": "commitId", "is_folder": "isFolder", "url": "url" }) class Item(DevOps): def _get(self, path): r = self._repository.get_path(f"items/{urllib.parse.quote(path)}") for name in self.__auto_properties__: setattr(self, f"_{name}", r.json().get(name, None)) def __init__(self, repository: Repository, path: str, **kwargs): self._repository = repository self._org_url = repository._org_url self._token = repository._token self._api_version = repository._api_version self._path = path self.set_auto_properties(**kwargs) # set properties defined in decorator @property def path(self): return self._path