"""HVCS
"""
import logging
import mimetypes
import os
from typing import Any, Optional, Union, cast
from urllib.parse import urlsplit
from gitlab import exceptions, gitlab
from requests import HTTPError, Session
from requests.auth import AuthBase
from urllib3 import Retry
from .errors import ImproperConfigurationError
from .helpers import LoggedFunction, build_requests_session
from .settings import config
from .vcs_helpers import get_formatted_tag
logger = logging.getLogger(__name__)
# Add a mime type for wheels
mimetypes.add_type("application/octet-stream", ".whl")
[docs]class Base(object):
[docs] @staticmethod
def domain() -> str:
raise NotImplementedError
[docs] @staticmethod
def api_url() -> str:
raise NotImplementedError
[docs] @staticmethod
def token() -> Optional[str]:
raise NotImplementedError
[docs] @staticmethod
def check_build_status(owner: str, repo: str, ref: str) -> bool:
raise NotImplementedError
[docs] @classmethod
def post_release_changelog(
cls, owner: str, repo: str, version: str, changelog: str
) -> bool:
raise NotImplementedError
[docs] @classmethod
def upload_dists(cls, owner: str, repo: str, version: str, path: str) -> bool:
# Skip on unsupported HVCS instead of raising error
return True
def _fix_mime_types():
"""Fix incorrect entries in the `mimetypes` registry.
On Windows, the Python standard library's `mimetypes` reads in
mappings from file extension to MIME type from the Windows
registry. Other applications can and do write incorrect values
to this registry, which causes `mimetypes.guess_type` to return
incorrect values, which causes TensorBoard to fail to render on
the frontend.
This method hard-codes the correct mappings for certain MIME
types that are known to be either used by python-semantic-release or
problematic in general.
"""
mimetypes.add_type("text/markdown", ".md")
[docs]class TokenAuth(AuthBase):
"""
requests Authentication for token based authorization
"""
def __init__(self, token):
self.token = token
def __eq__(self, other):
return all(
[
self.token == getattr(other, "token", None),
]
)
def __ne__(self, other):
return not self == other
def __call__(self, r):
r.headers["Authorization"] = f"token {self.token}"
return r
[docs]class Github(Base):
"""Github helper class"""
DEFAULT_DOMAIN = "github.com"
_fix_mime_types()
[docs] @staticmethod
def domain() -> str:
"""Github domain property
:return: The Github domain
"""
# ref: https://docs.github.com/en/actions/reference/environment-variables#default-environment-variables
hvcs_domain = config.get(
"hvcs_domain", os.getenv("GITHUB_SERVER_URL", "").replace("https://", "")
)
domain = hvcs_domain if hvcs_domain else Github.DEFAULT_DOMAIN
return domain
[docs] @staticmethod
def api_url() -> str:
"""Github api_url property
:return: The Github API URL
"""
# not necessarily prefixed with api in the case of a custom domain, so
# can't just default DEFAULT_DOMAIN to github.com
# ref: https://docs.github.com/en/actions/reference/environment-variables#default-environment-variables
hvcs_api_domain = config.get(
"hvcs_api_domain", os.getenv("GITHUB_API_URL", "").replace("https://", "")
)
hostname = (
hvcs_api_domain if hvcs_api_domain else "api." + Github.DEFAULT_DOMAIN
)
return f"https://{hostname}"
[docs] @staticmethod
def token() -> Optional[str]:
"""Github token property
:return: The Github token environment variable (GH_TOKEN) value
"""
return os.environ.get(config.get("github_token_var"))
[docs] @staticmethod
def auth() -> Optional[TokenAuth]:
"""Github token property
:return: The Github token environment variable (GH_TOKEN) value
"""
token = Github.token()
if not token:
return None
return TokenAuth(token)
[docs] @staticmethod
def session(
raise_for_status=True, retry: Union[Retry, bool, int] = True
) -> Session:
session = build_requests_session(raise_for_status=raise_for_status, retry=retry)
session.auth = Github.auth()
return session
[docs] @staticmethod
@LoggedFunction(logger)
def check_build_status(owner: str, repo: str, ref: str) -> bool:
"""Check build status
https://docs.github.com/rest/reference/repos#get-the-combined-status-for-a-specific-reference
:param owner: The owner namespace of the repository
:param repo: The repository name
:param ref: The sha1 hash of the commit ref
:return: Was the build status success?
"""
url = "{domain}/repos/{owner}/{repo}/commits/{ref}/status"
try:
response = Github.session().get(
url.format(domain=Github.api_url(), owner=owner, repo=repo, ref=ref)
)
return response.json().get("state") == "success"
except HTTPError as e:
logger.warning(f"Build status check on Github has failed: {e}")
return False
[docs] @classmethod
@LoggedFunction(logger)
def create_release(cls, owner: str, repo: str, tag: str, changelog: str) -> bool:
"""Create a new release
https://docs.github.com/rest/reference/repos#create-a-release
:param owner: The owner namespace of the repository
:param repo: The repository name
:param tag: Tag to create release for
:param changelog: The release notes for this version
:return: Whether the request succeeded
"""
try:
Github.session().post(
f"{Github.api_url()}/repos/{owner}/{repo}/releases",
json={
"tag_name": tag,
"name": tag,
"body": changelog,
"draft": False,
"prerelease": False,
},
)
return True
except HTTPError as e:
logger.warning(f"Release creation on Github has failed: {e}")
return False
[docs] @classmethod
@LoggedFunction(logger)
def get_release(cls, owner: str, repo: str, tag: str) -> Optional[int]:
"""Get a release by its tag name
https://docs.github.com/rest/reference/repos#get-a-release-by-tag-name
:param owner: The owner namespace of the repository
:param repo: The repository name
:param tag: Tag to get release for
:return: ID of found release
"""
try:
response = Github.session().get(
f"{Github.api_url()}/repos/{owner}/{repo}/releases/tags/{tag}"
)
return response.json().get("id")
except HTTPError as e:
if e.response.status_code != 404:
logger.debug(f"Get release by tag on Github has failed: {e}")
return None
[docs] @classmethod
@LoggedFunction(logger)
def edit_release(cls, owner: str, repo: str, id: int, changelog: str) -> bool:
"""Edit a release with updated change notes
https://docs.github.com/rest/reference/repos#update-a-release
:param owner: The owner namespace of the repository
:param repo: The repository name
:param id: ID of release to update
:param changelog: The release notes for this version
:return: Whether the request succeeded
"""
try:
Github.session().post(
f"{Github.api_url()}/repos/{owner}/{repo}/releases/{id}",
json={"body": changelog},
)
return True
except HTTPError as e:
logger.warning(f"Edit release on Github has failed: {e}")
return False
[docs] @classmethod
@LoggedFunction(logger)
def post_release_changelog(
cls, owner: str, repo: str, version: str, changelog: str
) -> bool:
"""Post release changelog
:param owner: The owner namespace of the repository
:param repo: The repository name
:param version: The version number
:param changelog: The release notes for this version
:return: The status of the request
"""
tag = get_formatted_tag(version)
logger.debug(f"Attempting to create release for {tag}")
success = Github.create_release(owner, repo, tag, changelog)
if not success:
logger.debug("Unsuccessful, looking for an existing release to update")
release_id = Github.get_release(owner, repo, tag)
if release_id:
logger.debug(f"Updating release {release_id}")
success = Github.edit_release(owner, repo, release_id, changelog)
else:
logger.debug(f"Existing release not found")
return success
[docs] @classmethod
@LoggedFunction(logger)
def get_asset_upload_url(
cls, owner: str, repo: str, release_id: str
) -> Optional[str]:
"""Get the correct upload url for a release
https://docs.github.com/en/enterprise-server@3.5/rest/releases/releases#get-a-release
:param owner: The owner namespace of the repository
:param repo: The repository name
:param release_id: ID of the release to upload to
:return: URL found to upload for a release
"""
try:
response = Github.session().get(
f"{Github.api_url()}/repos/{owner}/{repo}/releases/{release_id}"
)
return str(response.json().get("upload_url")).split("{")[0]
except HTTPError as e:
if e.response.status_code != 404:
logger.debug(f"Get release asset upload on Github has failed: {e}")
return None
[docs] @classmethod
@LoggedFunction(logger)
def upload_asset(
cls, owner: str, repo: str, release_id: int, file: str, label: str = None
) -> bool:
"""Upload an asset to an existing release
https://docs.github.com/rest/reference/repos#upload-a-release-asset
:param owner: The owner namespace of the repository
:param repo: The repository name
:param release_id: ID of the release to upload to
:param file: Path of the file to upload
:param label: Custom label for this file
:return: The status of the request
"""
url = cls.get_asset_upload_url(owner, repo, release_id)
if not url:
logger.warning("Could not get release upload url")
return False
content_type = mimetypes.guess_type(file, strict=False)[0]
if not content_type:
content_type = "application/octet-stream"
try:
response = Github.session().post(
url,
params={"name": os.path.basename(file), "label": label},
headers={
"Content-Type": content_type,
},
data=open(file, "rb").read(),
)
logger.debug(
f"Asset upload on Github completed, url: {response.url}, status code: {response.status_code}"
)
return True
except HTTPError as e:
logger.warning(f"Asset upload {file} on Github has failed: {e}")
return False
[docs] @classmethod
def upload_dists(cls, owner: str, repo: str, version: str, path: str) -> bool:
"""Upload distributions to a release
:param owner: The owner namespace of the repository
:param repo: The repository name
:param version: Version to upload for
:param path: Path to the dist directory
:return: The status of the request
"""
# Find the release corresponding to this version
release_id = Github.get_release(owner, repo, get_formatted_tag(version))
if not release_id:
logger.debug("No release found to upload assets to")
return False
# Upload assets
one_or_more_failed = False
for file in os.listdir(path):
file_path = os.path.join(path, file)
if not Github.upload_asset(owner, repo, release_id, file_path):
one_or_more_failed = True
return not one_or_more_failed
[docs]class Gitea(Base):
"""Gitea helper class"""
DEFAULT_DOMAIN = "gitea.com"
DEFAULT_API_PATH = "/api/v1"
_fix_mime_types()
[docs] @staticmethod
def domain() -> str:
"""Gitea domain property
:return: The Gitea domain
"""
# ref: https://docs.gitea.com/en/actions/reference/environment-variables#default-environment-variables
hvcs_domain = config.get(
"hvcs_domain", os.getenv("GITEA_SERVER_URL", "").replace("https://", "")
)
domain = hvcs_domain if hvcs_domain else Gitea.DEFAULT_DOMAIN
return domain
[docs] @staticmethod
def api_url() -> str:
"""Gitea api_url property
:return: The Gitea API URL
"""
hvcs_domain = config.get(
"hvcs_domain", os.getenv("GITEA_SERVER_URL", "").replace("https://", "")
)
hostname = config.get(
"hvcs_api_domain", os.getenv("GITEA_API_URL", "").replace("https://", "")
)
if hvcs_domain and not hostname:
hostname = hvcs_domain + Gitea.DEFAULT_API_PATH
elif not hostname:
hostname = Gitea.DEFAULT_DOMAIN + Gitea.DEFAULT_API_PATH
return f"https://{hostname}"
[docs] @staticmethod
def token() -> Optional[str]:
"""Gitea token property
:return: The Gitea token environment variable (GITEA_TOKEN) value
"""
return os.environ.get(config.get("gitea_token_var"))
[docs] @staticmethod
def auth() -> Optional[TokenAuth]:
"""Gitea token property
:return: The Gitea token environment variable (GITEA_TOKEN) value
"""
token = Gitea.token()
if not token:
return None
return TokenAuth(token)
[docs] @staticmethod
def session(
raise_for_status=True, retry: Union[Retry, bool, int] = True
) -> Session:
session = build_requests_session(raise_for_status=raise_for_status, retry=retry)
session.auth = Gitea.auth()
return session
[docs] @staticmethod
@LoggedFunction(logger)
def check_build_status(owner: str, repo: str, ref: str) -> bool:
"""Check build status
https://gitea.com/api/swagger#/repository/repoCreateStatus
:param owner: The owner namespace of the repository
:param repo: The repository name
:param ref: The sha1 hash of the commit ref
:return: Was the build status success?
"""
url = "{domain}/repos/{owner}/{repo}/statuses/{ref}"
try:
response = Gitea.session().get(
url.format(domain=Gitea.api_url(), owner=owner, repo=repo, ref=ref)
)
data = response.json()
if type(data) == list:
return data[0].get("status") == "success"
else:
return data.get("status") == "success"
except HTTPError as e:
logger.warning(f"Build status check on Gitea has failed: {e}")
return False
[docs] @classmethod
@LoggedFunction(logger)
def create_release(cls, owner: str, repo: str, tag: str, changelog: str) -> bool:
"""Create a new release
https://gitea.com/api/swagger#/repository/repoCreateRelease
:param owner: The owner namespace of the repository
:param repo: The repository name
:param tag: Tag to create release for
:param changelog: The release notes for this version
:return: Whether the request succeeded
"""
try:
Gitea.session().post(
f"{Gitea.api_url()}/repos/{owner}/{repo}/releases",
json={
"tag_name": tag,
"name": tag,
"body": changelog,
"draft": False,
"prerelease": False,
},
)
return True
except HTTPError as e:
logger.warning(f"Release creation on Gitea has failed: {e}")
return False
[docs] @classmethod
@LoggedFunction(logger)
def get_release(cls, owner: str, repo: str, tag: str) -> Optional[int]:
"""Get a release by its tag name
https://gitea.com/api/swagger#/repository/repoGetReleaseByTag
:param owner: The owner namespace of the repository
:param repo: The repository name
:param tag: Tag to get release for
:return: ID of found release
"""
try:
response = Gitea.session().get(
f"{Gitea.api_url()}/repos/{owner}/{repo}/releases/tags/{tag}"
)
return response.json().get("id")
except HTTPError as e:
if e.response.status_code != 404:
logger.debug(f"Get release by tag on Gitea has failed: {e}")
return None
[docs] @classmethod
@LoggedFunction(logger)
def edit_release(cls, owner: str, repo: str, id: int, changelog: str) -> bool:
"""Edit a release with updated change notes
https://gitea.com/api/swagger#/repository/repoEditRelease
:param owner: The owner namespace of the repository
:param repo: The repository name
:param id: ID of release to update
:param changelog: The release notes for this version
:return: Whether the request succeeded
"""
try:
Gitea.session().post(
f"{Gitea.api_url()}/repos/{owner}/{repo}/releases/{id}",
json={"body": changelog},
)
return True
except HTTPError as e:
logger.warning(f"Edit release on Gitea has failed: {e}")
return False
[docs] @classmethod
@LoggedFunction(logger)
def post_release_changelog(
cls, owner: str, repo: str, version: str, changelog: str
) -> bool:
"""Post release changelog
:param owner: The owner namespace of the repository
:param repo: The repository name
:param version: The version number
:param changelog: The release notes for this version
:return: The status of the request
"""
tag = get_formatted_tag(version)
logger.debug(f"Attempting to create release for {tag}")
success = Gitea.create_release(owner, repo, tag, changelog)
if not success:
logger.debug("Unsuccessful, looking for an existing release to update")
release_id = Gitea.get_release(owner, repo, tag)
if release_id:
logger.debug(f"Updating release {release_id}")
success = Gitea.edit_release(owner, repo, release_id, changelog)
else:
logger.debug(f"Existing release not found")
return success
[docs] @classmethod
@LoggedFunction(logger)
def upload_asset(
cls, owner: str, repo: str, release_id: int, file: str, label: str = None
) -> bool:
"""Upload an asset to an existing release
https://gitea.com/api/swagger#/repository/repoCreateReleaseAttachment
:param owner: The owner namespace of the repository
:param repo: The repository name
:param release_id: ID of the release to upload to
:param file: Path of the file to upload
:param label: Custom label for this file
:return: The status of the request
"""
url = f"{Gitea.api_url()}/repos/{owner}/{repo}/releases/{release_id}/assets"
try:
name = os.path.basename(file)
response = Gitea.session().post(
url,
params={"name": name},
data={},
files={
"attachment": (
name,
open(file, "rb"),
"application/octet-stream",
),
},
)
logger.debug(
f"Asset upload on Gitea completed, url: {response.url}, status code: {response.status_code}"
)
return True
except HTTPError as e:
logger.warning(f"Asset upload {file} on Gitea has failed: {e}")
return False
[docs] @classmethod
def upload_dists(cls, owner: str, repo: str, version: str, path: str) -> bool:
"""Upload distributions to a release
:param owner: The owner namespace of the repository
:param repo: The repository name
:param version: Version to upload for
:param path: Path to the dist directory
:return: The status of the request
"""
# Find the release corresponding to this version
release_id = Gitea.get_release(owner, repo, get_formatted_tag(version))
if not release_id:
logger.debug("No release found to upload assets to")
return False
# Upload assets
one_or_more_failed = False
for file in os.listdir(path):
file_path = os.path.join(path, file)
if not Gitea.upload_asset(owner, repo, release_id, file_path):
one_or_more_failed = True
return not one_or_more_failed
[docs]class Gitlab(Base):
"""Gitlab helper class"""
[docs] @staticmethod
def domain() -> str:
"""Gitlab domain property
:return: The Gitlab instance domain
"""
# Use Gitlab-CI environment vars if available
if "CI_SERVER_URL" in os.environ:
url = urlsplit(os.environ["CI_SERVER_URL"])
return f"{url.netloc}{url.path}".rstrip("/")
domain = config.get("hvcs_domain", os.environ.get("CI_SERVER_HOST", None))
return domain if domain else "gitlab.com"
[docs] @staticmethod
def api_url() -> str:
"""Gitlab api_url property
:return: The Gitlab instance API url
"""
# Use Gitlab-CI environment vars if available
if "CI_SERVER_URL" in os.environ:
return os.environ["CI_SERVER_URL"]
return f"https://{Gitlab.domain()}"
[docs] @staticmethod
def token() -> Optional[str]:
"""Gitlab token property
:return: The Gitlab token environment variable (GL_TOKEN) value
"""
return os.environ.get(config.get("gitlab_token_var"))
[docs] @staticmethod
@LoggedFunction(logger)
def check_build_status(owner: str, repo: str, ref: str) -> bool:
"""Check last build status
:param owner: The owner namespace of the repository. It includes all groups and subgroups.
:param repo: The repository name
:param ref: The sha1 hash of the commit ref
:return: the status of the pipeline (False if a job failed)
"""
gl = gitlab.Gitlab(Gitlab.api_url(), private_token=Gitlab.token())
gl.auth()
jobs = gl.projects.get(owner + "/" + repo).commits.get(ref).statuses.list()
for job in jobs:
if job["status"] not in ["success", "skipped"]: # type: ignore[index]
if job["status"] == "pending": # type: ignore[index]
logger.debug(
f"check_build_status: job {job['name']} is still in pending status" # type: ignore[index]
)
return False
elif job["status"] == "failed" and not job["allow_failure"]: # type: ignore[index]
logger.debug(f"check_build_status: job {job['name']} failed") # type: ignore[index]
return False
return True
[docs] @classmethod
@LoggedFunction(logger)
def post_release_changelog(
cls, owner: str, repo: str, version: str, changelog: str
) -> bool:
"""Post release changelog
:param owner: The owner namespace of the repository
:param repo: The repository name
:param version: The version number
:param changelog: The release notes for this version
:return: The status of the request
"""
ref = get_formatted_tag(version)
gl = gitlab.Gitlab(Gitlab.api_url(), private_token=Gitlab.token())
gl.auth()
try:
logger.debug(f"Before release create call")
gl.projects.get(owner + "/" + repo).releases.create(
{
"name": "Release " + version,
"tag_name": ref,
"description": changelog,
}
)
except exceptions.GitlabCreateError:
logger.debug(
f"Release {ref} could not be created for project {owner}/{repo}"
)
return False
return True
[docs]@LoggedFunction(logger)
def get_hvcs() -> Base:
"""Get HVCS helper class
:raises ImproperConfigurationError: if the hvcs option provided is not valid
"""
hvcs = config.get("hvcs")
try:
return globals()[hvcs.capitalize()]
except KeyError:
raise ImproperConfigurationError(
'"{0}" is not a valid option for hvcs. Please use "github"|"gitlab"|"gitea"'.format(
hvcs
)
)
[docs]def check_build_status(owner: str, repository: str, ref: str) -> bool:
"""
Checks the build status of a commit on the api from your hosted version control provider.
:param owner: The owner of the repository
:param repository: The repository name
:param ref: Commit or branch reference
:return: A boolean with the build status
"""
logger.debug(f"Checking build status for {owner}/{repository}#{ref}")
return get_hvcs().check_build_status(owner, repository, ref)
[docs]def post_changelog(owner: str, repository: str, version: str, changelog: str) -> bool:
"""
Posts the changelog to the current hvcs release API
:param owner: The owner of the repository
:param repository: The repository name
:param version: A string with the new version
:param changelog: A string with the changelog in correct format
:return: a tuple with success status and payload from hvcs
"""
logger.debug(f"Posting release changelog for {owner}/{repository} {version}")
return get_hvcs().post_release_changelog(owner, repository, version, changelog)
[docs]def upload_to_release(owner: str, repository: str, version: str, path: str) -> bool:
"""
Upload distributions to the current hvcs release API
:param owner: The owner of the repository
:param repository: The repository name
:param version: A string with the version to upload for
:param path: Path to dist directory
:return: Status of the request
"""
return get_hvcs().upload_dists(owner, repository, version, path)
[docs]def get_token() -> Optional[str]:
"""
Returns the token for the current VCS
:return: The token in string form
"""
return get_hvcs().token()
[docs]def get_domain() -> Optional[str]:
"""
Returns the domain for the current VCS
:return: The domain in string form
"""
return get_hvcs().domain()
[docs]def check_token() -> bool:
"""
Checks whether there exists a token or not.
:return: A boolean telling if there is a token.
"""
return get_hvcs().token() is not None