feat(store): promote command

charmcraft/application/commands/__init__.py

@@ -46,6 +46,7 @@
     # release process, and show status
     CreateTrack,
     ReleaseCommand,
+    PromoteCommand,
     PromoteBundleCommand,
     StatusCommand,
     CloseCommand,
@@ -91,6 +92,7 @@ def fill_command_groups(app: craft_application.Application) -> None:
             # release process, and show status
             CreateTrack,
             ReleaseCommand,
+            PromoteCommand,
             PromoteBundleCommand,
             StatusCommand,
             CloseCommand,

charmcraft/application/commands/store.py

@@ -25,13 +25,14 @@
 import re
 import shutil
 import string
+import sys
 import tempfile
 import textwrap
 import typing
 import zipfile
 from collections.abc import Collection
 from operator import attrgetter
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 import yaml
 from craft_application import util
@@ -43,11 +44,13 @@
 from craft_store.models import ResponseCharmResourceBase
 from humanize import naturalsize
 from tabulate import tabulate
+from typing_extensions import override
 
 import charmcraft.store.models
 from charmcraft import const, env, errors, parts, utils
 from charmcraft.application.commands.base import CharmcraftCommand
 from charmcraft.models import project
+from charmcraft.services.store import StoreService
 from charmcraft.store import Store
 from charmcraft.store.models import Entity
 from charmcraft.utils import cli
@@ -830,6 +833,141 @@ def run(self, parsed_args):
         emit.message(msg.format(*args))
 
 
+class PromoteCommand(CharmcraftCommand):
+    """Promote a charm in the Store."""
+
+    name = "promote"
+    help_msg = "Promote a charm from one channel to another on Charmhub."
+    overview = "TODO"
+
+    @override
+    def needs_project(self, parsed_args: argparse.Namespace) -> bool:
+        if parsed_args.name is None:
+            emit.progress("Inferring name from project file.", permanent=True)
+            return True
+        return False
+
+    @override
+    def fill_parser(self, parser: "ArgumentParser") -> None:
+        parser.add_argument(
+            "--name",
+            help="the name of the charm to promote. If not specified, the name will be inferred from the charm in the current directory.",
+        )
+        parser.add_argument(
+            "--from-channel",
+            metavar="from-channel",
+            help="the channel to promote from",
+            required=True,
+        )
+        parser.add_argument(
+            "--to-channel",
+            metavar="to-channel",
+            help="the channel to promote to",
+            required=True,
+        )
+        parser.add_argument(
+            "--yes",
+            default=False,
+            action="store_true",
+            help="Answer yes to all questions.",
+        )
+
+    @override
+    def run(self, parsed_args: argparse.Namespace) -> int | None:
+        emit.progress(
+            f"{self._app.name} {self.name} does not have a stable CLI interface. "
+            "Use with caution in scripts.",
+            permanent=True,
+        )
+        store = cast(StoreService, self._services.get("store"))
+
+        name = parsed_args.name or self._services.project.name
+
+        # Check snapcraft for equiv logic
+        from_channel = charmcraft.store.models.ChannelData.from_str(
+            parsed_args.from_channel
+        )
+        to_channel = charmcraft.store.models.ChannelData.from_str(
+            parsed_args.to_channel
+        )
+        if None in (from_channel.track, to_channel.track):
+            package_metadata = store.get_package_metadata(name)
+            default_track = package_metadata.default_track
+            if from_channel.track is None:
+                from_channel = dataclasses.replace(from_channel, track=default_track)
+            if to_channel.track is None:
+                to_channel = dataclasses.replace(to_channel, track=default_track)
+
+        if to_channel == from_channel:
+            raise CraftError(
+                "Cannot promote from a channel to the same channel.",
+                retcode=64,  # Replace with os.EX_USAGE once we drop Windows.
+            )
+        if to_channel.risk > from_channel.risk:
+            command_parts = [
+                self._app.name,
+                f"--from-channel={to_channel.name}",
+                f"--to-channel={from_channel.name}",
+                self.name,
+            ]
+            command = " ".join(command_parts)
+            raise CraftError(
+                f"Target channel ({to_channel.name}) must be lower risk "
+                f"than the source channel ({from_channel.name}).",
+                resolution=f"Did you mean: {command}",
+            )
+        if to_channel.track != from_channel.track:
+            if not parsed_args.yes and not utils.confirm_with_user(
+                "Did you mean to promote to a different track? (from "
+                f"{from_channel.track} to {to_channel.track})",
+            ):
+                emit.message("Cancelling.")
+                return 64  # Replace with os.EX_USAGE once we drop Windows.
+
+        candidates = store.get_revisions_on_channel(name, from_channel.name)
+
+        def get_base_strings(bases):
+            if bases is None:
+                return ""
+            return ",".join(
+                f"{base.name}@{base.channel}:{base.architecture}" for base in bases
+            )
+
+        presentable_candidates = [
+            {
+                "Revision": info["revision"],
+                "Platforms": get_base_strings(info["bases"]),
+                "Resource revisions": ", ".join(
+                    f"{res['name']}: {res['revision']}" for res in info["resources"]
+                ),
+            }
+            for info in sorted(candidates, key=lambda x: x["revision"])
+        ]
+        emit.progress(
+            f"The following revisions are on the {from_channel.name} channel:",
+            permanent=True,
+        )
+        with emit.pause():
+            print(
+                tabulate(presentable_candidates, tablefmt="plain", headers="keys"),
+                file=sys.stderr,
+            )
+        if not parsed_args.yes and not utils.confirm_with_user(
+            f"Do you want to promote these revisions to the {to_channel.name} channel?"
+        ):
+            emit.message("Channel promotion cancelled.")
+            return 1
+
+        promotion_results = store.release_promotion_candidates(
+            name, to_channel.name, candidates
+        )
+
+        emit.message(
+            f"{len(promotion_results)} revisions promoted from {from_channel.name} to {to_channel.name}"
+        )
+        return 0
+
+
 class PromoteBundleCommand(CharmcraftCommand):
     """Promote a bundle in the Store."""
 
