Improve a package publishing process

HISTORY.rst

@@ -27,6 +27,16 @@ PlatformIO Core 5
     * `Cppcheck <https://docs.platformio.org/page/plus/check-tools/cppcheck.html>`__ v2.4.1 with new checks and MISRA improvements
     * `PVS-Studio <https://docs.platformio.org/page/plus/check-tools/pvs-studio.html>`__ v7.12 with new diagnostics and extended capabilities for security and safety standards
 
+* **Package Management**
+
+  - Improved a package publishing process:
+
+    * Show package details
+    * Check for conflicting names in the PlatformIO Trusted Registry
+    * Check for duplicates and used version
+
+  - Added a new option ``--non-interactive`` to `pio package publish <https://docs.platformio.org/page/core/userguide/package/cmd_publish.html>`__ command
+
 * **Miscellaneous**
 
   - Ensure that a serial port is ready before running unit tests on a remote target (`issue #3742 <https://github.com/platformio/platformio-core/issues/3742>`_)

platformio/clients/account.py

@@ -207,6 +207,9 @@ class AccountClient(HTTPClient):  # pylint:disable=too-many-public-methods
         app.set_state_item("account", account)
         return result
 
+    def get_logged_username(self):
+        return self.get_account_info(offline=True).get("profile").get("username")
+
     def destroy_account(self):
         return self.send_auth_request("delete", "/v1/account")
 

platformio/clients/registry.py

@@ -15,7 +15,6 @@
 from platformio import __registry_api__, fs
 from platformio.clients.account import AccountClient
 from platformio.clients.http import HTTPClient, HTTPClientError
-from platformio.package.meta import PackageType
 
 # pylint: disable=too-many-arguments
 
@@ -32,18 +31,13 @@ class RegistryClient(HTTPClient):
         kwargs["headers"] = headers
         return self.fetch_json_data(*args, **kwargs)
 
