import pickle import re from argparse import ArgumentParser from dataclasses import replace from os import path from tempfile import gettempdir from typing import Optional import yaml from release import versioning from release.context import ReleaseContext from release.project import (ArtefactDescription, DeploymentCondition, DeploymentDescription, HelmRelease, Npm, OciImage, ProjectDescription, Sdist, Tarball, Wheel, parse_project_description) from release.release import (create_release, publish_artefacts, update_deployments) def load_project_description(filename) -> ProjectDescription: with open(filename, 'rb') as f: return pickle.load(f) def save_project_description(filename, project_description): with open(filename, 'wb') as f: pickle.dump(project_description, f) def make_project_description( version_descriptor_filename: str) -> ProjectDescription: return ProjectDescription(version_descriptor=version_descriptor_filename) def make_context(gitea_instance: str, repository_name: str, ref_name: str, run_number: str, commit_sha: str, is_pre_release: bool) -> ReleaseContext: return ReleaseContext(gitea_instance=gitea_instance, repository_name=repository_name, ref_name=ref_name, run_number=run_number, commit_sha=commit_sha, is_pre_release=is_pre_release) def dump_project_description(project_description: ProjectDescription): print('version_descriptor: %s' % project_description.version_descriptor) print('project version: %s' % project_description.project_version) print('planned version: %s' % project_description.planned_version) print('is already released: %s' % project_description.is_released) print('') print('artefacts:') for artefact in project_description.artefacts: generated = artefact.generated version_descriptor = ( artefact.version_descriptor or project_description.version_descriptor) release_info = generated.make_release_info( project_description.context, project_description.planned_version) if isinstance(generated, OciImage): print(' - oci image: %s' % generated.name) print(' repository: %s' % generated.repository) print(' local tag: %s' % release_info.local_tag) print(' local name: %s' % release_info.local_full_name) print(' remote names:') for name in release_info.remote_full_names: print(' - %s' % name) print(' docker tags: %s' % ", ".join(release_info.tags)) elif isinstance(generated, Tarball): print(' - tarball: %s' % generated.filename) print(' package name: %s' % generated.package_name) print(' repository: %s' % generated.repository) print(' release version name: %s' % release_info.version_str) elif isinstance(generated, Wheel): print(' - wheel: %s' % generated.pattern) print(' repository: %s' % generated.repository) print(' release version name: %s' % release_info.version_str) elif isinstance(generated, Sdist): print(' - sdist: %s' % generated.filename) print(' repository: %s' % generated.repository) print(' release version name: %s' % release_info.version_str) elif isinstance(generated, Npm): print(' - npm: %s' % generated.directory) print(' version_descriptor: %s' % version_descriptor) print(' artefact version: %s' % versioning.use_any(version_descriptor).version) print('') print('deployments:') for desc in project_description.deployments: deployment = desc.deployment condition = desc.condition if condition == DeploymentCondition.ALWAYS: condition_str = 'always' elif condition == DeploymentCondition.NEVER: condition_str = 'never' elif condition == DeploymentCondition.PRE_RELEASE_ONLY: condition_str = 'pre_release_only' elif condition == DeploymentCondition.RELEASE_ONLY: condition_str = 'release_only' print(' - condition: %s' % condition_str) if isinstance(deployment, HelmRelease): print(' type: helm release') print(' release name: %s' % deployment.release_name) print(' image paths: %s' % deployment.image_paths) print(' namespace: %s' % deployment.namespace) print(' repository: %s' % deployment.repository) print('') print('context:') context = project_description.context print(' gitea instance: %s' % context.gitea_instance) print(' repository name: %s' % context.repository_name) print(' ref name: %s' % context.ref_name) print(' run number: %s' % context.run_number) print(' commit sha: %s' % context.commit_sha) print(' is pre-release: %s' % context.is_pre_release) print('') print('release info:') release_info = project_description.release_info print(' title: %s' % release_info.gitea_release_title) print(' description: %s' % release_info.gitea_release_description) print(' is prerelease: %s' % release_info.gitea_is_prerelease) print(' git target commitish: %s' % release_info.gitea_git_commitish) print(' git tags: %s' % ', '.join(release_info.git_tags)) print('') print('environment variables:') for key, value in project_description.environment_variables.items(): print(' %s: %s' % (key, value)) def make_artefact(type: str, repository: str, name: str, filename: str, package_name: str, pattern: str, directory: str, version_descriptor) -> ArtefactDescription: maybe_repository = ({'repository': repository} if repository is not None else {}) if type == 'oci_image': assert name is not None generated = OciImage(name=name, **maybe_repository) elif type == 'tarball': assert filename is not None assert package_name is not None generated = Tarball(filename=filename, package_name=package_name, **maybe_repository) elif type == 'wheel': assert pattern is not None generated = Wheel(pattern=pattern, **maybe_repository) elif type == 'sdist': assert filename is not None generated = Sdist(filename=filename, **maybe_repository) elif type == 'npm': assert directory is not None generated = Npm(directory=directory) else: raise Exception('unknown artefact type: %s' % type) return ArtefactDescription( generated=generated, version_descriptor=version_descriptor) def make_deployment(type: str, release_name: str, image_paths: list[str], namespace: Optional[str], repository: Optional[str], condition_str: Optional[str]) -> DeploymentDescription: if type == 'helm_release': maybe_image_paths = ({'image_paths': image_paths} if len(image_paths) > 0 else {}) maybe_repository = ({'repository': repository} if repository is not None else {}) deployment = HelmRelease(release_name=release_name, namespace=(namespace or release_name), **maybe_image_paths, **maybe_repository) else: raise Exception('unknown deployment type: %s' % type) if condition_str == 'always': condition = DeploymentCondition.ALWAYS elif condition_str == 'never': condition = DeploymentCondition.NEVER elif condition_str == 'pre_release_only': condition = DeploymentCondition.PRE_RELEASE_ONLY elif condition_str == 'release_only': condition = DeploymentCondition.RELEASE_ONLY elif condition_str is None: condition = None else: raise Exception('unknown condition: %s' % condition_str) maybe_condition = ({'condition': condition} if condition is not None else {}) return DeploymentDescription(deployment=deployment, **maybe_condition) def sync_versions(project_description: ProjectDescription): planned_version = project_description.planned_version def sync(descriptor_filename: str): v = versioning.use_any(descriptor_filename) v.version = planned_version v.store() sync(project_description.version_descriptor) for artefact in project_description.artefacts: if artefact.version_descriptor is not None: sync(artefact.version_descriptor) if __name__ == '__main__': parser = ArgumentParser() parser.add_argument('action', choices=[ 'declare', 'check', 'add-artefact', 'add-deployment', 'sync-versions', 'publish-artefacts', 'update-deployments', 'create-release', 'dump' ]) def nullable_string(val): if not val: return None return val def space_separated(val): return list(filter(lambda it: it != '', (val or '').split(' '))) def true_or_false(val): if val == '0' or val == 'false': return False elif val == '1' or val == 'true': return True else: raise Exception('flag can be "0" or "1". got: %s' % val) parser.add_argument('--state', type=nullable_string) parser.add_argument('--version-descriptor', type=nullable_string) parser.add_argument('--release-yaml-filename', type=nullable_string) parser.add_argument('--dry-run', type=true_or_false) parser.add_argument('--gitea-instance') parser.add_argument('--release-repository-name') parser.add_argument('--release-ref-name') parser.add_argument('--release-run-number') parser.add_argument('--release-commit-sha') parser.add_argument('--is-pre-release', type=true_or_false) parser.add_argument( '--artefact-type', type=nullable_string, choices=['oci_image', 'tarball', 'wheel', 'sdist', 'npm']) parser.add_argument('--artefact-repository', type=nullable_string) parser.add_argument('--artefact-name', type=nullable_string) parser.add_argument('--artefact-package-name', type=nullable_string) parser.add_argument('--artefact-filename', type=nullable_string) parser.add_argument('--artefact-pattern', type=nullable_string) parser.add_argument('--artefact-directory', type=nullable_string) parser.add_argument('--deployment-type', type=nullable_string, choices=['helm_release']) parser.add_argument('--deployment-release-name', type=nullable_string) parser.add_argument('--deployment-image-paths', type=space_separated) parser.add_argument('--deployment-namespace', type=nullable_string) parser.add_argument('--deployment-repository', type=nullable_string) parser.add_argument('--deployment-condition', type=nullable_string) parser.add_argument('--write-env-vars-to-filename') args = parser.parse_args() state_file = (args.state or str(path.join(gettempdir(), 'release_project_state'))) def clean_repository_name(name: str) -> str: if name.startswith('//'): return re.match('^[/][/][^/]+[/](.+)$', name).group(1) else: return name if args.action == 'declare': if args.version_descriptor is None: with open(args.release_yaml_filename, 'r') as f: project_description = parse_project_description( yaml.safe_load(f)) else: project_description = make_project_description( args.version_descriptor) if args.is_pre_release is None: assert args.release_ref_name is not None is_pre_release = not any( map(lambda rn: rn in args.release_ref_name, ['master', 'main'])) else: is_pre_release = args.is_pre_release project_description = replace( project_description, context=make_context( args.gitea_instance, clean_repository_name(args.release_repository_name), args.release_ref_name, args.release_run_number, args.release_commit_sha, is_pre_release)) save_project_description(state_file, project_description) elif args.action == 'add-artefact': project_description = load_project_description(state_file) artefact = make_artefact(args.artefact_type, args.artefact_repository, args.artefact_name, args.artefact_filename, args.artefact_package_name, args.artefact_pattern, args.artefact_directory, args.version_descriptor) project_description = replace( project_description, artefacts=project_description.artefacts + [artefact]) save_project_description(state_file, project_description) elif args.action == 'add-deployment': project_description = load_project_description(state_file) deployment = make_deployment(args.deployment_type, args.deployment_release_name, args.deployment_image_paths, args.deployment_namespace, args.deployment_repository, args.deployment_condition) project_description = replace( project_description, deployments=project_description.deployments + [deployment]) save_project_description(state_file, project_description) elif args.action == 'sync-versions': project_description = load_project_description(state_file) sync_versions(project_description) elif args.action == 'publish-artefacts': project_description = load_project_description(state_file) publish_artefacts(project_description, args.dry_run) elif args.action == 'update-deployments': project_description = load_project_description(state_file) update_deployments(project_description, args.dry_run) elif args.action == 'create-release': project_description = load_project_description(state_file) create_release(project_description, args.dry_run) elif args.action == 'dump': project_description = load_project_description(state_file) dump_project_description(project_description) elif args.action == 'check': project_description = load_project_description(state_file) else: raise NotImplementedError() assert project_description is not None assert state_file is not None env_var_filename = args.write_env_vars_to_filename if env_var_filename is not None: env_vars = { 'RELEASE_ACTION_STATEFILE': state_file, **project_description.environment_variables } assert not any(map(lambda v: '"' in v, env_vars.values())) with open(env_var_filename, 'a') as f: f.write( '\n'.join(map(lambda it: '%s=%s' % it, env_vars.items())))