Hide navigation sidebar
Hide table of contents sidebar
Skip to content
Back to top
Toggle table of contents sidebar

# Source code for semantic_release.version.algorithm

    from __future__ import annotations

    import logging
    from contextlib import suppress
    from functools import reduce
    from queue import LifoQueue
    from typing import TYPE_CHECKING, Iterable

    from semantic_release.commit_parser import ParsedCommit
    from semantic_release.commit_parser.token import ParseError
    from semantic_release.const import DEFAULT_VERSION
    from semantic_release.enums import LevelBump, SemanticReleaseLogLevels
    from semantic_release.errors import InternalError, InvalidVersion
    from semantic_release.helpers import validate_types_in_sequence

    if TYPE_CHECKING:  # pragma: no cover
        from typing import Sequence

        from git.objects.commit import Commit
        from git.refs.tag import Tag
        from git.repo.base import Repo

        from semantic_release.commit_parser import (
            CommitParser,
            ParseResult,
            ParserOptions,
        )
        from semantic_release.version.translator import VersionTranslator
        from semantic_release.version.version import Version

    logger = logging.getLogger(__name__)

    def tags_and_versions(
        tags: Iterable[Tag], translator: VersionTranslator
    ) -> list[tuple[Tag, Version]]:
        """
        Return a list of 2-tuples, where each element is a tuple (tag, version)
        from the tags in the Git repo and their corresponding `Version` according
        to `Version.from_tag`. The returned list is sorted according to semver
        ordering rules.

        Tags which are not matched by `translator` are ignored.
        """
        ts_and_vs: list[tuple[Tag, Version]] = []
        for tag in tags:
            try:
                version = translator.from_tag(tag.name)
            except (NotImplementedError, InvalidVersion) as e:
                logger.warning(
                    "Couldn't parse tag %s as as Version: %s",
                    tag.name,
                    str(e),
                    exc_info=logger.isEnabledFor(logging.DEBUG),
                )
                continue

            if version:
                ts_and_vs.append((tag, version))

        logger.info("found %s previous tags", len(ts_and_vs))
        return sorted(ts_and_vs, reverse=True, key=lambda v: v[1])

    def _traverse_graph_for_commits(
        head_commit: Commit,
        latest_release_tag_str: str = "",
    ) -> Sequence[Commit]:
        # Depth-first search
        def dfs(start_commit: Commit, stop_nodes: set[Commit]) -> Sequence[Commit]:
            # Create a stack for DFS
            stack: LifoQueue[Commit] = LifoQueue()

            # Create a set to store visited graph nodes (commit objects in this case)
            visited: set[Commit] = set()

            # Initialize the result
            commits: list[Commit] = []

            # Add the source node in the queue to start the search
            stack.put(start_commit)

            # Traverse the git history capturing each commit found before it reaches a stop node
            while not stack.empty():
                if (node := stack.get()) in visited or node in stop_nodes:
                    continue

                visited.add(node)
                commits.append(node)

                # Add all parent commits to the stack from left to right so that the rightmost is popped first
                # as the left side is generally the merged into branch
                for parent in node.parents:
                    stack.put(parent)

            return commits

        # Run a Depth First Search to find all the commits since the last release
        return dfs(
            start_commit=head_commit,
            stop_nodes=set(
                head_commit.repo.iter_commits(latest_release_tag_str)
                if latest_release_tag_str
                else []
            ),
        )

    def _increment_version(
        latest_version: Version,
        latest_full_version: Version,
        level_bump: LevelBump,
        prerelease: bool,
        prerelease_token: str,
        major_on_zero: bool,
        allow_zero_version: bool,
    ) -> Version:
        """
        Using the given versions, along with a given `level_bump`, increment to
        the next version according to whether or not this is a prerelease.

        `latest_version` is the most recent version released from this branch's history.
        `latest_full_version`, the most recent full release (i.e. not a prerelease)
        in this branch's history.

        `latest_version` and `latest_full_version` can be the same, but aren't necessarily.
        """
        local_vars = list(locals().items())
        logger.log(
            SemanticReleaseLogLevels.SILLY,
            "_increment_version: %s",
            str.join(", ", [f"{k} = {v}" for k, v in local_vars]),
        )

        # Handle variations where the latest version is 0.x.x
        if latest_version.major == 0:
            if not allow_zero_version:
                # Set up default version to be 1.0.0 if currently 0.x.x which means a commented
                # breaking change is not required to bump to 1.0.0
                logger.debug(
                    "Bumping major version as 0.x.x versions are disabled because of allow_zero_version=False"
                )
                level_bump = LevelBump.MAJOR

            elif not major_on_zero:
                # if we are a 0.x.y release and have set `major_on_zero`,
                # breaking changes should increment the minor digit
                # Correspondingly, we reduce the level that we increment the
                # version by.
                logger.debug(
                    "reducing version increment due to 0. version and major_on_zero=False"
                )

                level_bump = min(level_bump, LevelBump.MINOR)

        logger.debug(
            "prerelease=%s and the latest version %s %s prerelease",
            prerelease,
            latest_version,
            "is a" if latest_version.is_prerelease else "is not a",
        )

        if level_bump == LevelBump.NO_RELEASE:
            raise ValueError("level_bump must be at least PRERELEASE_REVISION")

        if level_bump == LevelBump.PRERELEASE_REVISION and not latest_version.is_prerelease:
            raise ValueError(
                "Cannot increment a non-prerelease version with a prerelease level bump"
            )

        # assume we always want to increment the version that is the latest in the branch's history
        base_version = latest_version

        # if the current version is a prerelease & we want a new prerelease, then
        # figure out if we need to bump the prerelease revision or start a new prerelease
        if latest_version.is_prerelease:
            # find the change since the last full release because if the current version is a prerelease
            # then we need to predict properly the next full version
            diff_with_last_released_version = latest_version - latest_full_version
            logger.debug(
                "the diff b/w the latest version '%s' and the latest full release version '%s' is: %s",
                latest_version,
                latest_full_version,
                diff_with_last_released_version,
            )

            # Since the difference is less than or equal to the level bump and we want a new prerelease,
            # we can abort early and just increment the revision
            if level_bump <= diff_with_last_released_version:
                # 6a ii) if level_bump <= the level bump introduced by the previous tag (latest_version)
                if prerelease:
                    logger.debug(
                        "there has already been at least a %s release since the last full release %s",
                        level_bump,
                        latest_full_version,
                    )
                    logger.debug("Incrementing the prerelease revision...")
                    new_revision = base_version.to_prerelease(
                        token=prerelease_token,
                        revision=(
                            1
                            if latest_version.prerelease_token != prerelease_token
                            else (latest_version.prerelease_revision or 0) + 1
                        ),
                    )
                    logger.debug("Incremented %s to %s", base_version, new_revision)
                    return new_revision

                # When we don't want a prerelease, but the previous version is a prerelease that
                # had a greater bump than we currently are applying, choose the larger bump instead
                # as it consumes this bump
                logger.debug("Finalizing the prerelease version...")
                return base_version.finalize_version()

            # Fallthrough to handle all larger level bumps
            logger.debug(
                "this release has a greater bump than any change since the last full release, %s",
                latest_full_version,
            )

            # Fallthrough, if we don't want a prerelease, or if we do but the level bump is greater
            #
            # because the current version is a prerelease, we must start from the last full version
            # Case 1: we identified that the level bump is greater than the change since
            #         the last full release, this will also reset the prerelease revision
            # Case 2: we don't want a prerelease, so consider only the last full version in history
            base_version = latest_full_version

        # From the base version, we can now increment the version according to the level bump
        # regardless of the prerelease status as bump() handles the reset and pass through
        logger.debug("Bumping %s with a %s bump", base_version, level_bump)
        target_next_version = base_version.bump(level_bump)

        # Converting to/from a prerelease if necessary
        target_next_version = (
            target_next_version.to_prerelease(token=prerelease_token)
            if prerelease
            else target_next_version.finalize_version()
        )

        logger.debug("Incremented %s to %s", base_version, target_next_version)
        return target_next_version

    def next_version(
        repo: Repo,
        translator: VersionTranslator,
        commit_parser: CommitParser[ParseResult, ParserOptions],
        prerelease: bool = False,
        major_on_zero: bool = True,
        allow_zero_version: bool = True,
    ) -> Version:
        """
        Evaluate the history within `repo`, and based on the tags and commits in the repo
        history, identify the next semantic version that should be applied to a release
        """
        # Default initial version
        # Since the translator is configured by the user, we can't guarantee that it will
        # be able to parse the default version. So we first cast it to a tag using the default
        # value and the users configured tag format, then parse it back to a version object
        default_initial_version = translator.from_tag(
            translator.str_to_tag(DEFAULT_VERSION)
        )
        if default_initial_version is None:
            # This should never happen, but if it does, it's a bug
            raise InternalError(
                "Translator was unable to parse the embedded default version"
            )

        # Step 1. All tags, sorted descending by semver ordering rules
        all_git_tags_as_versions = tags_and_versions(repo.tags, translator)

        # Retrieve all commit hashes (regardless of merges) in the current branch's history from repo origin
        commit_hash_set = {
            commit.hexsha
            for commit in _traverse_graph_for_commits(head_commit=repo.active_branch.commit)
        }

        # Filter all releases that are not found in the current branch's history
        historic_versions: list[Version] = []
        for tag, version in all_git_tags_as_versions:
            # TODO: move this to tags_and_versions() function?
            # Ignore the error that is raised when tag points to a Blob or Tree object rather
            # than a commit object (tags that point to tags that then point to commits are resolved automatically)
            with suppress(ValueError):
                if tag.commit.hexsha in commit_hash_set:
                    historic_versions.append(version)

        # Step 2. Get the latest final release version in the history of the current branch
        #  or fallback to the default 0.0.0 starting version value if none are found
        latest_full_release_version = next(
            filter(
                lambda version: not version.is_prerelease,
                historic_versions,
            ),
            default_initial_version,
        )

        logger.info(
            f"The last full version in this branch's history was {latest_full_release_version}"
            if latest_full_release_version != default_initial_version
            else "No full releases found in this branch's history"
        )

        # Step 3. Determine the latest release version in the history of the current branch
        # If we the desired result is a prerelease, we must determine if there was any previous
        # prerelease in the history of the current branch beyond the latest_full_release_version.
        # Important to note that, we only consider prereleases that are of the same prerelease token
        # as the basis of incrementing the prerelease revision.
        # If we are not looking for a prerelease, this is the same as the last full release.
        latest_version = (
            latest_full_release_version
            if not prerelease
            else next(
                filter(
                    lambda version: all(
                        [
                            version.is_prerelease,
                            version.prerelease_token == translator.prerelease_token,
                            version >= latest_full_release_version,
                        ]
                    ),
                    historic_versions,
                ),
                latest_full_release_version,  # default
            )
        )

        logger.info("The latest release in this branch's history was %s", latest_version)

        # Step 4. Walk the git tree to find all commits that have been made since the last release
        commits_since_last_release = _traverse_graph_for_commits(
            head_commit=repo.active_branch.commit,
            latest_release_tag_str=(
                # NOTE: the default_initial_version should not actually exist on the repository (ie v0.0.0)
                # so we provide an empty tag string when there are no tags on the repository yet
                latest_version.as_tag() if latest_version != default_initial_version else ""
            ),
        )

        logger.info(
            f"Found {len(commits_since_last_release)} commits since the last release!"
            if len(commits_since_last_release) > 0
            else "No commits found since the last release!"
        )

        # Step 5. apply the parser to each commit in the history (could return multiple results per commit)
        parsed_results = list(map(commit_parser.parse, commits_since_last_release))

        # Step 5A. Validation type check for the parser results (important because of possible custom parsers)
        for parsed_result in parsed_results:
            if not any(
                (
                    isinstance(parsed_result, (ParseError, ParsedCommit)),
                    type(parsed_result) == list
                    and validate_types_in_sequence(
                        parsed_result, (ParseError, ParsedCommit)
                    ),
                    type(parsed_result) == tuple
                    and validate_types_in_sequence(
                        parsed_result, (ParseError, ParsedCommit)
                    ),
                )
            ):
                raise TypeError("Unexpected type returned from commit_parser.parse")

        # Step 5B. Accumulate all parsed results into a single list accounting for possible multiple results per commit
        consolidated_results: list[ParseResult] = reduce(
            lambda accumulated_results, p_results: [
                *accumulated_results,
                *(
                    # Cast to list if not already a list
                    p_results
                    if isinstance(p_results, list) or type(p_results) == tuple
                    else [p_results]
                ),
            ],
            parsed_results,
            [],
        )

        # Step 5C. Parse the commits to determine the bump level that should be applied
        parsed_levels: set[LevelBump] = {
            parsed_result.bump  # type: ignore[union-attr] # too complex for type checkers
            for parsed_result in filter(
                # Filter out any non-ParsedCommit results (i.e. ParseErrors)
                lambda parsed_result: isinstance(parsed_result, ParsedCommit),
                consolidated_results,
            )
        }

        logger.debug(
            "parsed the following distinct levels from the commits since the last release: %s",
            parsed_levels,
        )

        level_bump = max(parsed_levels, default=LevelBump.NO_RELEASE)
        logger.info("The type of the next release release is: %s", level_bump)

        if all(
            [
                level_bump is LevelBump.NO_RELEASE,
                latest_version.major != 0 or allow_zero_version,
            ]
        ):
            logger.info("No release will be made")
            return latest_version

        return _increment_version(
            latest_version=latest_version,
            latest_full_version=latest_full_release_version,
            level_bump=level_bump,
            prerelease=prerelease,
            prerelease_token=translator.prerelease_token,
            major_on_zero=major_on_zero,
            allow_zero_version=allow_zero_version,
        )