@@ -2098,7 +2236,6 @@ def run(self, parsed_args: argparse.Namespace) -> int:
                     resolution="Pass a valid container transport string.",
                 )
             emit.debug(f"Using source path {source_path!r}")
-
             emit.progress("Inspecting source image")
             image_metadata = image_service.inspect(source_path)
 

charmcraft/services/store.py

@@ -19,6 +19,7 @@
 
 import platform
 from collections.abc import Collection, Mapping, Sequence
+from typing import Any, cast
 
 import craft_application
 import craft_store
@@ -30,7 +31,7 @@
 from charmcraft import const, env, errors, store
 from charmcraft.models import CharmLib
 from charmcraft.store import AUTH_DEFAULT_PERMISSIONS, AUTH_DEFAULT_TTL
-from charmcraft.store.models import Library, LibraryMetadataRequest
+from charmcraft.store.models import ChannelData, Library, LibraryMetadataRequest
 
 
 class BaseStoreService(craft_application.AppService):
@@ -185,6 +186,7 @@ class StoreService(BaseStoreService):
     ClientClass = store.Client
     client: store.Client  # pyright: ignore[reportIncompatibleVariableOverride]
     anonymous_client: store.AnonymousClient
+    _publisher: craft_store.PublisherGateway
 
     @override
     def setup(self) -> None:
@@ -205,6 +207,91 @@ def setup(self) -> None:
             auth=self._auth,
         )
 
