Added Repository class and autoproperties.
This commit is contained in:
109
devops.py
109
devops.py
@@ -1,4 +1,3 @@
|
||||
from unicodedata import name
|
||||
import requests
|
||||
import urllib.request
|
||||
import urllib.parse
|
||||
@@ -8,6 +7,29 @@ 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):
|
||||
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
|
||||
@@ -42,8 +64,15 @@ class Organization:
|
||||
) 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):
|
||||
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
|
||||
@@ -66,9 +95,77 @@ class Project:
|
||||
self._url = proj_data.get("url", "")
|
||||
self._description = proj_data.get("description", "")
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self._name
|
||||
|
||||
def __str__(self):
|
||||
return f"Project(name={self._name}, id={self._id})"
|
||||
|
||||
@property
|
||||
def repositories(self):
|
||||
r = self._org.get(f"{self._id}/_apis/git/repositories")
|
||||
r.raise_for_status()
|
||||
|
||||
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._org.get(f"{self._project._id}/_apis/git/repositories/{urllib.parse.quote(repo_name)}")
|
||||
r.raise_for_status()
|
||||
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._disabled = json_data.get("isDisabled", False)
|
||||
self._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._disabled = disabled if disabled is not None else None
|
||||
self._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})"
|
||||
|
||||
13
harvester.py
13
harvester.py
@@ -1,13 +1,18 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from devops import Organization, Project, DEVOPS_SCOPE
|
||||
from devops import Organization, Project, Repository, DEVOPS_SCOPE
|
||||
from azure.identity import DefaultAzureCredential
|
||||
from json import dumps
|
||||
|
||||
org = Organization("https://dev.azure.com/mcovsandbox", DefaultAzureCredential().get_token(DEVOPS_SCOPE).token)
|
||||
projects = org.projects
|
||||
print([str(p) for p in projects])
|
||||
|
||||
ado_sandbox = Project(org, id="bafe0cf1-6c97-4088-864a-ea6dc02b2727")
|
||||
# print([str(repo) for repo in ado_sandbox.repositories])
|
||||
|
||||
print(ado_sandbox)
|
||||
# print(str(ado_sandbox.repositories[0]))
|
||||
|
||||
repo = Repository(ado_sandbox, id_or_name="feac266f-84d2-41bc-839b-736925a85eaa")
|
||||
# repo = Repository(ado_sandbox, id_or_name="ado-auth-lab")
|
||||
|
||||
print(str(repo))
|
||||
print(f"Repository name: {repo.name} and URL: {repo.web_url}")
|
||||
Reference in New Issue
Block a user