für einen Hobby-Projekt habe ich mir vorgenommen einen Client für die Azure DevOps Rest Api zu schreiben (jedenfalls soweit, dass man die wichtigsten Funktionen verwenden kann)
Es gibt bereits einen OpenSource Client von Microsoft, welcher auf dem msrest package - auch von Microsoft - basiert:
https://github.com/microsoft/azure-devops-python-api
https://github.com/Azure/msrest-for-python
Erstens gefiel mir die Umsetzung von Microsoft aber nicht wirklich und zweitens habe ich Zweifel, wie "async" das Ganze ist.
Das heißt natürlich bei weitem nicht, dass ich es besser könnte, ... aber wenn man es selbst mal versucht, kann man dabei ja was lernen...
Ein paar Anforderungen:
- Komplett async
- Der Client soll sich stark am Aufbau der API orientieren, mit der recht guten Dokumentation von Microsoft sollte die Benutzung des Clients dann ziemlich intuitiv sein.
- Die API - Dokumentation weicht teilweise von der Realität ab und die Anzahl der Parameter ist groß. Durch Type-Annotations soll die Benutzung vereinfacht werden.
- Alle empfangenen JSON Objekte sollen in Klassen deserialisiert werden.
Aktuell kann man den Aufbau schon erkennen.
Man kann Projekte anlegen, löschen und auflisten.
Das Hinzufügen weiterer Funktionen sollte eigentlich nur noch (viel) Fleißarbeit sein.
Ich frage mich aber ob man das Konzept noch verbessern kann. An einigen Stellen fand ich meine Umsetzung etwas umständlich. Bisher habe ich aber nichts besseres gefunden.
Für Feedback wäre ich sehr dankbar! Es ist etwas lang, daher schonmal Danke für's lesen!
(Dass sich alles in einem Modul befindet liegt nur daran, dass es leichter zu posten ist)
Code: Alles auswählen
"""
Async Client for the Azure DevOps Rest Api
Documentation:
https://docs.microsoft.com/en-us/rest/api/azure/devops/core/projects?view=azure-devops-server-rest-5.0
"""
import asyncio
import time
from functools import lru_cache
from dataclasses import dataclass, field
from enum import Enum
from types import TracebackType
from typing import List, Optional, Type
import aiohttp
from aiohttp.helpers import BasicAuth
USER = "roger"
PAT = "izgaygo5xldxwoms2xy447lqy7trhtauqlprsi7k6y3u4uyptd2a"
INSTANCE = "roger-pc"
COLLECTION = "DefaultCollection"
STANDRAD_CAPABILITIES = {
"versioncontrol": {"sourceControlType": "Git"},
"processTemplate": {"templateTypeId": "6b724908-ef14-45cf-84f8-768b5384da45"},
}
def snakify_str(string: str) -> str:
if string.startswith("_"):
string = string[1:]
new_letters = []
for letter in string:
if letter.isupper():
new_letters.append(f"_{letter.lower()}")
else:
new_letters.append(letter)
return "".join(new_letters)
def snakify_args(arguments: dict) -> dict:
"""
since the api returns fields in lower-camel-case
they need to be converted to pythonic snake-case names
"""
return {snakify_str(key): value for key, value in arguments.items()}
class ProjectState(Enum):
all = "all"
create_pending = "createPending"
deleted = "deleted"
deleting = "deleting"
new = "new"
unchanged = "unchanged"
well_formed = "wellFormed"
class ProjectVisibility(Enum):
private = "private"
public = "public"
class OperationStatus(Enum):
cancelled = "cancelled"
failed = "failed"
in_progress = "inProgress"
not_set = "notSet"
queued = "queued"
succeeded = "succeeded"
@dataclass
class TeamProjectReference:
id: str
name: str
url: str
state: ProjectState
revision: int
visibility: ProjectVisibility
last_update_time: str
description: str = field(default=None)
@dataclass
class ReferenceLinks:
links: dict
@dataclass
class WebApiTeamRef:
id: str
name: str
url: str
@dataclass
class TeamProject:
id: str
name: str
url: str
state: ProjectState
revision: int
links: ReferenceLinks
visibility: ProjectVisibility
default_team: WebApiTeamRef
last_update_time: str
description: str
capabilities: dict = field(default=None)
@dataclass
class OperationReference:
id: str
status: OperationStatus
url: str
class BaseResource:
"""Baseclass for all Services and Resources"""
def __init__(self) -> None:
self._session = BaseResource.get_session()
self._base_url = f"http://{INSTANCE}/{COLLECTION}/_apis"
@staticmethod
@lru_cache
def get_session() -> aiohttp.ClientSession:
return aiohttp.ClientSession(auth=BasicAuth(USER, PAT))
async def __aenter__(self) -> "BaseResource":
return self._session
async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType],
) -> None:
await self._session.close()
async def await_status(self, url: str, status: OperationStatus, timeout: int = None) -> None:
start_time = time.time()
while True:
response = await self._session.get(url)
content = await response.json()
if content["status"] == status.value:
return
if timeout and time.time() - start_time > timeout:
print("Timeout")
return
await asyncio.sleep(1)
class Projects(BaseResource):
def __init__(self) -> None:
super().__init__()
self._resource_url = f"{self._base_url}/projects"
async def list(
self,
skip: int = None,
top: int = None,
continuation_token: str = None,
get_default_team_image_url: bool = None,
state_filter: ProjectState = None,
) -> List[TeamProjectReference]:
url = self._resource_url
params = {
"$skip": skip,
"$top": top,
"continuationToken": continuation_token,
"getDefaultTeamImageUrl": get_default_team_image_url,
"stateFilter": state_filter,
"api_version": "5.0",
}
params = {key: value for key, value in params.items() if value is not None}
async with self._session.get(url, params=params) as response:
content = await response.json()
return [TeamProjectReference(**snakify_args(element)) for element in content["value"]]
async def get(self, project_id: str, include_capabilities: bool = None, include_history: bool = None) -> TeamProject:
url = f"{self._resource_url}/{project_id}"
params = {"includeCapabilities": include_capabilities, "includeHistory": include_history}
async with self._session.get(url, params=params) as response:
content = await response.json()
return TeamProject(**snakify_args(content))
async def create(self, name: str, capabilities: dict, description: str = None, url: str = None) -> None:
url = self._resource_url
params = {"api-version": "5.0"}
payload = {"name": name, "description": description, "capabilities": capabilities}
async with self._session.post(url, params=params, json=payload) as response:
content = await response.json()
operation_reference = OperationReference(**content)
await self.await_status(operation_reference.url, OperationStatus.succeeded, timeout=30)
async def delete(self, project_id: str) -> None:
url = f"{self._resource_url}/{project_id}"
params = {"api-version": "5.0"}
async with self._session.delete(url, params=params) as response:
content = await response.json()
operation_reference = OperationReference(**content)
await self.await_status(operation_reference.url, OperationStatus.succeeded, timeout=30)
class Core(BaseResource):
def __init__(self) -> None:
super().__init__()
# Core resources
self.projects = Projects()
# self.processes = Processes()
# self.teams = Teams()
class AsyncClient:
"""Client class to give access to all services and resources"""
def __init__(self) -> None:
# services
# self.build = Build()
self.core = Core()
# self.dashboard = Dashboard()
# ...
async def main():
client = AsyncClient()
for project in await client.core.projects.list():
if project.name == "TestProject":
await client.core.projects.delete(project.id)
await client.core.projects.create(name="TestProject", capabilities=STANDRAD_CAPABILITIES, description="TestDescription")
for project in await client.core.projects.list():
if project.name == "TestProject":
await client.core.projects.delete(project.id)
asyncio.run(main())
SCNR.