Decouple ArtifactsHelper commit status

This commit is contained in:
Mikhail f. Shiryaev 2023-09-12 21:35:30 +02:00
parent 40e04fee6c
commit 414594a472
No known key found for this signature in database
GPG Key ID: 4B02ED204C7D93F4
2 changed files with 23 additions and 4 deletions

View File

@ -7,7 +7,7 @@ from fnmatch import fnmatch
from os import path as op
from pathlib import Path
from shutil import copy2
from typing import List, Union
from typing import List, Optional, Union
from github.Commit import Commit
@ -50,6 +50,7 @@ class ArtifactsHelper:
# The s3 prefix is done with trailing slash!
self._s3_prefix = op.join(s3_prefix, self.commit, "")
self._s3_index_key = f"{self.s3_prefix}{self.INDEX}"
self._s3_index_url = None # type: Optional[str]
@property
def commit(self) -> str:
@ -68,6 +69,14 @@ class ArtifactsHelper:
"""Prefix with the trailing slash"""
return self._s3_index_key
@property
def s3_index_url(self) -> str:
if self._s3_index_url is None:
self._s3_index_url = self.s3_helper.get_url(
S3_BUILDS_BUCKET, self.s3_index_key
)
return self._s3_index_url
def upload(self, artifact_name: str, artifact_path: Path) -> None:
"""Creates archive 'artifact_name.tar{compress_files.SUFFIX} with directory of"""
assert not any(s in artifact_name for s in self.RESTRICTED_SYMBOLS)
@ -118,6 +127,12 @@ class ArtifactsHelper:
)
return list(results)
@staticmethod
def post_commit_status(commit: Commit, url: str) -> None:
post_commit_status(
commit, "success", url, "Artifacts for workflow", "Artifacts"
)
def _regenerate_index(self) -> None:
objects = self.s3_helper.client.list_objects_v2(
Bucket=S3_BUILDS_BUCKET, Prefix=self.s3_prefix
@ -150,6 +165,4 @@ class ArtifactsHelper:
index_path.write_text(index_content, encoding="utf-8")
url = self.s3_helper.upload_build_file_to_s3(index_path, self.s3_index_key)
if isinstance(self._commit, Commit):
post_commit_status(
self._commit, "success", url, "Artifacts for workflow", "Artifacts"
)
self.post_commit_status(self._commit, url)

View File

@ -305,6 +305,12 @@ class S3Helper:
except Exception:
return ""
@staticmethod
def get_url(bucket: str, key: str) -> str:
if CI:
return S3Helper.s3_url(bucket, key)
return S3Helper.local_path(bucket, key).as_uri()
@staticmethod
def s3_url(bucket: str, key: str) -> str:
url = f"{S3_DOWNLOAD}/{bucket}/{key}"