Compare commits
9 Commits
850bc99aa2
...
26b192c840
| Author | SHA1 | Date | |
|---|---|---|---|
| 26b192c840 | |||
| 71d83b8d76 | |||
| 837543d44a | |||
| 91b806637c | |||
| 96ccd4d7c7 | |||
| deed075727 | |||
| ce38e275a9 | |||
| 4685103e60 | |||
| 3ce14912e4 |
3
.gitignore
vendored
3
.gitignore
vendored
@@ -1,3 +1,6 @@
|
|||||||
# Python
|
# Python
|
||||||
.venv
|
.venv
|
||||||
__pycache__/
|
__pycache__/
|
||||||
|
|
||||||
|
# Ignore sample JSON files
|
||||||
|
*.sample.json
|
||||||
|
|||||||
248
devops.py
248
devops.py
@@ -1,20 +1,69 @@
|
|||||||
from unicodedata import name
|
|
||||||
import requests
|
import requests
|
||||||
import urllib.request
|
|
||||||
import urllib.parse
|
import urllib.parse
|
||||||
import json
|
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
|
|
||||||
DEVOPS_SCOPE = "https://app.vssps.visualstudio.com/.default"
|
DEVOPS_SCOPE = "https://app.vssps.visualstudio.com/.default"
|
||||||
DEVOPS_API_VERSION = "7.1"
|
DEVOPS_API_VERSION = "7.1"
|
||||||
|
|
||||||
class Organization:
|
# Define a class decorator
|
||||||
def __init__(self, org_url: str, token: str, api_version: str = DEVOPS_API_VERSION):
|
def auto_properties(mapping: dict[str,str] | None = None):
|
||||||
self._org_url = org_url.rstrip("/") + "/" # Ensure trailing slash
|
|
||||||
self._token = token
|
def make_property(name: str):
|
||||||
self._api_version = api_version
|
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(getattr(self, "_id"))
|
||||||
|
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):
|
||||||
|
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 = {
|
request_parameters = {
|
||||||
"api-version": self._api_version,
|
"api-version": self._api_version,
|
||||||
**params
|
**params
|
||||||
@@ -24,51 +73,164 @@ class Organization:
|
|||||||
r = requests.get(url=url, params=request_parameters, headers={
|
r = requests.get(url=url, params=request_parameters, headers={
|
||||||
"Authorization": f"Bearer {self._token}"
|
"Authorization": f"Bearer {self._token}"
|
||||||
})
|
})
|
||||||
|
r.raise_for_status() # Ensure we raise an error for bad responses
|
||||||
return r
|
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
|
@property
|
||||||
def projects(self):
|
def projects(self):
|
||||||
r = self.get("_apis/projects")
|
return self._entities(
|
||||||
r.raise_for_status()
|
entity_class=Project,
|
||||||
|
key_name="id",
|
||||||
|
list_url="_apis/projects")
|
||||||
|
|
||||||
# Return a list of Project instances
|
@auto_properties({
|
||||||
projects_data = r.json().get("value", [])
|
"name": "name",
|
||||||
return [
|
"url": "url",
|
||||||
Project(self,
|
"description": "description"
|
||||||
id=proj.get("id"),
|
})
|
||||||
name=proj.get("name"),
|
class Project(DevOps):
|
||||||
url=proj.get("url"),
|
def _get(self, id: str):
|
||||||
description=proj.get("description"),
|
r = self.get(f"_apis/projects/{id}")
|
||||||
) for proj in projects_data
|
self.from_json(r.json())
|
||||||
]
|
|
||||||
|
|
||||||
class Project:
|
def __init__(self, parent: Organization, id: str, **kwargs):
|
||||||
def __init__(self, org: Organization, id: str, name: str | None = None, url: str | None = None, description: str | None = None):
|
self._org_url = parent._org_url
|
||||||
self._org = org
|
self._token = parent._token
|
||||||
|
self._api_version = parent._api_version
|
||||||
|
|
||||||
# Check, if the id is a valid UUID
|
|
||||||
try:
|
try:
|
||||||
self._id = str(UUID(id))
|
self._id = str(UUID(id))
|
||||||
except ValueError:
|
except ValueError:
|
||||||
raise ValueError(f"Invalid project ID: {self._id}")
|
raise ValueError(f"Invalid project ID: {id}")
|
||||||
|
|
||||||
if name is not None and url is not None and name != "" and url != "":
|
self.set_auto_properties(**kwargs)
|
||||||
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/{urllib.parse.quote(self._id)}")
|
|
||||||
r.raise_for_status()
|
|
||||||
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", "")
|
|
||||||
|
|
||||||
@property
|
|
||||||
def name(self):
|
|
||||||
return self._name
|
|
||||||
|
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
return f"Project(name={self._name}, id={self._id})"
|
return f"Project(name={self._name}, id={self._id})"
|
||||||
|
|
||||||
|
def get_path(self, path: str, params: dict = {}):
|
||||||
|
return self.get(f"{self._id}/{path.lstrip('/')}", params=params)
|
||||||
|
|
||||||
|
@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, repo_name: str):
|
||||||
|
r = self._project.get_path(f"_apis/git/repositories/{urllib.parse.quote(repo_name)}")
|
||||||
|
self._id = r.json().get("id", None)
|
||||||
|
self.from_json(r.json())
|
||||||
|
|
||||||
|
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
|
||||||
|
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
|
||||||
|
if kwargs:
|
||||||
|
raise ValueError("Automatic properties cannot be set when retrieving by name.")
|
||||||
|
return
|
||||||
|
|
||||||
|
# set other properties if provided
|
||||||
|
self.set_auto_properties(**kwargs)
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return f"Repository(name={self.name}, id={self._id})"
|
||||||
|
|
||||||
|
def get_path(self, path: str, params: dict = {}):
|
||||||
|
return self._project.get_path(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
|
||||||
|
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 _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, **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
|
||||||
|
|
||||||
|
@property
|
||||||
|
def path(self):
|
||||||
|
return self._path
|
||||||
22
harvester.py
22
harvester.py
@@ -1,13 +1,27 @@
|
|||||||
#!/usr/bin/env python3
|
#!/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 azure.identity import DefaultAzureCredential
|
||||||
from json import dumps
|
from json import dumps
|
||||||
|
|
||||||
org = Organization("https://dev.azure.com/mcovsandbox", DefaultAzureCredential().get_token(DEVOPS_SCOPE).token)
|
org = Organization("https://dev.azure.com/mcovsandbox", DefaultAzureCredential().get_token(DEVOPS_SCOPE).token)
|
||||||
projects = org.projects
|
|
||||||
print([str(p) for p in projects])
|
# Listing projects in the organization
|
||||||
|
print("Projects in the organization:")
|
||||||
|
for project in org.projects:
|
||||||
|
print(f"- {project.name} (ID: {project._id})")
|
||||||
|
|
||||||
ado_sandbox = Project(org, id="bafe0cf1-6c97-4088-864a-ea6dc02b2727")
|
ado_sandbox = Project(org, id="bafe0cf1-6c97-4088-864a-ea6dc02b2727")
|
||||||
|
|
||||||
print(ado_sandbox)
|
repo = Repository(ado_sandbox, id_or_name="ado-auth-lab")
|
||||||
|
print(str(repo))
|
||||||
|
print(f"Repository name: {repo.name} and URL: {repo.web_url}")
|
||||||
|
|
||||||
|
repo = Repository(ado_sandbox, id_or_name="feac266f-84d2-41bc-839b-736925a85eaa")
|
||||||
|
print(str(repo))
|
||||||
|
print(f"Repository name: {repo.name} and URL: {repo.web_url}")
|
||||||
|
|
||||||
|
print("Listing items in the repository:")
|
||||||
|
|
||||||
|
for item in repo.items:
|
||||||
|
print(f"{item.path} ({item.git_object_type}): {item.commit_id}")
|
||||||
|
|||||||
Reference in New Issue
Block a user