opsml.registry.semver

  1# Copyright (c) Shipt, Inc.
  2# This source code is licensed under the MIT license found in the
  3# LICENSE file in the root directory of this source tree.
  4import re
  5from enum import Enum
  6from typing import Any, List, Optional
  7
  8import semver
  9from pydantic import BaseModel, model_validator
 10
 11from opsml.helpers.exceptions import VersionError
 12from opsml.helpers.logging import ArtifactLogger
 13
 14logger = ArtifactLogger.get_logger()
 15
 16
 17class VersionType(str, Enum):
 18    MAJOR = "major"
 19    MINOR = "minor"
 20    PATCH = "patch"
 21    PRE = "pre"
 22    BUILD = "build"
 23    PRE_BUILD = "pre_build"
 24
 25    @staticmethod
 26    def from_str(name: str) -> "VersionType":
 27        l_name = name.strip().lower()
 28        if l_name == "major":
 29            return VersionType.MAJOR
 30        if l_name == "minor":
 31            return VersionType.MINOR
 32        if l_name == "patch":
 33            return VersionType.PATCH
 34        if l_name == "pre":
 35            return VersionType.PRE
 36        if l_name == "build":
 37            return VersionType.BUILD
 38        if l_name == "pre_build":
 39            return VersionType.PRE_BUILD
 40        raise NotImplementedError()
 41
 42
 43class CardVersion(BaseModel):
 44    version: str
 45    version_splits: List[str] = []
 46    is_full_semver: bool = False
 47
 48    @model_validator(mode="before")
 49    @classmethod
 50    def validate_inputs(cls, values: Any) -> Any:
 51        """Validates a user-supplied version"""
 52        version = values.get("version")
 53        splits = version.split(".")
 54        values["version_splits"] = splits
 55
 56        if cls.check_full_semver(splits):
 57            values["is_full_semver"] = True
 58            cls._validate_full_semver(version)
 59        else:
 60            values["is_full_semver"] = False
 61            cls._validate_partial_semver(splits)
 62
 63        return values
 64
 65    @classmethod
 66    def check_full_semver(cls, version_splits: List[str]) -> bool:
 67        """Checks if a version is a full semver"""
 68        return len(version_splits) >= 3
 69
 70    @classmethod
 71    def _validate_full_semver(cls, version: str) -> None:
 72        """Validates a full semver"""
 73        if not semver.VersionInfo.isvalid(version):
 74            raise ValueError("Version is not a valid Semver")
 75
 76    @classmethod
 77    def _validate_partial_semver(cls, version_splits: List[str]) -> None:
 78        """Validates a partial semver"""
 79        try:
 80            assert all((i.isdigit() for i in version_splits))
 81        except AssertionError as exc:
 82            version = ".".join(version_splits)
 83            raise AssertionError(f"Version {version} is not a valid semver or partial semver") from exc
 84
 85    def _get_version_split(self, split: int) -> str:
 86        """Splits a version into its major, minor, and patch components"""
 87
 88        try:
 89            return self.version_splits[split]
 90        except IndexError as exc:
 91            raise IndexError(f"Version split {split} not found: {self.version}") from exc
 92
 93    @property
 94    def has_major_minor(self) -> bool:
 95        """Checks if a version has a major and minor component"""
 96        return len(self.version_splits) >= 2
 97
 98    @property
 99    def major(self) -> str:
