feat: add publisher gateway client

This adds a very basic publisher gateway client that meets the needs
of charmcraft's new 'create-track' command.

The client cannot login yet and still depends on the old requests-based
clients for many features.

craft_store/__init__.py

@@ -20,9 +20,10 @@
 __version__ = "3.0.0"
 
 
-from . import creds, endpoints, errors, models
+from . import creds, endpoints, errors, models, publishergateway
 from .auth import Auth
 from .base_client import BaseClient
+from .candidauth import CandidAuth
 from .developer_token_auth import DeveloperTokenAuth
 from .http_client import HTTPClient
 from .store_client import StoreClient
@@ -33,8 +34,10 @@
     "endpoints",
     "errors",
     "models",
+    "publishergateway",
     "Auth",
     "BaseClient",
+    "CandidAuth",
     "HTTPClient",
     "StoreClient",
     "UbuntuOneStoreClient",

craft_store/candidauth.py

@@ -0,0 +1,72 @@
+# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
+#
+# Copyright 2024 Canonical Ltd.
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License version 3 as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Client for making requests towards publisher gateway."""
+
+from collections.abc import Generator
+from logging import getLogger
+from typing import Literal
+
+import httpx
+
+from craft_store import auth, creds, errors
+
+logger = getLogger(__name__)
+
+
+class CandidAuth(httpx.Auth):
+    """Request authentication using developer token."""
+
+    def __init__(
+        self,
+        *,
+        auth: auth.Auth,
+        auth_type: Literal["bearer", "macaroon"] = "bearer",
+    ) -> None:
+        self._auth = auth
+        self._auth_type = auth_type
+        self._token: str | None = None
+
+    def auth_flow(
+        self,
+        request: httpx.Request,
+    ) -> Generator[httpx.Request, httpx.Response, None]:
+        """Update request to include Authorization header."""
+        if self._token is None:
+            logger.debug("Getting candid macaroon from keyring")
+            self._token = self.get_token_from_keyring()
+
+        self._update_headers(request)
+        yield request
+
+    def get_token_from_keyring(self) -> str:
+        """Get token stored in the credentials storage."""
+        logger.debug("Getting candid from credential storage")
+        return creds.unmarshal_candid_credentials(self._auth.get_credentials())
+
+    def _update_headers(self, request: httpx.Request) -> None:
+        """Add token to the request."""
+        logger.debug("Adding ephemeral token to request headers")
+        if self._token is None:
+            raise errors.DeveloperTokenUnavailableError(
+                message="Candid token is not available"
+            )
+        request.headers["Authorization"] = self._format_auth_header()
+
+    def _format_auth_header(self) -> str:
+        if self._auth_type == "bearer":
+            return f"Bearer {self._token}"
+        return f"Macaroon {self._token}"

craft_store/errors.py

@@ -15,11 +15,13 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 """Craft Store errors."""
+from __future__ import annotations
 
 import contextlib
 import logging
 from typing import Any
 
+import httpx
 import requests
 import urllib3
 import urllib3.exceptions
@@ -31,9 +33,19 @@
 class CraftStoreError(Exception):
     """Base class error for craft-store."""
 
-    def __init__(self, message: str, resolution: str | None = None) -> None:
+    def __init__(
+        self,
+        message: str,
+        details: str | None = None,
+        resolution: str | None = None,
+        store_errors: StoreErrorList | None = None,
+    ) -> None:
         super().__init__(message)
+        if not details:
+            details = str(store_errors)
+        self.details = details
         self.resolution = resolution
+        self.store_errors = store_errors
 
 
 class NetworkError(CraftStoreError):
@@ -75,7 +87,7 @@ def __repr__(self) -> str:
             if code:
                 code_list.append(code)
 
-        return "<StoreErrorList: {' '.join(code_list)}>"
+        return f"<StoreErrorList: {' '.join(code_list)}>"
 
     def __contains__(self, error_code: str) -> bool:
         return any(error.get("code") == error_code for error in self._error_list)
@@ -111,7 +123,7 @@ def _get_raw_error_list(self) -> list[dict[str, str]]:
 
         return error_list
 
-    def __init__(self, response: requests.Response) -> None:
+    def __init__(self, response: requests.Response | httpx.Response) -> None:
         self.response = response
 
         try:
@@ -126,9 +138,13 @@ def __init__(self, response: requests.Response) -> None:
             with contextlib.suppress(KeyError):
                 message = "Store operation failed:\n" + str(self.error_list)
         if message is None:
+            if isinstance(response, httpx.Response):
+                reason = response.reason_phrase
+            else:
+                reason = response.reason
             message = (
                 "Issue encountered while processing your request: "
-                f"[{response.status_code}] {response.reason}."
+                f"[{response.status_code}] {reason}."
             )
 
         super().__init__(message)

craft_store/publishergateway/__init__.py

@@ -0,0 +1,35 @@
+# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
+#
+# Copyright 2024 Canonical Ltd.
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License version 3 as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""Package containing the Publisher Gateway client and relevant metadata."""
+
+from ._request import (
+    CreateTrackRequest,
+)
+
+from ._response import (
+    PackageMetadata,
+    PublisherMetadata,
+    TrackMetadata,
+)
+from ._publishergw import PublisherGateway
+
+__all__ = [
+    "CreateTrackRequest",
+    "PackageMetadata",
+    "PublisherMetadata",
+    "TrackMetadata",
+    "PublisherGateway",
+]

