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 Organization: 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(self, path: str, params: dict = {}): 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 @property def projects(self): r = self.get("_apis/projects") # Return a list of Project instances projects_data = r.json().get("value", []) return [ Project(self, id=proj.get("id"), name=proj.get("name"), url=proj.get("url"), description=proj.get("description"), ) for proj in projects_data ] @auto_properties(names=["name", "url", "description"]) class Project: def __init__(self, org: Organization, id: str, name: str | None = None, url: str | None = None, description: str | None = None ): self._org = org # Check, if the id is a valid UUID try: self._id = str(UUID(id)) except ValueError: raise ValueError(f"Invalid project ID: {self._id}") if name is not None and url is not None and name != "" and url != "": self._name = name self._url = url self._description = description if description is not None else "" # Ensure description is a string else: # Fetch project details from the API r = org.get(f"_apis/projects/{self._id}") proj_data = r.json() self._id = proj_data.get("id", "") self._name = proj_data.get("name", "") self._url = proj_data.get("url", "") self._description = proj_data.get("description", "") def __str__(self): return f"Project(name={self._name}, id={self._id})" def get(self, path: str, params: dict = {}): return self._org.get(f"{self._id}/{path.lstrip('/')}", params=params) @property def repositories(self): r = self.get(f"_apis/git/repositories") repos_data = r.json().get("value", []) # Return a list of Repository instances return [ Repository(self, id_or_name=repo.get("id"), name=repo.get("name"), url=repo.get("url"), default_branch=repo.get("defaultBranch"), disabled=repo.get("isDisabled"), in_maintenance=repo.get("isInMaintenance"), remote_url=repo.get("remoteUrl"), ssh_url=repo.get("sshUrl"), web_url=repo.get("webUrl"), ) for repo in repos_data ] @auto_properties(names=["name", "url", "default_branch", "disabled", "in_maintenance", "remote_url", "ssh_url", "web_url"]) class Repository: def _get(self, repo_name: str): r = self._project.get(f"_apis/git/repositories/{urllib.parse.quote(repo_name)}") json_data = r.json() self._id = json_data.get("id", "") self._name = json_data.get("name", "") self._url = json_data.get("url", "") self._default_branch = json_data.get("defaultBranch", "") self._is_disabled = json_data.get("isDisabled", False) self._is_in_maintenance = json_data.get("isInMaintenance", False) self._remote_url = json_data.get("remoteUrl", "") self._ssh_url = json_data.get("sshUrl", "") self._web_url = json_data.get("webUrl", "") def __init__(self, _project: Project, id_or_name: str, name: str | None = None, url: str | None = None, default_branch: str | None = None, disabled: bool | None = None, in_maintenance: bool | None = None, remote_url: str | None = None, ssh_url: str | None = None, web_url: str | None = None ): self._project = _project try: self._id = str(UUID(id_or_name)) except ValueError: # Id not available, use API to get the repository object self._get(id_or_name) return # set other properties if provided self._name = name if name is not None else None self._url = url if url is not None else None self._default_branch = default_branch if default_branch is not None else None self._is_disabled = disabled if disabled is not None else None self._is_in_maintenance = in_maintenance if in_maintenance is not None else None self._remote_url = remote_url if remote_url is not None else None self._ssh_url = ssh_url if ssh_url is not None else None self._web_url = web_url if web_url is not None else None def __str__(self): return f"Repository(name={self._name}, id={self._id})" def _get_path(self, path: str, params: dict = {}): return self._project.get(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 items_data = self._get_items().get("value", []) # Return a list of Item instances return [ Item(self, path=item.get("path"), object_id=item.get("objectId"), git_object_type=item.get("gitObjectType"), commit_id=item.get("commitId"), is_folder=item.get("isFolder"), url=item.get("url"), ) for item in items_data ] @auto_properties(names=["object_id", "git_object_type", "commit_id", "folder", "url"]) class Item: 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, object_id: str | None = None, git_object_type: str | None = None, commit_id: str | None = None, is_folder: bool | None = None, url: str | None = None ): self._repository = repository self._path = path self._object_id = object_id if object_id is not None else None self._git_object_type = git_object_type if git_object_type is not None else None self._commit_id = commit_id if commit_id is not None else None self._is_folder = is_folder if is_folder is not None else None self._url = url if url is not None else None