-    def publish_package(
-        self, archive_path, owner=None, released_at=None, private=False, notify=True
+    def publish_package(  # pylint: disable=redefined-builtin
+        self, owner, type, archive_path, released_at=None, private=False, notify=True
     ):
-        account = AccountClient()
-        if not owner:
-            owner = (
-                account.get_account_info(offline=True).get("profile").get("username")
-            )
         with open(archive_path, "rb") as fp:
             return self.send_auth_request(
                 "post",
-                "/v3/packages/%s/%s" % (owner, PackageType.from_archive(archive_path)),
+                "/v3/packages/%s/%s" % (owner, type),
                 params={
                     "private": 1 if private else 0,
                     "notify": 1 if notify else 0,
@@ -59,13 +53,8 @@ class RegistryClient(HTTPClient):
             )
 
     def unpublish_package(  # pylint: disable=redefined-builtin
-        self, type, name, owner=None, version=None, undo=False
+        self, owner, type, name, version=None, undo=False
     ):
-        account = AccountClient()
-        if not owner:
-            owner = (
-                account.get_account_info(offline=True).get("profile").get("username")
-            )
         path = "/v3/packages/%s/%s/%s" % (owner, type, name)
         if version:
             path += "/" + version

platformio/commands/account.py

@@ -184,7 +184,7 @@ def account_destroy():
     click.confirm(
         "Are you sure you want to delete the %s user account?\n"
         "Warning! All linked data will be permanently removed and can not be restored."
-        % client.get_account_info().get("profile").get("username"),
+        % client.get_logged_username(),
         abort=True,
     )
     client.destroy_account()

platformio/commands/package.py

@@ -17,9 +17,13 @@ import tempfile
 from datetime import datetime
 
 import click
+from tabulate import tabulate
 
 from platformio import fs
+from platformio.clients.account import AccountClient
 from platformio.clients.registry import RegistryClient
+from platformio.exception import UserSideException
+from platformio.package.manifest.parser import ManifestParserFactory
 from platformio.package.meta import PackageSpec, PackageType
 from platformio.package.pack import PackagePacker
 from platformio.package.unpack import FileUnpacker, TARArchiver
@@ -79,26 +83,94 @@ def package_pack(package, output):
     default=True,
     help="Notify by email when package is processed",
 )
-def package_publish(package, owner, released_at, private, notify):
-    # publish .tar.gz instantly without repacking
-    if not os.path.isdir(package) and isinstance(
+@click.option(
+    "--non-interactive",
+    is_flag=True,
+    help="Do not show interactive prompt",
+)
+def package_publish(  # pylint: disable=too-many-arguments, too-many-locals
+    package, owner, released_at, private, notify, non_interactive
+):
+    click.secho("Preparing a package...", fg="cyan")
+    owner = owner or AccountClient().get_logged_username()
+    do_not_pack = not os.path.isdir(package) and isinstance(
         FileUnpacker.new_archiver(package), TARArchiver
-    ):
-        response = RegistryClient().publish_package(
-            package, owner, released_at, private, notify
-        )
-        click.secho(response.get("message"), fg="green")
-        return
-
+    )
+    archive_path = None
     with tempfile.TemporaryDirectory() as tmp_dir:  # pylint: disable=no-member
-        with fs.cd(tmp_dir):
-            p = PackagePacker(package)
-            archive_path = p.pack()
-            response = RegistryClient().publish_package(
-                archive_path, owner, released_at, private, notify
+        # publish .tar.gz instantly without repacking
+        if do_not_pack:
+            archive_path = package
+        else:
+            with fs.cd(tmp_dir):
+                p = PackagePacker(package)
+                archive_path = p.pack()
+
+        type_ = PackageType.from_archive(archive_path)
+        manifest = ManifestParserFactory.new_from_archive(archive_path).as_dict()
+        name = manifest.get("name")
+        version = manifest.get("version")
+        data = [
+            ("Type:", type_),
+            ("Owner:", owner),
+            ("Name:", name),
+            ("Version:", version),
+        ]
+        click.echo(tabulate(data, tablefmt="plain"))
+
+        # look for duplicates
+        _check_duplicates(owner, type_, name, version)
+
+        if not non_interactive:
+            click.confirm(
+                "Are you sure you want to publish the %s %s to the registry?\n"
+                % (
+                    type_,
+                    click.style(
+                        "%s/%s@%s" % (owner, name, version),
+                        fg="cyan",
+                    ),
+                ),
+                abort=True,
             )
+
+        response = RegistryClient().publish_package(
+            owner, type_, archive_path, released_at, private, notify
+        )
+        if not do_not_pack:
             os.remove(archive_path)
-            click.secho(response.get("message"), fg="green")
+        click.secho(response.get("message"), fg="green")
+
+
+def _check_duplicates(owner, type, name, version):  # pylint: disable=redefined-builtin
+    items = (
+        RegistryClient()
+        .list_packages(filters=dict(types=[type], names=[name]))
+        .get("items")
+    )
+    if not items:
+        return True
+    # duplicated version by owner
+    if any(
+        item["owner"]["username"] == owner and item["version"]["name"] == version
+        for item in items
+    ):
+        raise UserSideException(
+            "The package `%s/%s@%s` is already published in the registry"
+            % (owner, name, version)
+        )
+    other_owners = [
+        item["owner"]["username"]
+        for item in items
+        if item["owner"]["username"] != owner
+    ]
+    if other_owners:
+        click.secho(
+            "\nWarning! A package with the name `%s` is already published by the next "
+            "owners: `%s`\n" % (name, ", ".join(other_owners)),
+            fg="yellow",
+        )
+    return True
 
 
 @cli.command("unpublish", short_help="Remove a pushed package from the registry")
@@ -119,9 +191,9 @@ def package_publish(package, owner, released_at, private, notify):
 def package_unpublish(package, type, undo):  # pylint: disable=redefined-builtin
     spec = PackageSpec(package)
     response = RegistryClient().unpublish_package(
+        owner=spec.owner or AccountClient().get_logged_username(),
         type=type,
         name=spec.name,
-        owner=spec.owner,
         version=str(spec.requirements),
         undo=undo,
     )