from __future__ import annotations import requests import urllib.parse from uuid import UUID 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]): 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_auto_properties() return getattr(self, private_var) return property(fget=getter) 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)) 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, "from_args", from_args) setattr(cls, "from_json", from_json) 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.") 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 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_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"): # 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: 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.") def __str__(self): return f"Organization(url=\"{self._org_url}\")" @auto_properties({ "id": "id", "name": "name", "url": "url", "description": "description" }) class Project(): def __init__(self, org: Organization, **kwargs): self._org = org self.from_args(**kwargs) if not hasattr(self, "_id") or self._id is None: raise ValueError("Project ID must be provided.") try: self._id = str(UUID(self._id)) except ValueError: raise ValueError(f"Invalid project ID: {self._id}") 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})" @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 = [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: 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({ "id": "id", "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(): def __init__(self, project: Project, **kwargs): self._project = project 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.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): 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(): def __init__(self, repository: Repository, path: str, **kwargs): 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): 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): 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.") def __str__(self): return f"Item(path=\"{self._path}\")"