+    def get_package_metadata(self, name: str) -> publisher.RegisteredName:
+        """Get the metadata for a package.
+
+        :param name: The name of the package in this namespace.
+        :returns: A RegisteredName model containing store metadata.
+        """
+        return self._publisher.get_package_metadata(name)
+
+    def release(
+        self, name: str, requests: list[publisher.ReleaseRequest]
+    ) -> Sequence[publisher.ReleaseResult]:
+        """Release one or more revisions to one or more channels.
+
+        :param name: The name of the package to update.
+        :param requests: A list of dictionaries containing the requests.
+        :returns: A sequence of results of the release requests, as returned
+            by the store.
+
+        Each request dictionary requires a "channel" key with the channel name and
+        a "revision" key with the revision number. If the revision in the store has
+        resources, it requires a "resources" key that is a list of dictionaries
+        containing a "name" key with the resource name and a "revision" key with
+        the resource number to attach to that channel release.
+        """
+        return self._publisher.release(name, requests=requests)
+
+    def get_revisions_on_channel(
+        self, name: str, channel: str
+    ) -> Sequence[Mapping[str, Any]]:
+        """Get the current set of revisions on a specific channel.
+
+        :param name: The name on the store to look up.
+        :param channel: The channel on which to get the revisions.
+        :returns: A sequence of mappings of these, containing their revision,
+            bases, resources and version.
+
+        The mapping here may be passed directly into release_promotion_candidates
+        in order promote items from one channel to another.
+        """
+        releases = self._publisher.list_releases(name)
+        channel_data = ChannelData.from_str(channel)
+        channel_revisions = {
+            info.revision: info
+            for info in releases.channel_map
+            if info.channel == channel_data
+        }
+        revisions = {
+            rev.revision: cast(publisher.CharmRevision, rev)
+            for rev in releases.revisions
+        }
+
+        return [
+            {
+                "revision": revision,
+                "bases": revisions[revision].bases,
+                "resources": [
+                    {"name": res.name, "revision": res.revision}
+                    for res in info.resources or ()
+                ],
+                "version": revisions[revision].version,
+            }
+            for revision, info in channel_revisions.items()
+        ]
+
+    def release_promotion_candidates(
+        self, name: str, channel: str, candidates: Collection[Mapping[str, Any]]
+    ) -> Sequence[publisher.ReleaseResult]:
+        """Promote a set of revisions to a specific channel.
+
+        :param name: the store name to operate on.
+        :param channel: The channel to which these should be promoted.
+        :param candidates: A collection of mappings containing the revision and
+            resource revisions to promote.
+        :returns: The result of the release in the store.
+        """
+        requests = [
+            publisher.ReleaseRequest(
+                channel=channel,
+                resources=candidate["resources"],
+                revision=candidate["revision"],
+            )
+            for candidate in candidates
+        ]
+        return self.release(name, requests)
+
     def create_tracks(
         self, name: str, *tracks: publisher.CreateTrackRequest
     ) -> Sequence[publisher.Track]:

charmcraft/store/models.py

@@ -15,6 +15,7 @@
 # For further info, check https://github.com/canonical/charmcraft
 """Internal models for store data structiues."""
 
+import contextlib
 import dataclasses
 import datetime
 import enum
@@ -282,6 +283,20 @@ def name(self) -> str:
         risk = self.risk.name.lower()
         return "/".join(i for i in (self.track, risk, self.branch) if i is not None)
 
+    def __eq__(self, other: object, /) -> bool:
+        if isinstance(other, ChannelData):
+            return (
+                self.track == other.track
+                and self.risk == other.risk
+                and self.branch == other.branch
+            )
+
+        if isinstance(other, str):
+            with contextlib.suppress(CraftError):
+                return self == ChannelData.from_str(other)
+
+        return NotImplemented
+
 
 LibraryMetadataRequest = TypedDict(
     "LibraryMetadataRequest",

pyproject.toml

@@ -11,7 +11,8 @@ dependencies = [
     "craft-providers>=2.1.0",
     "craft-platforms~=0.5",
     "craft-providers>=2.0.0",
-    "craft-store>=3.1.0",
+    # "craft-store>=3.1.0",
+    "craft-store @ git+https://github.com/canonical/craft-store",
     "distro>=1.7.0",
     "docker>=7.0.0",
     "humanize>=2.6.0",

tests/spread/store/charm-upload-and-release/task.yaml

@@ -66,3 +66,4 @@ execute: |
   if [ $edge_revision -lt $uploaded_revno ]; then
     ERROR "Revision wasn't released. Uploaded revision: $uploaded_revno; Currently on edge: $edge_revision"
   fi
+  charmcraft promote --yes --from-channel=latest/edge --to-channel=latest/beta