|
| 1 | +import re |
| 2 | +from dataclasses import dataclass |
| 3 | +from typing import Any, Literal, Protocol, Self |
| 4 | + |
| 5 | +from pydantic import GetCoreSchemaHandler |
| 6 | +from pydantic_core import CoreSchema, core_schema |
| 7 | + |
| 8 | +from questionpy_common.constants import RE_SEMVER |
| 9 | + |
| 10 | +type _Operator = Literal["==", "!=", ">=", "<=", ">", "<"] |
| 11 | +_OPERATORS: set[_Operator] = {"==", "!=", ">=", "<=", ">", "<"} |
| 12 | + |
| 13 | +_SEMVER_PATTERN = re.compile(RE_SEMVER) |
| 14 | + |
| 15 | + |
| 16 | +class VersionProtocol(Protocol): |
| 17 | + """Partial protocol for SemVer version objects. |
| 18 | +
|
| 19 | + We don't want `questionpy_common` to depend on the `semver` package, so we define this protocol instead of using |
| 20 | + `semver.Version` directly. |
| 21 | + """ |
| 22 | + |
| 23 | + def __gt__(self, other: str) -> bool: ... |
| 24 | + |
| 25 | + def __ge__(self, other: str) -> bool: ... |
| 26 | + |
| 27 | + def __lt__(self, other: str) -> bool: ... |
| 28 | + |
| 29 | + def __le__(self, other: str) -> bool: ... |
| 30 | + |
| 31 | + |
| 32 | +@dataclass(frozen=True) |
| 33 | +class QPyDependencyVersionSpecifier: |
| 34 | + """One or more clauses restricting allowed versions for a QPy package dependency.""" |
| 35 | + |
| 36 | + @dataclass(frozen=True) |
| 37 | + class Clause: |
| 38 | + """A single comparison clause such as `>= 1.2.2`.""" |
| 39 | + |
| 40 | + operator: _Operator |
| 41 | + operand: str |
| 42 | + |
| 43 | + def allows(self, version: VersionProtocol) -> bool: |
| 44 | + """Check if this clause is fulfilled by the given version.""" |
| 45 | + # Note: The semver package we use does already implement a `match` method, but we would like to validate |
| 46 | + # each clause early, before the matching needs to be done. |
| 47 | + match self.operator: |
| 48 | + case "<": |
| 49 | + return version < self.operand |
| 50 | + case "<=": |
| 51 | + return version <= self.operand |
| 52 | + case "==": |
| 53 | + return version == self.operand |
| 54 | + case ">=": |
| 55 | + return version >= self.operand |
| 56 | + case ">": |
| 57 | + return version > self.operand |
| 58 | + case _: |
| 59 | + # Shouldn't be reachable. |
| 60 | + msg = f"Invalid operator: {self.operator}" |
| 61 | + raise ValueError(msg) |
| 62 | + |
| 63 | + @classmethod |
| 64 | + def from_string(cls, string: str) -> Self: |
| 65 | + string = string.strip() |
| 66 | + |
| 67 | + operator = next(filter(string.startswith, _OPERATORS), None) |
| 68 | + if operator: |
| 69 | + version_string = string.removeprefix(operator).lstrip() |
| 70 | + if not _SEMVER_PATTERN.match(string): |
| 71 | + msg = f"Comparison version '{version_string}' of clause '{string}' does not conform to SemVer." |
| 72 | + raise ValueError(msg) |
| 73 | + |
| 74 | + operand = version_string |
| 75 | + else: |
| 76 | + # No operator. Check if string is a version, since we allow "==" to be omitted. |
| 77 | + if not _SEMVER_PATTERN.match(string): |
| 78 | + msg = ( |
| 79 | + f"Version specifier clause '{string}' does not start with a valid operator and isn't a " |
| 80 | + f"version itself. Valid operators are {', '.join(_OPERATORS)}." |
| 81 | + ) |
| 82 | + raise ValueError(msg) |
| 83 | + |
| 84 | + operator = "==" |
| 85 | + operand = string |
| 86 | + |
| 87 | + return cls(operator, operand) |
| 88 | + |
| 89 | + def __str__(self) -> str: |
| 90 | + return f"{self.operator} {self.operand}" |
| 91 | + |
| 92 | + clauses: tuple[Clause, ...] |
| 93 | + |
| 94 | + def __str__(self) -> str: |
| 95 | + return ", ".join(map(str, self.clauses)) |
| 96 | + |
| 97 | + @classmethod |
| 98 | + def from_string(cls, string: str) -> Self: |
| 99 | + return cls(tuple(map(cls.Clause.from_string, string.split(",")))) |
| 100 | + |
| 101 | + def allows(self, version: str) -> bool: |
| 102 | + """Checks if _all_ clauses allow the given version.""" |
| 103 | + return all(clause.allows(version) for clause in self.clauses) |
| 104 | + |
| 105 | + @classmethod |
| 106 | + def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema: |
| 107 | + return core_schema.json_or_python_schema( |
| 108 | + core_schema.no_info_after_validator_function(cls.from_string, handler(str)), |
| 109 | + core_schema.is_instance_schema(cls), |
| 110 | + serialization=core_schema.to_string_ser_schema() |
| 111 | + ) |
0 commit comments