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_attributes() 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_attributes(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_attributes 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, api_version: str = DEVOPS_API_VERSION): super().__init__(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_attributes(self): self._get_entity_attributes( 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): 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_attributes(self): self._get_entity_attributes( 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_attributes() 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 def __str__(self): return f"Repository(name={self.name}, id={self._id})" @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 __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 @property def path(self): return self._path