craft_store/publishergateway/_publishergw.py

@@ -0,0 +1,106 @@
+# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
+#
+# Copyright 2024 Canonical Ltd.
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License version 3 as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""Client for the publisher gateway."""
+
+from typing import cast
+import httpx
+
+from craft_store import auth, candidauth, errors
+
+from . import _request, _response
+
+
+class PublisherGateway:
+    """Client for the publisher gateway.
+
+    This class is a client wrapper for the Canonical Publisher Gateway.
+    The latest version of the server API can be seen at: https://api.charmhub.io/docs/
+
+    Each instance is only valid for one particular namespace.
+    """
+
+    def __init__(self, base_url: str, namespace: str, auth: auth.Auth) -> None:
+        self._namespace = namespace
+        self._client = httpx.Client(
+            base_url=base_url,
+            auth=candidauth.CandidAuth(auth=auth, auth_type="macaroon"),
+        )
+
+    @staticmethod
+    def _check_error(response: httpx.Response) -> None:
+        if response.is_success:
+            return
+        try:
+            error_response = response.json()
+        except Exception as exc:
+            raise errors.CraftStoreError(
+                f"Invalid response from server ({response.status_code})",
+                details=response.text,
+            ) from exc
+        error_list = error_response.get("error-list", [])
+        if response.status_code >= 500:
+            brief = f"Store had an error ({response.status_code})"
+        else:
+            brief = f"Error {response.status_code} returned from store"
+        if len(error_list) == 1:
+            brief = f"{brief}: {error_list[0].get('message')}"
+        else:
+            brief = f"{brief}. See log for details"
+        raise errors.CraftStoreError(
+            brief, store_errors=errors.StoreErrorList(error_list)
+        )
+
+    def get_package_metadata(self, name: str) -> _response.PackageMetadata:
+        """Get general metadata for a package.
+
+        :param name: The name of the package to query.
+        :returns: A dictionary matching the result from the publisher gateway.
+
+        API docs: https://api.charmhub.io/docs/default.html#package_metadata
+        """
+        response = self._client.get(
+            url=f"/v1/{self._namespace}/{name}",
+        )
+        self._check_error(response)
+        return cast(_response.PackageMetadata, response.json()["metadata"])
+
+    def create_tracks(self, name: str, *tracks: _request.CreateTrackRequest) -> int:
+        """Create one or more tracks in the store.
+
+        :param name: The store name (i.e. the specific charm, snap or other package)
+            to which this track will be attached.
+        :param tracks: Each track is a dictionary mapping query values.
+        :returns: The number of tracks created by the store.
+        :raises: ValueError if a track name is invalid.
+
+        API docs: https://api.charmhub.io/docs/default.html#create_tracks
+        """
+        bad_track_names = {
+            track["name"]
+            for track in tracks
+            if not _request.TRACK_NAME_REGEX.match(track["name"])
+            or len(track["name"]) > 28
+        }
+        if bad_track_names:
+            bad_tracks = ", ".join(sorted(bad_track_names))
+            raise ValueError(f"The following track names are invalid: {bad_tracks}")
+
+        response = self._client.post(
+            f"/v1/{self._namespace}/{name}/tracks", json=tracks
+        )
+        self._check_error(response)
+
+        return int(response.json()["num-tracks-created"])

craft_store/publishergateway/_request.py

@@ -0,0 +1,41 @@
+# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
+#
+# Copyright 2024 Canonical Ltd.
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License version 3 as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""Request models for the publisher gateway."""
+
+import re
+from typing import Annotated, TypedDict
+
+import annotated_types
+from typing_extensions import NotRequired
+
+TRACK_NAME_REGEX = re.compile(r"^[a-zA-Z0-9](?:[_.-]?[a-zA-Z0-9])*$")
+"""A regular expression guarding track names.
+
+Retrieved from https://api.staging.charmhub.io/docs/default.html#create_tracks
+"""
+
+CreateTrackRequest = TypedDict(
+    "CreateTrackRequest",
+    {
+        "name": Annotated[
+            str,
+            annotated_types.Len(1, 28),
+            annotated_types.Predicate(lambda name: bool(TRACK_NAME_REGEX.match(name))),
+        ],
+        "version-pattern": NotRequired[str | None],
+        "automatic-phasing-percentage": NotRequired[str | None],
+    },
+)