Fixed auto_properties.
This commit is contained in:
1
.gitignore
vendored
1
.gitignore
vendored
@@ -10,6 +10,7 @@ prototype_*.py
|
|||||||
|
|
||||||
# Shell secrets
|
# Shell secrets
|
||||||
*.secret
|
*.secret
|
||||||
|
*.client_secret
|
||||||
|
|
||||||
# Certificate files
|
# Certificate files
|
||||||
*.pem
|
*.pem
|
||||||
|
|||||||
@@ -8,4 +8,6 @@ token = get_token(
|
|||||||
pem_path="./slawek-mba.pem"
|
pem_path="./slawek-mba.pem"
|
||||||
)
|
)
|
||||||
|
|
||||||
print(f"Obtained token: {token}")
|
# print(f"Obtained token: {token}")
|
||||||
|
|
||||||
|
print(f"export ADO_TOKEN='{token}'")
|
||||||
91
sk/devops.py
91
sk/devops.py
@@ -3,6 +3,7 @@ import requests
|
|||||||
import urllib.parse
|
import urllib.parse
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
import logging
|
import logging
|
||||||
|
from sk.logger import log_entity_creation
|
||||||
|
|
||||||
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"
|
||||||
@@ -27,6 +28,7 @@ def auto_properties(mapping: dict[str,str]):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
# Fetch repository details from the API if it is set to None or not existing
|
# Fetch repository details from the API if it is set to None or not existing
|
||||||
|
log.debug(f"Auto-fetching property '{name}' for {self.__class__.__name__}", extra={"property_name": name})
|
||||||
self.get_auto_properties()
|
self.get_auto_properties()
|
||||||
return getattr(self, private_var)
|
return getattr(self, private_var)
|
||||||
|
|
||||||
@@ -34,15 +36,16 @@ def auto_properties(mapping: dict[str,str]):
|
|||||||
|
|
||||||
def from_args(self, **kwargs):
|
def from_args(self, **kwargs):
|
||||||
for name in kwargs:
|
for name in kwargs:
|
||||||
if name in self.__class__.__auto_properties__:
|
if name in self.__class__.__auto_properties__.values():
|
||||||
setattr(self, f"_{name}", kwargs.get(name, None))
|
setattr(self, f"_{name}", kwargs.get(name, None))
|
||||||
|
|
||||||
def from_json(self, json_data: dict):
|
def from_json(self, json_data: dict):
|
||||||
for name in self.__class__.__auto_properties__:
|
for json_name in self.__class__.__auto_properties_reversed__:
|
||||||
setattr(self, f"_{name}", json_data.get(self.__class__.__auto_properties__[name], None))
|
setattr(self, f"_{self.__class__.__auto_properties_reversed__[json_name]}", json_data.get(json_name, None))
|
||||||
|
|
||||||
def decorator(cls):
|
def decorator(cls):
|
||||||
cls.__auto_properties__ = mapping # Make a copy of the mapping
|
cls.__auto_properties__ = mapping # Make a copy of the mapping
|
||||||
|
cls.__auto_properties_reversed__ = {v: k for k, v in mapping.items()} # Store reversed mapping for JSON parsing
|
||||||
|
|
||||||
# Create properties dynamically
|
# Create properties dynamically
|
||||||
for name in mapping:
|
for name in mapping:
|
||||||
@@ -75,7 +78,7 @@ class Organization():
|
|||||||
self._org_url = org_url.rstrip("/") + "/" # Ensure trailing slash
|
self._org_url = org_url.rstrip("/") + "/" # Ensure trailing slash
|
||||||
self._token = token
|
self._token = token
|
||||||
self._api_version = api_version
|
self._api_version = api_version
|
||||||
log.debug(f"Created new Organization object", extra={"entity_class": "Organization"})
|
log_entity_creation(log, Organization, self._org_url)
|
||||||
|
|
||||||
def get_path(self, path: str, params: dict = {}) -> dict:
|
def get_path(self, path: str, params: dict = {}) -> dict:
|
||||||
return get_url(
|
return get_url(
|
||||||
@@ -120,7 +123,7 @@ class Project():
|
|||||||
except ValueError:
|
except ValueError:
|
||||||
raise ValueError(f"Invalid project ID: {self._id}")
|
raise ValueError(f"Invalid project ID: {self._id}")
|
||||||
|
|
||||||
log.debug(f"Created new Project object", extra={"entity_class": "Project"})
|
log_entity_creation(log, Project, self.id)
|
||||||
|
|
||||||
def get_auto_properties(self):
|
def get_auto_properties(self):
|
||||||
r = get_url(
|
r = get_url(
|
||||||
@@ -179,7 +182,7 @@ class Repository():
|
|||||||
|
|
||||||
# set other properties if provided
|
# set other properties if provided
|
||||||
self.from_args(**kwargs)
|
self.from_args(**kwargs)
|
||||||
log.debug(f"Created new {self.__class__.__name__} object", extra={"entity_class": self.__class__.__name__})
|
log_entity_creation(log, Repository, self.id)
|
||||||
|
|
||||||
def get_auto_properties(self):
|
def get_auto_properties(self):
|
||||||
id = self._id if hasattr(self, "_id") else self._name
|
id = self._id if hasattr(self, "_id") else self._name
|
||||||
@@ -202,16 +205,21 @@ class Repository():
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def items(self):
|
def items(self):
|
||||||
|
log.debug(f"Fetching items for repository '{self.name}'", extra={"repository_name": self.name})
|
||||||
if not hasattr(self, "_items"):
|
if not hasattr(self, "_items"):
|
||||||
self._items = self._entities(
|
objects = self._project.organization.get_path(
|
||||||
entity_class=Item,
|
path=f"{self._project.id}/_apis/git/repositories/{self.id}/items",
|
||||||
key_name="path",
|
|
||||||
list_url=f"{self._project.id}/_apis/git/repositories/{self._id}/items",
|
|
||||||
params={
|
params={
|
||||||
"scopePath": "/",
|
"scopePath": "/",
|
||||||
"recursionLevel": "oneLevel"
|
"recursionLevel": "oneLevel"
|
||||||
}
|
}
|
||||||
)
|
).get("value", [])
|
||||||
|
self._items = []
|
||||||
|
for obj in objects:
|
||||||
|
i = Item(repository=self, path=obj.get("path"))
|
||||||
|
i.from_json(obj)
|
||||||
|
self._items.append(i)
|
||||||
|
|
||||||
return self._items
|
return self._items
|
||||||
|
|
||||||
def __getitem__(self, key: str) -> Item:
|
def __getitem__(self, key: str) -> Item:
|
||||||
@@ -221,6 +229,7 @@ class Repository():
|
|||||||
raise KeyError(f"Item with path '{key}' not found.")
|
raise KeyError(f"Item with path '{key}' not found.")
|
||||||
|
|
||||||
@auto_properties({
|
@auto_properties({
|
||||||
|
"path": "path",
|
||||||
"object_id": "objectId",
|
"object_id": "objectId",
|
||||||
"git_object_type": "gitObjectType",
|
"git_object_type": "gitObjectType",
|
||||||
"commit_id": "commitId",
|
"commit_id": "commitId",
|
||||||
@@ -228,43 +237,41 @@ class Repository():
|
|||||||
"url": "url"
|
"url": "url"
|
||||||
})
|
})
|
||||||
class Item():
|
class Item():
|
||||||
def __init__(self, repository: Repository, path: str, **kwargs):
|
def __init__(self, repository: Repository, **kwargs):
|
||||||
self._repository = repository
|
self._repository = repository
|
||||||
self._path = path
|
self.from_args(**kwargs)
|
||||||
self.set_auto_properties(**kwargs) # set properties defined in decorator
|
log_entity_creation(log, Item, self.path)
|
||||||
log.debug(f"Created new {self.__class__.__name__} object", extra={"entity_class": self.__class__.__name__})
|
|
||||||
|
|
||||||
def _get(self):
|
def get_auto_properties(self):
|
||||||
pass
|
r = self._repository._project.organization.get_path(
|
||||||
# self._get_entity(
|
path=f"{self._repository._project.id}/_apis/git/repositories/{self._repository.id}/items",
|
||||||
# key_name="path",
|
params={
|
||||||
# get_url=f"{self._repository._project.id}/_apis/git/repositories/{self._repository.id}/items",
|
"path": self.path,
|
||||||
# params={
|
"$format": "json",
|
||||||
# "path": self._path,
|
"recursionLevel": "none"
|
||||||
# "$format": "json",
|
}
|
||||||
# "recursionLevel": "none"
|
)
|
||||||
# }
|
self.from_json(r)
|
||||||
# )
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def path(self):
|
def path(self):
|
||||||
return self._path
|
return self._path
|
||||||
|
|
||||||
@property
|
# @property
|
||||||
def children(self):
|
# def children(self):
|
||||||
"""List items under this item if it is a folder."""
|
# """List items under this item if it is a folder."""
|
||||||
if self.git_object_type == "tree":
|
# if self.git_object_type == "tree":
|
||||||
return self._entities(
|
# return self._entities(
|
||||||
entity_class=Item,
|
# entity_class=Item,
|
||||||
key_name="path",
|
# key_name="path",
|
||||||
list_url=f"{self._repository._project.id}/_apis/git/repositories/{self._repository.id}/items",
|
# list_url=f"{self._repository._project.id}/_apis/git/repositories/{self._repository.id}/items",
|
||||||
params={
|
# params={
|
||||||
"scopePath": self._path,
|
# "scopePath": self._path,
|
||||||
"recursionLevel": "oneLevel"
|
# "recursionLevel": "oneLevel"
|
||||||
}
|
# }
|
||||||
)
|
# )
|
||||||
else:
|
# else:
|
||||||
raise ValueError("Items can only be listed for folder items.")
|
# raise ValueError("Items can only be listed for folder items.")
|
||||||
|
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
return f"Item(path=\"{self._path}\")"
|
return f"Item(path=\"{self._path}\" type={self.git_object_type} commit_id={self.commit_id})"
|
||||||
|
|||||||
@@ -35,3 +35,10 @@ def setup(name: str, handlers: int) -> logging.Logger:
|
|||||||
logger.addHandler(custom_handler)
|
logger.addHandler(custom_handler)
|
||||||
|
|
||||||
return logger
|
return logger
|
||||||
|
|
||||||
|
def log_entity_creation(logger: logging.Logger, entity_class: type, entity_key: str):
|
||||||
|
logger.debug(f'Created new "{entity_class.__name__}" object with key: "{entity_key}"',
|
||||||
|
extra={
|
||||||
|
"entity_class": entity_class.__name__,
|
||||||
|
"entity_key": entity_key
|
||||||
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user