docs-harvester/devops.py

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(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 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", [])

        project_list = []
        for proj in projects_data:
            p = Project(self, project_id=proj.get("id"))
            p.from_json(proj)
            project_list.append(p)

        return project_list

@auto_properties({
    "name":        "name",
    "url":         "url",
    "description": "description"
    })
class Project():
    def _get(self, project_id: str):
        r = self._org.get(f"_apis/projects/{project_id}")
        self.from_json(r.json())

    def __init__(self, org: Organization, project_id: str, **kwargs):
        self._org = org
        try:
            self._id = str(UUID(project_id))
        except ValueError:
            raise ValueError(f"Invalid project ID: {project_id}")

        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._org.get(f"{self._id}/{path.lstrip('/')}", params=params)

    @property
    def repositories(self):
        r = self.get_path(f"_apis/git/repositories")

        repos_data = r.json().get("value", [])

        repository_list = []
        for repo in repos_data:
            r = Repository(self, id_or_name=repo.get("id"))
            r.from_json(repo)
            repository_list.append(r)

        return repository_list

@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:
    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

        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)
            # 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 a list of Item instances
        items_data = self._get_items().get("value", [])

        item_list = []
        for item in items_data:
            i = Item(self, path=item.get("path"))
            i.from_json(item)
            item_list.append(i)

        return item_list

@auto_properties({
    "object_id":       "objectId",
    "git_object_type": "gitObjectType",
    "commit_id":       "commitId",
    "is_folder":       "isFolder",
    "url":             "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, **kwargs):

        self._repository = repository
        self._path = path
        self.set_auto_properties(**kwargs) # set properties defined in decorator

    @property
    def path(self):
        return self._path