Add service hook subscriptions client

CHANGELOG.md

@@ -4,6 +4,16 @@
 
 All notable changes to this project will be documented in this file.
 
+## 0.2.0 - 2025-12-15
+
+### <!-- 0 -->🚀 Features
+
+* **docs**: Add a quick start
+* **servicehooks**: New client to interact with hook subscriptions
+  * Add support for creating subscriptions.
+  * Add support for listing subscriptions.
+  * Add support for deleting subscriptions.
+
 ## 0.1.1 - 2025-12-11
 
 ### <!-- 1 -->🐛 Bug Fixes

README.md

@@ -5,6 +5,7 @@ A simple Python interface to interact with Azure DevOps API.
 Contents:
 
 * [Installation](#installation)
+* [Quick start](#quick-start)
 * [Development](#development)
   * [Requirements](#requirements)
   * [Install tools](#install-tools)
@@ -19,6 +20,23 @@ With [uv](https://docs.astral.sh/uv/):
 uv add git+https://github.com/bigbrozer/ez-azure-devops.git
 ```
 
+## Quick start
+
+```python
+from ez_ados import AzureDevOps
+
+# Init a client for an organization
+my_org = AzureDevOps("https://dev.azure.com/myorg")
+
+# Authenticate using EntraID
+# See https://learn.microsoft.com/en-us/dotnet/azure/sdk/authentication/credential-chains?tabs=dac#defaultazurecredential-overview
+my_org.authenticate()
+
+# Get a project
+projects = my_org.projects_client()
+print(projects.get(name="my_project"))
+```
+
 ## Development
 
 ### Requirements

src/ez_ados/__init__.py

@@ -1 +1,5 @@
 """Main package."""
+
+from ez_ados.client import AzureDevOps
+
+__all__ = ["AzureDevOps"]

src/ez_ados/client.py

@@ -13,6 +13,7 @@
 from .pipelines.clients import PipelineClient
 from .policy.configurations.clients import PolicyConfigurationClient
 from .policy.types.clients import PolicyTypeClient
+from .servicehooks.subscriptions.clients import HookSubscriptionClient
 
 # Get a logger for this module
 logger = logging.getLogger(__name__)
@@ -114,3 +115,8 @@ def builds_client(self, project: str) -> BuildClient:
         """Return an HTTP client for interacting with Builds endpoint."""
         _endpoint = [self.org_url, project, "_apis", "build"]
         return BuildClient(self._build_client(endpoint="/".join(_endpoint)))
+
+    def hook_subscriptions_client(self) -> HookSubscriptionClient:
+        """Return an HTTP client for interacting with Service Hooks' subscriptions endpoint."""
+        _endpoint = [self.org_url, "_apis", "hooks", "subscriptions"]
+        return HookSubscriptionClient(self._build_client(endpoint="/".join(_endpoint)))

src/ez_ados/models.py

@@ -42,30 +42,26 @@
     PolicySettings,
     PolicyType,
 )
+from .servicehooks.subscriptions.models import HookSubscription, HookSubscriptionCollection, HookSubscriptionCreate
 
 __all__ = [
     "BuildDefinition",
+    "BuildDefinitionBase",
     "BuildDefinitionCollection",
     "BuildDefinitionCreate",
-    "BuildDefinitionBase",
     "BuildProcess",
     "BuildRepository",
-    "Project",
-    "Properties",
+    "CommentTypeEnum",
     "GitItem",
     "GitItemCollection",
     "GitItemDescriptor",
     "GitItemsBatch",
     "GitRef",
-    "GitRepository",
-    "CommentTypeEnum",
-    "ThreadStatusEnum",
-    "PullRequestThread",
-    "PullRequestThreadCreate",
-    "PullRequestThreadCollection",
-    "PullRequestThreadComment",
-    "PullRequestThreadCommentCreate",
     "GitRefCollection",
+    "GitRepository",
+    "HookSubscription",
+    "HookSubscriptionCollection",
+    "HookSubscriptionCreate",
     "Pipeline",
     "PipelineCollection",
     "PipelineConfiguration",
@@ -78,4 +74,12 @@
     "PolicyScope",
     "PolicySettings",
     "PolicyType",
+    "Project",
+    "Properties",
+    "PullRequestThread",
+    "PullRequestThreadCollection",
+    "PullRequestThreadComment",
+    "PullRequestThreadCommentCreate",
+    "PullRequestThreadCreate",
+    "ThreadStatusEnum",
 ]

src/ez_ados/servicehooks/__init__.py

@@ -0,0 +1 @@
+"""Service Hooks API module."""

src/ez_ados/servicehooks/subscriptions/__init__.py

@@ -0,0 +1 @@
+"""Service Hooks' Subscriptions API models."""

src/ez_ados/servicehooks/subscriptions/clients.py

@@ -0,0 +1,42 @@
+"""Module for Service Hooks' Subscriptions client API."""
+
+import logging
+
+from ...base.clients import Client
+from .models import HookSubscription, HookSubscriptionCollection, HookSubscriptionCreate
+
+# Get a logger for this module
+logger = logging.getLogger(__name__)
+
+
+class HookSubscriptionClient(Client):
+    """Represent a client to Git repository API in Azure DevOps."""
+
+    def get(self, id: str) -> HookSubscription:
+        """Fetch a single Service Hook's subscription."""
+        logger.info("Fetching details for hook subscription with id '%s'...", id)
+        resource = HookSubscription.model_validate(self._client.get(id).raise_for_status().json())
+        logger.debug(resource)
+        return resource
+
+    def list(self) -> HookSubscriptionCollection:
+        """List all Service Hook's available in organization."""
+        return HookSubscriptionCollection.model_validate(self._client.get("").raise_for_status().json())
+
+    def delete(self, id: str) -> None:
+        """Fetch a single Service Hook's subscription."""
+        logger.info("Deleting service hook subscription with id=%s...", id)
+        self._client.delete(id).raise_for_status()
+
+    def create(self, definition: HookSubscriptionCreate) -> HookSubscription:
+        """Create a new Service Hook's subscription."""
+        resource = HookSubscriptionCreate.model_validate(definition)
+        logger.debug("Create new hook subscription: %s", definition)
+
+        post_request = self._client.post("", json=resource.model_dump(exclude_none=True))
+        post_request.raise_for_status()
+        new_resource = post_request.json()
+        logger.debug("Create new hook subscription status code: %s", post_request.status_code)
+        logger.debug("Create new hook subscription response: %s", post_request.content)
+
+        return HookSubscription.model_validate(new_resource)

src/ez_ados/servicehooks/subscriptions/enums.py

@@ -0,0 +1,17 @@
+"""Enumerations for Service Hooks' Subscriptions types."""
+
+from ...base.enums import StrIntEnum
+
+
+class SubscriptionStatus(StrIntEnum):
+    """
+    Enum for available hook subscriptions states types.
+
+    Reference: https://learn.microsoft.com/en-us/javascript/api/azure-devops-extension-api/subscriptionstatus
+    """
+
+    enabled = 0
+    onProbation = 10
+    disabledByUser = 20
+    disabledBySystem = 30
+    disabledByInactiveIdentity = 40

src/ez_ados/servicehooks/subscriptions/models.py

@@ -0,0 +1,70 @@
+"""Module for Service Hooks' Subscriptions models."""
+
+from typing import Annotated, Any, Self
+
+from pydantic import BeforeValidator, Field
+
+from ...base.models import BaseCollection, JSONModel
+from .enums import SubscriptionStatus
+
+
+class HookSubscription(JSONModel):
+    """Represents a hook subscription."""
+
+    id: str
+    status: Annotated[SubscriptionStatus, BeforeValidator(SubscriptionStatus.validate)]
+    publisher_id: Annotated[str, Field(alias="publisherId")]
+    event_type: Annotated[str, Field(alias="eventType")]
+    resource_version: Annotated[str, Field(alias="resourceVersion")]
+    event_description: Annotated[str, Field(alias="eventDescription")]
+    consumer_id: Annotated[str, Field(alias="consumerId")]
+    consumer_action_id: Annotated[str, Field(alias="consumerActionId")]
+    action_description: Annotated[str, Field(alias="actionDescription")]
+    publisher_inputs: Annotated[dict[str, Any], Field(alias="publisherInputs")]
+    consumer_inputs: Annotated[dict[str, Any], Field(alias="consumerInputs")]
+
+
+class HookSubscriptionCreate(JSONModel):
+    """Creates a new hook subscription."""
+
+    publisher_id: Annotated[str, Field(alias="publisherId")]
+    event_type: Annotated[str, Field(alias="eventType")]
+    resource_version: Annotated[str, Field(alias="resourceVersion")]
+    consumer_id: Annotated[str, Field(alias="consumerId")]
+    consumer_action_id: Annotated[str, Field(alias="consumerActionId")]
+    publisher_inputs: Annotated[dict[str, Any], Field(alias="publisherInputs")]
+    consumer_inputs: Annotated[dict[str, Any], Field(alias="consumerInputs")]
+
+
+class HookSubscriptionCollection(BaseCollection[HookSubscription]):
+    """Represents a collection of hook subscriptions."""
+
+    def for_event(self, event_type: str) -> Self:
+        """Filter hook subscription for a type of event."""
+
+        def _filter(p: HookSubscription) -> bool:
+            return p.event_type == event_type
+
+        filtered_results = list(filter(_filter, self))
+        return type(self)(count=len(filtered_results), value=filtered_results)
+
+    def for_git_push_event(
+        self, branch_name: str | None = None, project_id: str | None = None, repository_id: str | None = None
+    ) -> Self:
+        """Filter hook subscription for a git.push event."""
+        _filtered_events = self.for_event("git.push")
+
+        def _filter(p: HookSubscription) -> bool:
+            if p.event_type == "git.push":
+                if branch_name is not None and p.publisher_inputs["branch"] != branch_name:
+                    return False
+                if project_id is not None and p.publisher_inputs["projectId"] != project_id:
+                    return False
+                if repository_id is not None and p.publisher_inputs["repository"] != repository_id:
+                    return False
+                return True
+            else:
+                return False
+
+        filtered_results = list(filter(_filter, _filtered_events))
+        return type(self)(count=len(filtered_results), value=filtered_results)

src/ez_ados/version.py

@@ -1,4 +1,4 @@
 """Program version."""
 
 # Will be automatically updated in CI.
-__version__ = "0.1.1"
+__version__ = "0.2.0"

tests/test_servicehooks.py

@@ -0,0 +1,56 @@
+"""Test Service Hooks Models."""
+
+import pydantic
+import pytest
+
+from ez_ados.models import HookSubscription, HookSubscriptionCollection, HookSubscriptionCreate
+
+
+def test_hook_subscription_collection():
+    """Test a collection of Service Hooks' Subscription resource."""
+    spec = {
+        "id": "123",
+        "status": "enabled",
+        "publisher_id": "tfs",
+        "event_type": "git.push",
+        "resource_version": "1.0-preview.1",
+        "event_description": "Resource was updated",
+        "consumer_id": "azureStorageQueue",
+        "consumer_action_id": "enqueue",
+        "action_description": "Consumer performed action",
+        "publisher_inputs": {"branch": "test", "projectId": "1234"},
+        "consumer_inputs": {"accountName": "stafoobar001"},
+    }
+    resources = [
+        HookSubscription.model_validate(spec),
+        HookSubscription.model_validate(spec | {"event_type": "git.repo.deleted", "status": "disabledBySystem"}),
+    ]
+    assert resources[0].status.value == 0  # noqa: PLR2004
+    assert resources[1].status.value == 30  # noqa: PLR2004
+
+    try:
+        results = HookSubscriptionCollection.model_validate({"count": len(resources), "value": resources})
+    except pydantic.ValidationError as exc:
+        pytest.fail(reason=str(exc))
+
+    with pytest.raises(pydantic.ValidationError):
+        HookSubscription(id=1)
+
+    filtered = results.for_git_push_event()
+    assert filtered.count == 1  # noqa: PLR2004
+
+
+def test_hook_subscription_create():
+    """Test creation of a Service Hooks' Subscription resource."""
+    spec = {
+        "publisher_id": "tfs",
+        "event_type": "git.push",
+        "resource_version": "1.0-preview.1",
+        "consumer_id": "azureStorageQueue",
+        "consumer_action_id": "enqueue",
+        "publisher_inputs": {"branch": "test", "projectId": "1234"},
+        "consumer_inputs": {"accountName": "stafoobar001"},
+    }
+    resource = HookSubscriptionCreate.model_validate(spec)
+    assert resource.publisher_id == "tfs"  # noqa: PLR2004
+    assert resource.event_type == "git.push"  # noqa: PLR2004