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
 | 
			
		||||
.venv
 | 
			
		||||
__pycache__/
 | 
			
		||||
 | 
			
		||||
# Ignore sample JSON files
 | 
			
		||||
*.sample.json
 | 
			
		||||
 
 | 
			
		||||
							
								
								
									
										248
									
								
								devops.py
									
									
									
									
									
								
							
							
						
						
									
										248
									
								
								devops.py
									
									
									
									
									
								
							@@ -1,20 +1,69 @@
 | 
			
		||||
from unicodedata import name
 | 
			
		||||
import requests
 | 
			
		||||
import urllib.request
 | 
			
		||||
import urllib.parse
 | 
			
		||||
import json
 | 
			
		||||
from uuid import UUID
 | 
			
		||||
 | 
			
		||||
DEVOPS_SCOPE = "https://app.vssps.visualstudio.com/.default"
 | 
			
		||||
DEVOPS_API_VERSION = "7.1"
 | 
			
		||||
 | 
			
		||||
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
 | 
			
		||||
# 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(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 = {
 | 
			
		||||
            "api-version": self._api_version,
 | 
			
		||||
            **params
 | 
			
		||||
@@ -24,51 +73,164 @@ class Organization:
 | 
			
		||||
        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 _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
 | 
			
		||||
    def projects(self):
 | 
			
		||||
        r = self.get("_apis/projects")
 | 
			
		||||
        r.raise_for_status()
 | 
			
		||||
        return self._entities(
 | 
			
		||||
            entity_class=Project,
 | 
			
		||||
            key_name="id",
 | 
			
		||||
            list_url="_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({
 | 
			
		||||
    "name":        "name",
 | 
			
		||||
    "url":         "url",
 | 
			
		||||
    "description": "description"
 | 
			
		||||
    })
 | 
			
		||||
class Project(DevOps):
 | 
			
		||||
    def _get(self, id: str):
 | 
			
		||||
        r = self.get(f"_apis/projects/{id}")
 | 
			
		||||
        self.from_json(r.json())
 | 
			
		||||
 | 
			
		||||
class Project:
 | 
			
		||||
    def __init__(self, org: Organization, id: str, name: str | None = None, url: str | None = None, description: str | None = None):
 | 
			
		||||
        self._org = org
 | 
			
		||||
    def __init__(self, parent: Organization, id: str, **kwargs):
 | 
			
		||||
        self._org_url = parent._org_url
 | 
			
		||||
        self._token = parent._token
 | 
			
		||||
        self._api_version = parent._api_version
 | 
			
		||||
 | 
			
		||||
        # Check, if the id is a valid UUID
 | 
			
		||||
        try:
 | 
			
		||||
            self._id = str(UUID(id))
 | 
			
		||||
        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._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
 | 
			
		||||
        self.set_auto_properties(**kwargs)
 | 
			
		||||
 | 
			
		||||
    def __str__(self):
 | 
			
		||||
        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
 | 
			
		||||
 | 
			
		||||
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])
 | 
			
		||||
 | 
			
		||||
# 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")
 | 
			
		||||
 | 
			
		||||
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