239 lines
7.7 KiB
Python
239 lines
7.7 KiB
Python
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()
|
|
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(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 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(self):
|
|
self._get_entity(
|
|
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(self):
|
|
self._get_entity(
|
|
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()
|
|
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
|
|
|
|
@property
|
|
def project(self):
|
|
return self._project
|
|
|
|
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
|
|
|
|
def _get(self):
|
|
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"
|
|
}
|
|
)
|
|
|
|
@property
|
|
def path(self):
|
|
return self._path
|