from __future__ import annotations import requests import urllib.parse from uuid import UUID import logging from sk.logger import log_entity_creation 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 log.debug(f"Auto-fetching property '{name}' for {self.__class__.__name__}", extra={"property_name": name}) self.get_auto_properties() return getattr(self, private_var) def setter(self, value): setattr(self, private_var, value) return property(fget=getter, fset=setter) def from_args(self, **kwargs): for name in kwargs: if name in self.__class__.__auto_properties__: log.debug(f"Setting property '{name}' for {self.__class__.__name__} from args", extra={"property_name": name}) setattr(self, f"_{name}", kwargs.get(name, None)) def from_json(self, json_data: dict): for json_name in self.__class__.__auto_properties_reversed__: setattr(self, f"_{self.__class__.__auto_properties_reversed__[json_name]}", json_data.get(json_name, None)) def decorator(cls): cls.__auto_properties__ = mapping # Make a copy of the mapping cls.__auto_properties_reversed__ = {v: k for k, v in mapping.items()} # Store reversed mapping for JSON parsing # 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 # Return response 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 log_entity_creation(log, Organization, self._org_url) def get_path(self, path: str, params: dict = {}) -> requests.Response: 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").json().get("value", [])] return self._projects def __getitem__(self, key: str) -> Project: for project in self.projects: if project.id == key or project.name == key: # type: ignore[attr-defined] 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) # type: ignore[attr-defined] 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_entity_creation(log, Project, self.id) 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.json()) # type: ignore[attr-defined] def __str__(self): return f"Project(name=\"{self.name}\", id={self.id})" # type: ignore[attr-defined] @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").json().get("value", [])] return self._repositories def __getitem__(self, key: str) -> Repository: for repo in self.repositories: if repo.id == key or repo.name == key: # type: ignore[attr-defined] 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) # type: ignore[attr-defined] log_entity_creation(log, Repository, self.id) def get_auto_properties(self): id = self._id if hasattr(self, "_id") else self._name # type: ignore[attr-defined] 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.json()) # type: ignore[attr-defined] @property def id(self): return self._id # type: ignore[attr-defined] @property def project(self): return self._project def __str__(self): return f"Repository(name={self.name}, id={self._id})" # type: ignore[attr-defined] @property def items(self): log.debug(f"Fetching items for repository '{self.name}'", extra={"repository_name": self.name}) # type: ignore[attr-defined] if not hasattr(self, "_items"): root = Item(repository=self, path="/", git_object_type="tree") self._items = root.get_child_items() 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({ "path": "path", "object_id": "objectId", "git_object_type": "gitObjectType", "commit_id": "commitId", "is_folder": "isFolder", "url": "url" }) class Item(): def __init__(self, repository: Repository, **kwargs): self._repository = repository self.from_args(**kwargs) # type: ignore[attr-defined] log_entity_creation(log, Item, self.path) def get_auto_properties(self): r = self._repository._project.organization.get_path( path=f"{self._repository._project.id}/_apis/git/repositories/{self._repository.id}/items", params={ "path": self.path, "$format": "json", "recursionLevel": "none" } ) self.from_json(r.json()) # type: ignore[attr-defined] @property def path(self): return self._path # type: ignore[attr-defined] def get_child_items(self) -> list[Item]: """Get child items if this item is a folder.""" if self.git_object_type != "tree": # type: ignore[attr-defined] raise ValueError("Child items can only be fetched for folder items.") # Fetch child objects objects = self._repository._project.organization.get_path( path=f"{self._repository._project.id}/_apis/git/repositories/{self._repository.id}/items", params={ "scopePath": self.path, "recursionLevel": "oneLevel" } ).json().get("value", []) child_items = [] for obj in objects: i = Item(repository=self._repository, path=obj.get("path")) i.from_json(obj) # type: ignore[attr-defined] child_items.append(i) return child_items @property def children(self): if not hasattr(self, "_children"): self._children = self.get_child_items() return self._children def __str__(self): return f"Item(path=\"{self._path}\" type={self.git_object_type})" # type: ignore[attr-defined]