Reegineered entity reading logic.

devops.py

@@ -20,7 +20,7 @@ def auto_properties(mapping: dict[str,str] | None = None):
                 pass
 
             # Fetch repository details from the API if it is set to None or not existing
-            self._get(getattr(self, "_id"))
+            self._get_attributes()
             return getattr(self, private_var)
 
         return property(fget=getter)
@@ -55,12 +55,12 @@ def auto_properties(mapping: dict[str,str] | None = None):
 class DevOps():
     """Base class for DevOps entities."""
 
-    def __init__(self):
-        self._org_url = None
-        self._token = None
-        self._api_version = None
+    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 = {}) -> requests.Response:
+    def _get_url_path(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.")
 
@@ -76,29 +76,38 @@ class DevOps():
         r.raise_for_status() # Ensure we raise an error for bad responses
         return r
 
+    def _get_entity_attributes(self, key_name: str, get_url: str, params: dict = {}) -> object:
+        """
+        Each entity class can use this method to populate its attributes, by defining
+        its own _get_attributes method that calls this one with the key name,
+        and the URL.
+        """
+        r = self._get_url_path(get_url, params=params) # Fetch the object data from the URL
+        setattr(self, f"_{key_name}", r.json().get(key_name, None)) # Set the key attribute
+        self.from_json(r.json()) # Populate other attributes from JSON
+
+    def _entity(self, entity_class: type, key_name: str, entity_data: dict) -> object:
+        """A generic method to create an entity from JSON data."""
+        args = { key_name: entity_data.get(key_name) }
+        e = entity_class(self, **args)
+        e.from_json(entity_data)
+        return e
+
     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)
+        r = self._get_url_path(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)
+            entities_list.append(self._entity(entity_class, key_name, entity))
 
         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
+        super().__init__(org_url, token, 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):
         return self._entities(
@@ -112,14 +121,14 @@ class Organization(DevOps):
     "description": "description"
     })
 class Project(DevOps):
-    def _get(self, id: str):
-        r = self.get(f"_apis/projects/{id}")
-        self.from_json(r.json())
+    def _get_attributes(self):
+        self._get_entity_attributes(
+            key_name="id",
+            get_url=f"_apis/projects/{self._id}"
+        )
 
-    def __init__(self, parent: Organization, id: str, **kwargs):
-        self._org_url = parent._org_url
-        self._token = parent._token
-        self._api_version = parent._api_version
+    def __init__(self, org: Organization, id: str, **kwargs):
+        super().__init__(org._org_url, org._token, org._api_version)
 
         try:
             self._id = str(UUID(id))
@@ -131,9 +140,6 @@ class Project(DevOps):
     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
@@ -156,28 +162,22 @@ class Project(DevOps):
     "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 _get_attributes(self):
+        self._get_entity_attributes(
+            key_name="id",
+            get_url=f"{self._project.id}/_apis/git/repositories/{self._id}"
+        )
 
-    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
+    def __init__(self, project: Project, id: str, **kwargs):
+        super().__init__(project._org_url, project._token, project._api_version)
+        self._project = project
+        self._id = id
 
         try:
-            self._id = str(UUID(id_or_name))
+            UUID(id) # Check if it's a valid UUID
         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
+            # Called with a repository name, fetch by name
+            self._get_attributes()
             if kwargs:
                 raise ValueError("Automatic properties cannot be set when retrieving by name.")
             return
@@ -185,16 +185,13 @@ class Repository(DevOps):
         # set other properties if provided
         self.set_auto_properties(**kwargs)
 
+    @property
+    def id(self):
+        return self._id
+
     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
@@ -216,18 +213,9 @@ class Repository(DevOps):
     "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):
-
+        super().__init__(repository._org_url, repository._token, repository._api_version)
         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
 

harvester.py

@@ -9,15 +9,19 @@ org = Organization("https://dev.azure.com/mcovsandbox", DefaultAzureCredential()
 # Listing projects in the organization
 print("Projects in the organization:")
 for project in org.projects:
-    print(f"- {project.name} (ID: {project._id})")
+    print(f"- {project.name} (ID: {project.id})")
 
 ado_sandbox = Project(org, id="bafe0cf1-6c97-4088-864a-ea6dc02b2727")
 
-repo = Repository(ado_sandbox, id_or_name="ado-auth-lab")
+print(f"Listing repositories in project: {ado_sandbox.name}")
+for repo in ado_sandbox.repositories:
+    print(f"- {repo.name} (ID: {repo.id})")
+
+repo = Repository(ado_sandbox, id="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")
+repo = Repository(ado_sandbox, id="feac266f-84d2-41bc-839b-736925a85eaa")
 print(str(repo))
 print(f"Repository name: {repo.name} and URL: {repo.web_url}")