# -*- 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 __future__ import annotations
import logging
import re
from collections.abc import Collection, Sequence
from json import JSONDecodeError
from typing import TYPE_CHECKING, Any
from urllib.parse import urlparse
import httpx
from craft_store import errors
from craft_store._httpx_auth import CandidAuth, UbuntuOneAuth
from craft_store.auth import Auth
from craft_store.models import RegisteredNameModel as RegisteredName
from ._response import (
ReleaseResult,
Releases,
Revision,
)
if TYPE_CHECKING:
from . import _request
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/#create-tracks
"""
logger = logging.getLogger(__name__)
[docs]
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: httpx.Auth | Auth,
) -> None:
"""Create a publisher gateway client.
:param base_url: The base URL of the publisher gateway API.
:param namespace: The package namespace (e.g. ``"charm"`` or ``"snap"``).
:param auth: Either an :class:`httpx.Auth` instance (used directly), or a
craft-store :class:`~craft_store.auth.Auth` instance (wrapped in
:class:`~craft_store._httpx_auth.CandidAuth` automatically).
"""
self._namespace = namespace
if isinstance(auth, httpx.Auth):
httpx_auth = auth
else:
httpx_auth = CandidAuth(auth=auth, auth_type="macaroon")
self._client = httpx.Client(
base_url=base_url,
auth=httpx_auth,
timeout=60.0,
)
[docs]
@classmethod
def with_ubuntu_one(
cls,
base_url: str,
namespace: str,
*,
auth: Auth | None = None,
application_name: str = "craft-store",
client_description: str = "craft-store",
) -> PublisherGateway:
"""Create a publisher gateway client using Ubuntu One SSO authentication.
This is a convenience constructor that creates the :class:`UbuntuOneAuth`
internally. Credentials must already be saved in the system keyring (e.g.
by calling :meth:`~craft_store.login.UbuntuOneLogin.login_with` first).
:param base_url: The base URL of the publisher gateway API (e.g.
``"https://api.charmhub.io"``).
:param namespace: The package namespace (e.g. ``"charm"`` or ``"snap"``).
:param auth: An optional :class:`~craft_store.auth.Auth` instance for
keyring access. If not provided, one is created automatically using
``application_name`` and the host derived from ``base_url``.
:param application_name: The application name used for keyring storage.
Must match the name used when credentials were saved. Defaults to
``"craft-store"``.
:param client_description: A description passed to the store during token
exchange. Defaults to ``"craft-store"``.
"""
if auth is None:
auth = Auth(
application_name=application_name,
host=urlparse(base_url).netloc,
)
httpx_auth = UbuntuOneAuth(
auth=auth,
api_base_url=base_url,
client_description=client_description,
)
return cls(base_url=base_url, namespace=namespace, auth=httpx_auth)
@staticmethod
def _check_error(response: httpx.Response) -> None:
"""Check a response for general errors.
:param response: an httpx response from the server.
:raises: InvalidResponseError if the response from the server is invalid.
:raises: CraftStoreError if the response status code is an error code.
"""
if response.is_success:
return
try:
error_response = response.json()
except JSONDecodeError as exc:
logger.debug(f"Error response: {response.text}")
raise errors.InvalidResponseError(response) 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:
fancy_error_list = errors.StoreErrorList(error_list)
brief = f"{brief}.\n{fancy_error_list}"
if error_list:
# Log the errors, but don't pass them to CraftStoreError or they will be
# duplicated.
logger.debug(f"Errors from the store:\n{errors.StoreErrorList(error_list)}")
try:
request = response.request
except RuntimeError:
details = None
else:
details = f"Error occurred on {request.method} request to {request.url}"
if request.content and logger.isEnabledFor(logging.DEBUG):
content = request.content
max_len = 2048
decoded = content[:max_len].decode("utf-8", errors="replace")
suffix = (
""
if len(content) <= max_len
else f"... (truncated, {len(content)} bytes total)"
)
logger.debug("Request content: %s%s", decoded, suffix)
raise errors.CraftStoreError(brief, details=details)
@staticmethod
def _check_keys(
response: httpx.Response, expected_keys: set[str]
) -> dict[str, Any]:
"""Check that a json dictionary has the expected keys.
:param json_response: The deserialised JSON from the server.
:param expected_keys: A set of keys that are expected in the JSON.
:returns: The deserialised JSON from the server.
:raises: InvalidResponseError if the response from the server is invalid.
"""
try:
json_response = response.json()
except JSONDecodeError as exc:
logger.debug(f"Server response: {response.text}")
raise errors.InvalidResponseError(response) from exc
if not isinstance(json_response, dict):
logger.debug(f"Server response: {response.text}")
raise errors.InvalidResponseError(response)
received_expected_keys = expected_keys & json_response.keys()
missing_keys = expected_keys - received_expected_keys
if missing_keys:
logger.debug(f"Server response: {response.text}")
raise errors.InvalidResponseError(
response, details=f"Missing JSON keys: {missing_keys}"
)
return json_response
[docs]
def list_registered_names(
self, include_collaborations: bool = False
) -> Sequence[RegisteredName]:
"""Return names registered by the authenticated user.
:param include_collaborations: if True, includes names the user is a
collaborator on but does not own.
:returns: A sequence of names registered to the user.
API docs: https://api.charmhub.io/docs/default/#list-registered-names
"""
response = self._client.get(
f"/v1/{self._namespace}",
params={"include-collaborations": include_collaborations},
)
self._check_error(response)
results = self._check_keys(response, expected_keys={"results"})["results"]
return [RegisteredName.unmarshal(item) for item in results]
[docs]
def whoami(self) -> dict[str, Any]:
"""Return whoami json data.
API docs: https://api.charmhub.io/docs/default/#macaroon-info
"""
response = self._client.get("/v1/tokens/whoami")
self._check_error(response)
return self._check_keys(response, expected_keys=set())
[docs]
def register_name(
self,
name: str,
*,
entity_type: str,
private: bool = False,
team: str | None = None,
) -> str:
"""Register a name on the store.
:param name: the name to register.
:param entity_type: The type of package to register (e.g. charm or snap)
:param private: Whether this entity is private or not.
:param team: An optional team ID to register the name with.
:returns: the ID of the registered name.
"""
request_json = {
"name": name,
"private": private,
"type": entity_type,
}
if team is not None:
request_json["team"] = team
response = self._client.post(f"/v1/{self._namespace}", json=request_json)
self._check_error(response)
return str(self._check_keys(response, expected_keys={"id"})["id"])
[docs]
def unregister_name(self, name: str) -> str:
"""Unregister a name with no published packages.
:param name: The name to unregister.
:returns: the ID of the deleted name.
API docs: https://api.charmhub.io/docs/default/#unregister-package
"""
response = self._client.delete(f"/v1/{self._namespace}/{name}")
self._check_error(response)
return str(
self._check_keys(response, expected_keys={"package-id"})["package-id"]
)
[docs]
def list_revisions(
self,
name: str,
*,
fields: Collection[str] | None = None,
include_craft_yaml: bool = False,
revision: int | None = None,
) -> Sequence[Revision]:
"""List the revisions for a specific name.
:param name: The name of the package to query.
:param fields: A list of fields to include. These vary by namespace and are only
checked server-side.
:param include_craft_yaml: Whether to include the craft YAML file in the response.
:param revision: If provided, get only the specified revision.
:returns: A list of revisions in the store and their metadata.
API docs: https://api.charmhub.io/docs/default/#list-revisions
"""
params = {}
if fields is not None:
params["fields"] = ",".join(fields)
if include_craft_yaml:
params["include-craft-yaml"] = "true"
if revision is not None:
params["revision"] = str(revision)
response = self._client.get(
f"/v1/{self._namespace}/{name}/revisions", params=params
)
self._check_error(response)
response_data = self._check_keys(response, {"revisions"})
return [Revision.unmarshal(revision) for revision in response_data["revisions"]]
[docs]
def list_releases(self, name: str) -> Releases:
"""Get the information about the releases of a name.
:param name: The name of the package to query.
:returns: Channel info, package info and revision info.
The revision information returned is only for the revisions that are currently
published in a channel.
API docs: https://api.charmhub.io/docs/default/#list-releases
"""
response = self._client.get(f"/v1/{self._namespace}/{name}/releases")
self._check_error(response)
return Releases.unmarshal(response.json())
def release(
self, name: str, requests: list[_request.ReleaseRequest]
) -> Sequence[ReleaseResult]:
response = self._client.post(
f"/v1/{self._namespace}/{name}/releases", json=requests
)
self._check_error(response)
return [
ReleaseResult.unmarshal(rel)
for rel in self._check_keys(response, {"released"})["released"]
]
[docs]
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: InvalidRequestError if the name field of any passed track is invalid.
API docs: https://api.charmhub.io/docs/default/#create-tracks
"""
bad_track_names = {
track["name"]
for track in tracks
if not TRACK_NAME_REGEX.match(track["name"]) or len(track["name"]) > 28
}
if bad_track_names:
bad_tracks = ", ".join(sorted(bad_track_names))
raise errors.InvalidRequestError(
f"The following track names are invalid: {bad_tracks}",
resolution="Ensure all tracks have valid names.",
)
response = self._client.post(
f"/v1/{self._namespace}/{name}/tracks", json=tracks
)
self._check_error(response)
return int(
self._check_keys(response, expected_keys={"num-tracks-created"})[
"num-tracks-created"
]
)
