diff --git a/devops.py b/devops.py index 2fc4af6..63583ab 100644 --- a/devops.py +++ b/devops.py @@ -52,13 +52,18 @@ def auto_properties(mapping: dict[str,str] | None = None): 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 +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.") - def get(self, path: str, params: dict = {}): request_parameters = { "api-version": self._api_version, **params @@ -71,37 +76,55 @@ class Organization: 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): - r = self.get("_apis/projects") - - # Return a list of Project instances - projects_data = r.json().get("value", []) - - project_list = [] - for proj in projects_data: - p = Project(self, project_id=proj.get("id")) - p.from_json(proj) - project_list.append(p) - - return project_list + return self._entities( + entity_class=Project, + key_name="id", + list_url="_apis/projects") @auto_properties({ "name": "name", "url": "url", "description": "description" }) -class Project(): - def _get(self, project_id: str): - r = self._org.get(f"_apis/projects/{project_id}") +class Project(DevOps): + def _get(self, id: str): + r = self.get(f"_apis/projects/{id}") self.from_json(r.json()) - def __init__(self, org: Organization, project_id: str, **kwargs): - self._org = org + 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(project_id)) + self._id = str(UUID(id)) except ValueError: - raise ValueError(f"Invalid project ID: {project_id}") + raise ValueError(f"Invalid project ID: {id}") self.set_auto_properties(**kwargs) @@ -109,21 +132,18 @@ class Project(): return f"Project(name={self._name}, id={self._id})" def get_path(self, path: str, params: dict = {}): - return self._org.get(f"{self._id}/{path.lstrip('/')}", params=params) + return self.get(f"{self._id}/{path.lstrip('/')}", params=params) + + @property + def id(self): + return self._id @property def repositories(self): - r = self.get_path(f"_apis/git/repositories") - - repos_data = r.json().get("value", []) - - repository_list = [] - for repo in repos_data: - r = Repository(self, id_or_name=repo.get("id")) - r.from_json(repo) - repository_list.append(r) - - return repository_list + return self._entities( + entity_class=Repository, + key_name="id", + list_url=f"{self._id}/_apis/git/repositories") @auto_properties({ "name": "name", @@ -135,7 +155,7 @@ class Project(): "ssh_url": "sshUrl", "web_url": "webUrl" }) -class Repository: +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) @@ -144,12 +164,17 @@ class Repository: 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 - self._get(id_or_name) + 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 @@ -173,17 +198,15 @@ class Repository: @property def items(self): # GET https://dev.azure.com/{organization}/{project}/_apis/git/repositories/{repositoryId}/items?api-version=7.1 - - # Return a list of Item instances - items_data = self._get_items().get("value", []) - - item_list = [] - for item in items_data: - i = Item(self, path=item.get("path")) - i.from_json(item) - item_list.append(i) - - return item_list + 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", @@ -192,7 +215,7 @@ class Repository: "is_folder": "isFolder", "url": "url" }) -class Item: +class Item(DevOps): def _get(self, path): r = self._repository.get_path(f"items/{urllib.parse.quote(path)}") @@ -202,6 +225,9 @@ class Item: 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