222 lines
8.2 KiB
Python
222 lines
8.2 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(names=None):
|
|
names = names or []
|
|
|
|
def make_property(name):
|
|
private_var = f"_{name}"
|
|
|
|
def getter(self):
|
|
i = getattr(self, private_var)
|
|
if i is not None:
|
|
return i
|
|
# Fetch repository details from the API
|
|
self._get(getattr(self, "_id"))
|
|
return getattr(self, private_var)
|
|
|
|
return property(getter)
|
|
|
|
def decorator(cls):
|
|
cls.__auto_properties__ = names
|
|
for name in names:
|
|
setattr(cls, name, make_property(name))
|
|
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
|