100        return self._get_version_split(0)
101
102    @property
103    def minor(self) -> str:
104        return self._get_version_split(1)
105
106    @property
107    def patch(self) -> str:
108        return self._get_version_split(2)
109
110    @property
111    def valid_version(self) -> str:
112        if self.is_full_semver:
113            return str(semver.VersionInfo.parse(self.version).finalize_version())
114        return self.version
115
116    @staticmethod
117    def finalize_partial_version(version: str) -> str:
118        """Finalizes a partial semver version
119
120        Args:
121            version:
122                version to finalize
123        Returns:
124            str: finalized version
125        """
126        version_splits = version.split(".")
127
128        if len(version_splits) == 1:
129            return f"{version}.0.0"
130        if len(version_splits) == 2:
131            return f"{version}.0"
132
133        return version
134
135    def get_version_to_search(self, version_type: VersionType) -> Optional[str]:
136        """Gets a version to search for in the database
137
138        Args:
139            version_type:
140                type of version to search for
141        Returns:
142            str: version to search for
143        """
144
145        if version_type == VersionType.PATCH:  # want to search major and minor if exists
146            if self.has_major_minor:
147                return f"{self.major}.{self.minor}"
148            return str(self.major)
149
150        if version_type == VersionType.MINOR:  # want to search major
151            return str(self.major)
152
153        if version_type in [VersionType.PRE, VersionType.BUILD, VersionType.PRE_BUILD]:
154            return self.valid_version
155
156        return None
157
158
159class SemVerUtils:
160    """Class for general semver-related functions"""
161
162    @staticmethod
163    def sort_semvers(versions: List[str]) -> List[str]:
164        """Implements bubble sort for semvers
165
166        Args:
167            versions:
168                list of versions to sort
169
170        Returns:
171            sorted list of versions with highest version first
172        """
173
174        n_ver = len(versions)
175
176        for i in range(n_ver):
177            already_sorted = True
178
179            for j in range(n_ver - i - 1):
180                j_version = semver.VersionInfo.parse(versions[j])
181                j1_version = semver.VersionInfo.parse(versions[j + 1])
182
183                # use semver comparison logic
184                if j_version > j1_version:
185                    # swap
186                    versions[j], versions[j + 1] = versions[j + 1], versions[j]
187
188                    already_sorted = False
189
190            if already_sorted:
191                break
192
193        versions.reverse()
194        return versions
195
196    @staticmethod
197    def is_release_candidate(version: str) -> bool:
198        """Ignores pre-release versions"""
199        ver = semver.VersionInfo.parse(version)
200        return bool(ver.prerelease)
201
202    @staticmethod
203    def increment_version(
204        version: str,
205        version_type: VersionType,
206        pre_tag: str,
207        build_tag: str,
208    ) -> str:
209        """
210        Increments a version based on version type
211
212        Args:
213            version:
214                Current version
215            version_type:
216                Type of version increment.
217            pre_tag:
218                Pre-release tag
219            build_tag:
220                Build tag
221
222        Raises:
223            ValueError:
224                unknown version_type
225
226        Returns:
227            New version
228        """
229        ver: semver.VersionInfo = semver.VersionInfo.parse(version)
230
231        # Set major, minor, patch
232        if version_type == VersionType.MAJOR:
233            return str(ver.bump_major())
234        if version_type == VersionType.MINOR:
235            return str(ver.bump_minor())
236        if version_type == VersionType.PATCH:
237            return str(ver.bump_patch())
238
239        # Set pre-release
240        if version_type == VersionType.PRE:
241            return str(ver.bump_prerelease(token=pre_tag))
242
243        # Set build
244        if version_type == VersionType.BUILD:
245            return str(ver.bump_build(token=build_tag))
246
247        if version_type == VersionType.PRE_BUILD:
248            ver = ver.bump_prerelease(token=pre_tag)
249            ver = ver.bump_build(token=build_tag)
250
251            return str(ver)
252
253        raise ValueError(f"Unknown version_type: {version_type}")
254
255    @staticmethod
256    def add_tags(
257        version: str,
258        pre_tag: Optional[str] = None,
259        build_tag: Optional[str] = None,
260    ) -> str:
261        if pre_tag is not None:
262            version = f"{version}-{pre_tag}"
263        if build_tag is not None:
264            version = f"{version}+{build_tag}"
265
266        return version
267
268
269class SemVerRegistryValidator:
270    """Class for obtaining the correct registry version"""
271
272    def __init__(
273        self,
274        name: str,
275        version_type: VersionType,
276        pre_tag: str,
277        build_tag: str,
278        version: Optional[CardVersion] = None,
279    ) -> None:
280        """Instantiate SemverValidator
281
282        Args:
283            name:
284                name of the artifact
285            version_type:
286                type of version increment
287            version:
288                version to use
289            pre_tag:
290                pre-release tag
291            build_tag:
292                build tag
293
294        Returns:
295            None
296        """
297        self.version = version
298        self._version_to_search = None
299        self.final_version = None
300        self.version_type = version_type
301        self.name = name
302        self.pre_tag = pre_tag
303        self.build_tag = build_tag
304
305    @property
306    def version_to_search(self) -> Optional[str]:
307        """Parses version and returns version to search for in the registry"""
308        if self.version is not None:
309            return self.version.get_version_to_search(version_type=self.version_type)
310        return self._version_to_search
311
312    def _set_version_from_existing(self, versions: List[str]) -> str:
313        """Search existing versions to find the correct version to use
314
315        Args:
316            versions:
317                list of existing versions
318
319        Returns:
320            str: version to use
321        """
322        version = versions[0]
323        recent_ver = semver.VersionInfo.parse(version)
324        # first need to check if increment is mmp
325        if self.version_type in [VersionType.MAJOR, VersionType.MINOR, VersionType.PATCH]:
326            # check if most recent version is a pre-release or build
327            if recent_ver.prerelease is not None:
328                version = str(recent_ver.finalize_version())
329                try:
330                    # if all versions are pre-release use finalized version
331                    # if not, increment version
332                    for ver in versions:
333                        parsed_ver = semver.VersionInfo.parse(ver)
334                        if parsed_ver.prerelease is None:
335                            raise VersionError("Major, minor and patch version combination already exists")
336                    return version
337                except VersionError:
338                    logger.info("Major, minor and patch version combination already exists")
339
340        while version in versions:
341            version = SemVerUtils.increment_version(
342                version=version,
343                version_type=self.version_type,
344                pre_tag=self.pre_tag,
345                build_tag=self.build_tag,
346            )
347
348        return version
349
350    def set_version(self, versions: List[str]) -> str:
351        """Sets the correct version to use for incrementing and adding the the registry
352
353        Args:
354            versions:
355                list of existing versions
356
357        Returns:
358            str: version to use
359        """
360        if bool(versions):
361            return self._set_version_from_existing(versions=versions)
362
363        final_version = None
364        if self.version is not None:
365            final_version = CardVersion.finalize_partial_version(version=self.version.valid_version)
366
367        version = final_version or "1.0.0"
368
369        if self.version_type in [VersionType.PRE, VersionType.BUILD, VersionType.PRE_BUILD]:
370            return SemVerUtils.increment_version(
371                version=version,
372                version_type=self.version_type,
373                pre_tag=self.pre_tag,
374                build_tag=self.build_tag,
375            )
376
377        return version
378
379
380class SemVerSymbols(str, Enum):
381    STAR = "*"
382    CARET = "^"
383    TILDE = "~"
384
385
386class SemVerParser:
387    """Base class for semver parsing"""
388
389    @staticmethod
390    def parse_version(version: str) -> str:
391        raise NotImplementedError
392
393    @staticmethod
394    def validate(version: str) -> bool:
395        raise NotImplementedError
396
397
398class StarParser(SemVerParser):
399    """Parses versions that contain * symbol"""
400
401    @staticmethod
402    def parse_version(version: str) -> str:
403        version_ = version.split(SemVerSymbols.STAR)[0]
404        return re.sub(".$", "", version_)
405
406    @staticmethod
407    def validate(version: str) -> bool:
408        return SemVerSymbols.STAR in version
409
410
411class CaretParser(SemVerParser):
412    """Parses versions that contain ^ symbol"""
413
414    @staticmethod
415    def parse_version(version: str) -> str:
416        return version.split(".")[0].replace(SemVerSymbols.CARET, "")
417
418    @staticmethod
419    def validate(version: str) -> bool:
420        return SemVerSymbols.CARET in version
421
422
423class TildeParser(SemVerParser):
424    """Parses versions that contain ~ symbol"""
425
426    @staticmethod
427    def parse_version(version: str) -> str:
428        return ".".join(version.split(".")[0:2]).replace(SemVerSymbols.TILDE, "")
429
430    @staticmethod
431    def validate(version: str) -> bool:
432        return SemVerSymbols.TILDE in version
433
434
435class NoParser(SemVerParser):
436    """Does not parse version"""
437
438    @staticmethod
439    def parse_version(version: str) -> str:
440        return version
441
442    @staticmethod
443    def validate(version: str) -> bool:
444        return version not in list(SemVerSymbols)
445
446
447def get_version_to_search(version: str) -> str:
448    """Parses a current version based on SemVer characters.
449
450    Args:
451        version (str): ArtifactCard version
452
453    Returns:
454        Version (str) to search based on presence of SemVer characters
455    """
456
457    # gut check
458    if sum(symbol in version for symbol in SemVerSymbols) > 1:
459        raise ValueError("Only one SemVer character is allowed in the version string")
460
461    parser = next(
462        (parser for parser in SemVerParser.__subclasses__() if parser.validate(version=version)),
463        NoParser,
464    )
465    return parser.parse_version(version=version)