feat: rewrite Twine adapter for uploading to artifact repositories

Artifact upload generalised to fully support custom repositories like
GitLab. Rewritten to use twine python api instead of running the
executable. No-op mode now respected by artifact upload.

semantic_release/cli.py

@@ -3,13 +3,13 @@
 import logging
 import os
 import sys
+from pathlib import Path
 
 import click
 import click_log
 
 from semantic_release import ci_checks
 from semantic_release.errors import GitError, ImproperConfigurationError
-
 from .changelog import markdown_changelog
 from .dist import build_dists, remove_dists, should_build, should_remove_dist
 from .history import (
@@ -28,7 +28,7 @@
     post_changelog,
     upload_to_release,
 )
-from .repository import get_repository
+from .repository import ArtifactRepo
 from .settings import config, overload_configuration
 from .vcs_helpers import (
     checkout,
@@ -49,7 +49,7 @@
     'pypi_token_var',
     'pypi_user_var',
     'repository_user_var',
-    'repository_password_var',
+    'repository_pass_var',
 ]
 
 COMMON_OPTIONS = [
@@ -231,11 +231,11 @@ def changelog(*, unreleased=False, noop=False, post=False, **kwargs):
             logger.error("Missing token: cannot post changelog to HVCS")
 
 
-def publish(**kwargs):
+def publish(retry: bool = False, noop: bool = False, **kwargs):
     """Run the version task, then push to git and upload to an artifact repository / GitHub Releases."""
     current_version = get_current_version()
 
-    retry = kwargs.get("retry")
+    verbose = logger.isEnabledFor(logging.DEBUG)
     if retry:
         logger.info("Retry is on")
         # The "new" version will actually be the current version, and the
@@ -259,7 +259,7 @@ def publish(**kwargs):
         current_version=current_version,
         new_version=new_version,
         retry=retry,
-        noop=kwargs.get("noop"),
+        noop=noop,
     ):
         log = generate_changelog(current_version)
         changelog_md = markdown_changelog(
@@ -286,13 +286,8 @@ def publish(**kwargs):
 
         # Get config options for uploads
         dist_path = config.get("dist_path")
-        upload_to_artifact_repository = config.get("upload_to_repository") and config.get("upload_to_pypi")
-        dist_glob_patterns = config.get("dist_glob_patterns", config.get("upload_to_pypi_glob_patterns"))
         upload_release = config.get("upload_to_release")
 
-        if dist_glob_patterns:
-            dist_glob_patterns = dist_glob_patterns.split(",")
-
         if should_build():
             # We need to run the command to build wheels for releasing
             logger.info("Building distributions")
@@ -301,14 +296,9 @@ def publish(**kwargs):
                 remove_dists(dist_path)
             build_dists()
 
-        if upload_to_artifact_repository:
+        if ArtifactRepo.upload_enabled():
             logger.info("Uploading to artifact Repository")
-            get_repository().upload(
-                path=dist_path,
-                # If we are retrying, we don't want errors for files that are already on PyPI.
-                skip_existing=retry,
-                glob_patterns=dist_glob_patterns,
-            )
+            ArtifactRepo(Path(dist_path)).upload(noop=noop, verbose=verbose, skip_existing=retry)
 
         if check_token():
             # Update changelog on HVCS

semantic_release/defaults.cfg

@@ -25,7 +25,7 @@ pypi_pass_var=PYPI_PASSWORD
 pypi_token_var=PYPI_TOKEN
 pypi_user_var=PYPI_USERNAME
 repository_user_var=REPOSITORY_USERNAME
-repository_password_var=REPOSITORY_PASSWORD
+repository_pass_var=REPOSITORY_PASSWORD
 remove_dist=true
 tag_commit=true
 tag_format=v{version}

semantic_release/repository.py

@@ -2,61 +2,164 @@
 """
 import logging
 import os
-from typing import List
+from dataclasses import InitVar, asdict as dataclass_asdict, dataclass, field
+from pathlib import Path
+from typing import Any, Dict, List, Optional
 
-from invoke import run
+import requests
+from twine.commands.upload import upload as twine_upload
+from twine.exceptions import TwineException
+from twine.settings import Settings as TwineSettings
 
 from semantic_release import ImproperConfigurationError
-
+from semantic_release.helpers import LoggedFunction
 from semantic_release.settings import config
 
-from .helpers import LoggedFunction
-
 logger = logging.getLogger(__name__)
 
 
-class Repository(object):
+def get_env_var(name: str) -> Optional[str]:
+    """
+    Resolve variable name from config and return matching environment variable
+
+    :param name: Variable name to retrieve from environment
 
-    def __init__(self, username: str = None, password: str = None) -> None:
-        self._username = username if username else os.environ.get("REPOSITORY_USERNAME")
-        self._password = password if password else os.environ.get("REPOSITORY_PASSWORD")
-        if not self._username or not self._password:
-            raise ImproperConfigurationError("Missing credentials for uploading to repository")
-        self.repository_url = config.get('repository_url', None)
-        super().__init__()
+    :returns Value of environment variable or None if not set.
+    """
+    return os.environ.get(config.get(name))
 
 
+@dataclass(eq=False)
+class ArtifactRepo:
+    """
+    Object that manages the configuration and execution of upload using Twine.
+
+    This object needs only one shared argument to be instantiated.
+    """
+
+    dist_path: InitVar[Path]
+    repository_name: str = "pypi"
+    repository_url: Optional[str] = None
+    username: Optional[str] = field(repr=False, default=None)
+    password: Optional[str] = field(repr=False, default=None)
+    dists: List[str] = field(init=False, default_factory=list)
+
+    def __post_init__(self, dist_path: Path) -> None:
+        """
+        :param dist_path: Path to dist folder containing the files to upload.
+        """
+        self._handle_credentials_init()
+        self._handle_repository_config()
+        self._handle_glob_patterns(dist_path)
+
     @LoggedFunction(logger)
-    def upload(self, path: str = "dist", skip_existing: bool = False, glob_patterns: List[str] = None):
+    def _handle_credentials_init(self) -> None:
         """
-        Upload artifact to repository with twine
+        Initialize credentials from environment variables.
 
-        :param path: Path to dist folder containing the files to upload.
-        :param skip_existing: Continue uploading files if one already exists.
-            (Only valid when uploading to PyPI. Other implementations may not support this.)
-        :param glob_patterns: List of glob patterns to include in the upload (["*"] by default)."""
-        if not glob_patterns:
-            glob_patterns = ["*"]
-        dist = " ".join(
-            ['"{}/{}"'.format(path, glob_pattern.strip()) for glob_pattern in glob_patterns]
-        )
+        For the transitional period until the *pypi* variables can be safely removed,
+        additional complexity is needed.
 
-        extra_parms = " ".join([""] + self._extra_twine_arguments()) if self._extra_twine_arguments() else ""
-        extra_parms += " --skip-existing" if skip_existing else ""
+        :raises ImproperConfigurationError:
+            Error while setting up credentials configuration.
+        """
+        username = get_env_var('repository_user_var') or get_env_var('pypi_user_var')
+        password = get_env_var('repository_pass_var') or get_env_var('pypi_pass_var') or get_env_var('pypi_token_var')
+        if username and password:
+            self.username = username
+            self.password = password
+        elif password and not username:
+            self.username = username or "__token__"
+            self.password = password
+            logger.warning("Providing only password or token without username is deprecated")
+        # neither username nor password provided, check for ~/.pypirc file
+        elif not Path("~/.pypirc").expanduser().exists():
+            raise ImproperConfigurationError("Missing credentials for uploading to artifact repository")
 
-        run(f"twine upload -u '{self._username}' -p '{self._password}'{extra_parms} {dist}")
+    @LoggedFunction(logger)
+    def _handle_glob_patterns(self, dist_path: Path) -> None:
+        """
+        Load glob patterns that select the distribution files to publish.
 
-    def _extra_twine_arguments(self) -> List[str]:
-        return [f"--repository-url '{self.repository_url}'"] if self.repository_url else []
+        :param dist_path: Path to folder with package files
+        """
+        glob_patterns = config.get("dist_glob_patterns") or config.get("upload_to_pypi_glob_patterns")
+        glob_patterns = (glob_patterns or "*").split(",")
 
+        self.dists = [str(dist_path.joinpath(pattern)) for pattern in glob_patterns]
 
-def get_repository() -> Repository:
-    """Return an artifact repository preconfigured based on env variables
-    """
-    token = os.environ.get("PYPI_TOKEN")
-    if not token:
-        return Repository(os.environ.get("PYPI_USERNAME"), os.environ.get("PYPI_PASSWORD"))
-    elif token.startswith("pypi-"):
-        return Repository("__token__", token)
-    else:
-        raise ImproperConfigurationError('PyPI token should begin with "pypi-"')
+    @LoggedFunction(logger)
+    def _handle_repository_config(self) -> None:
+        """
+        Initialize repository settings from config.
+
+        *repository_url* overrides *repository*, Twine handles this the same way.
+
+        Defaults to repository_name `pypi` when both are not set.
+        """
+        repository_url = config.get("repository_url")
+        repository_name = config.get("repository")
+
+        if repository_url:
+            self.repository_url = repository_url
+        elif repository_name:
+            self.repository_name = repository_name
+
+    @LoggedFunction(logger)
+    def _create_twine_settings(self, addon_kwargs: Dict[str, Any]) -> TwineSettings:
+        """
+        Gather all parameters that had a value set during instantiation and
+        pass them to Twine which then validates and laods the config.
+        """
+        params = {name: val for name, val in dataclass_asdict(self).items() if val}
+        settings = TwineSettings(**params, **addon_kwargs)
+
+        return settings
+
+    @LoggedFunction(logger)
+    def upload(
+        self, noop: bool, verbose: bool, skip_existing: bool, **additional_kwargs
+    ) -> bool:
+        """
+        Upload artifact to repository using Twine.
+
+        For known repositories (like PyPI), the web URLs of successfully uploaded packages
+        will be displayed.
+
+        :param noop: Do not apply any changes..
+        :param verbose: Show verbose output for Twine.
+        :param skip_existing: Continue uploading files if one already exists.
+            (May not work, check your repository for support.)
+
+        :raises ImproperConfigurationError:
+            The upload failed due to a configuration error.
+
+        :returns True if successfull, False otherwise.
+        """
+        addon_kwargs = {
+            "non_interactive": True,
+            "verbose": verbose,
+            "skip_existing": skip_existing,
+            **additional_kwargs,
+        }
+
+        try:
+            twine_settings = self._create_twine_settings(addon_kwargs)
+            if not noop:
+                twine_upload(upload_settings=twine_settings, dists=self.dists)
+        except TwineException as e:
+            raise ImproperConfigurationError("Upload to artifact repository has failed") from e
+        except requests.HTTPError as e:
+            logger.warning(f"Upload to artifact repository has failed: {e}")
+            return False
+        else:
+            return True
+
+    @staticmethod
+    def upload_enabled() -> bool:
+        """
+        Check if artifact repository upload is enabled
+
+        :returns True if upload is enabled, False otherwise.
+        """
+        return config.get("upload_to_repository") and config.get("upload_to_pypi")

setup.py

@@ -49,6 +49,7 @@ def _read_long_description():
         # for why tomlkit is pinned
         "tomlkit==0.7.0",
         "dotty-dict>=1.3.0,<2",
+        "dataclasses==0.8; python_version < '3.7.0'",
     ],
     extras_require={
         "test": [