Support Python >=3.8 (#112)

By using explicit typing classes.

Author: Alfus

.github/workflows/ci.yaml

@@ -16,7 +16,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.10", "3.11"]
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
     steps:
       - name: Checkout code
         uses: actions/checkout@v4

.github/workflows/conformance.yaml

@@ -16,7 +16,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.10", "3.11"]
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
     steps:
       - name: Checkout code
         uses: actions/checkout@v4

Pipfile

@@ -17,4 +17,4 @@ exceptiongroup = "*"
 tomli = "*"
 
 [requires]
-python_version = "3.10"
+python_version = "3.8"

Pipfile.lock

@@ -62,7 +62,7 @@ def unwrap(msg: message.Message) -> celtypes.Value:
 }
 
 
-def _msg_to_cel(msg: message.Message) -> dict[str, celtypes.Value]:
+def _msg_to_cel(msg: message.Message) -> typing.Dict[str, celtypes.Value]:
     ctor = _MSG_TYPE_URL_TO_CTOR.get(msg.DESCRIPTOR.full_name)
     if ctor is not None:
         return ctor(msg)
@@ -214,16 +214,21 @@ def validate(self, ctx: ConstraintContext, message: message.Message):  # noqa: A
 class CelConstraintRules(ConstraintRules):
     """A constraint that has rules written in CEL."""
 
-    _runners: list[tuple[celpy.Runner, expression_pb2.Constraint | private_pb2.Constraint]]
+    _runners: typing.List[typing.Tuple[celpy.Runner, typing.Union[expression_pb2.Constraint, private_pb2.Constraint]]]
     _rules_cel: celtypes.Value = None
 
-    def __init__(self, rules: message.Message | None):
+    def __init__(self, rules: typing.Optional[message.Message]):
         self._runners = []
         if rules is not None:
             self._rules_cel = _msg_to_cel(rules)
 
     def _validate_cel(
-        self, ctx: ConstraintContext, field_name: str, activation: dict[str, typing.Any], *, for_key: bool = False
+        self,
+        ctx: ConstraintContext,
+        field_name: str,
+        activation: typing.Dict[str, typing.Any],
+        *,
+        for_key: bool = False,
     ):
         activation["rules"] = self._rules_cel
         activation["now"] = celtypes.TimestampType(datetime.datetime.now(tz=datetime.timezone.utc))
@@ -241,8 +246,8 @@ def _validate_cel(
     def add_rule(
         self,
         env: celpy.Environment,
-        funcs: dict[str, celpy.CELFunction],
-        rules: expression_pb2.Constraint | private_pb2.Constraint,
+        funcs: typing.Dict[str, celpy.CELFunction],
+        rules: typing.Union[expression_pb2.Constraint, private_pb2.Constraint],
     ):
         ast = env.compile(rules.expression)
         prog = env.program(ast, functions=funcs)
@@ -256,7 +261,7 @@ def validate(self, ctx: ConstraintContext, message: message.Message):
         self._validate_cel(ctx, "", {"this": _msg_to_cel(message)})
 
 
-def check_field_type(field: descriptor.FieldDescriptor, expected: int, wrapper_name: str | None = None):
+def check_field_type(field: descriptor.FieldDescriptor, expected: int, wrapper_name: typing.Optional[str] = None):
     if field.type != expected and (
         field.type != descriptor.FieldDescriptor.TYPE_MESSAGE or field.message_type.full_name != wrapper_name
     ):
@@ -273,7 +278,7 @@ class FieldConstraintRules(CelConstraintRules):
     def __init__(
         self,
         env: celpy.Environment,
-        funcs: dict[str, celpy.CELFunction],
+        funcs: typing.Dict[str, celpy.CELFunction],
         field: descriptor.FieldDescriptor,
         field_level: validate_pb2.FieldConstraints,
     ):
@@ -320,13 +325,13 @@ def _validate_value(self, ctx: ConstraintContext, field_path: str, val: typing.A
 class AnyConstraintRules(FieldConstraintRules):
     """Rules for an Any field."""
 
-    _in: list[str] = []  # noqa: RUF012
-    _not_in: list[str] = []  # noqa: RUF012
+    _in: typing.List[str] = []  # noqa: RUF012
+    _not_in: typing.List[str] = []  # noqa: RUF012
 
     def __init__(
         self,
         env: celpy.Environment,
-        funcs: dict[str, celpy.CELFunction],
+        funcs: typing.Dict[str, celpy.CELFunction],
         field: descriptor.FieldDescriptor,
         field_level: validate_pb2.FieldConstraints,
     ):
@@ -362,7 +367,7 @@ class EnumConstraintRules(FieldConstraintRules):
     def __init__(
         self,
         env: celpy.Environment,
-        funcs: dict[str, celpy.CELFunction],
+        funcs: typing.Dict[str, celpy.CELFunction],
         field: descriptor.FieldDescriptor,
         field_level: validate_pb2.FieldConstraints,
     ):
@@ -387,15 +392,15 @@ def validate(self, ctx: ConstraintContext, message: message.Message):
 class RepeatedConstraintRules(FieldConstraintRules):
     """Rules for a repeated field."""
 
-    _item_rules: FieldConstraintRules | None = None
+    _item_rules: typing.Optional[FieldConstraintRules] = None
 
     def __init__(
         self,
         env: celpy.Environment,
-        funcs: dict[str, celpy.CELFunction],
+        funcs: typing.Dict[str, celpy.CELFunction],
         field: descriptor.FieldDescriptor,
         field_level: validate_pb2.FieldConstraints,
-        item_rules: FieldConstraintRules | None,
+        item_rules: typing.Optional[FieldConstraintRules],
     ):
         super().__init__(env, funcs, field, field_level)
         if item_rules is not None:
@@ -422,17 +427,17 @@ def validate(self, ctx: ConstraintContext, message: message.Message):
 class MapConstraintRules(FieldConstraintRules):
     """Rules for a map field."""
 
-    _key_rules: FieldConstraintRules | None = None
-    _value_rules: FieldConstraintRules | None = None
+    _key_rules: typing.Optional[FieldConstraintRules] = None
+    _value_rules: typing.Optional[FieldConstraintRules] = None
 
     def __init__(
         self,
         env: celpy.Environment,
-        funcs: dict[str, celpy.CELFunction],
+        funcs: typing.Dict[str, celpy.CELFunction],
         field: descriptor.FieldDescriptor,
         field_level: validate_pb2.FieldConstraints,
-        key_rules: FieldConstraintRules | None,
-        value_rules: FieldConstraintRules | None,
+        key_rules: typing.Optional[FieldConstraintRules],
+        value_rules: typing.Optional[FieldConstraintRules],
     ):
         super().__init__(env, funcs, field, field_level)
         if key_rules is not None:
@@ -480,15 +485,15 @@ class ConstraintFactory:
     """Factory for creating and caching constraints."""
 
     _env: celpy.Environment
-    _funcs: dict[str, celpy.CELFunction]
-    _cache: dict[descriptor.Descriptor, list[ConstraintRules] | Exception]
+    _funcs: typing.Dict[str, celpy.CELFunction]
+    _cache: typing.Dict[descriptor.Descriptor, typing.Union[typing.List[ConstraintRules], Exception]]
 
-    def __init__(self, funcs: dict[str, celpy.CELFunction]):
+    def __init__(self, funcs: typing.Dict[str, celpy.CELFunction]):
         self._env = celpy.Environment()
         self._funcs = funcs
         self._cache = {}
 
-    def get(self, descriptor: descriptor.Descriptor) -> list[ConstraintRules]:
+    def get(self, descriptor: descriptor.Descriptor) -> typing.List[ConstraintRules]:
         if descriptor not in self._cache:
             try:
                 self._cache[descriptor] = self._new_constraints(descriptor)
@@ -647,9 +652,9 @@ def _new_field_constraint(
             item_rule = self._new_scalar_field_constraint(field, rules.repeated.items)
         return RepeatedConstraintRules(self._env, self._funcs, field, rules, item_rule)
 
-    def _new_constraints(self, desc: descriptor.Descriptor) -> list[ConstraintRules]:
-        result: list[ConstraintRules] = []
-        constraint: ConstraintRules | None = None
+    def _new_constraints(self, desc: descriptor.Descriptor) -> typing.List[ConstraintRules]:
+        result: typing.List[ConstraintRules] = []
+        constraint: typing.Optional[ConstraintRules] = None
         if validate_pb2.message in desc.GetOptions().Extensions:
             message_level = desc.GetOptions().Extensions[validate_pb2.message]
             if message_level.disabled:

protovalidate/internal/constraints.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import math
+import typing
 from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network, ip_address, ip_network
 from urllib import parse as urlparse
 
@@ -59,8 +60,8 @@ def validate_email(addr):
     return _validate_hostname(parts[1])
 
 
-def is_ip(val: celtypes.Value, version: celtypes.Value | None = None) -> celpy.Result:
-    if not isinstance(val, celtypes.BytesType | celtypes.StringType):
+def is_ip(val: celtypes.Value, version: typing.Optional[celtypes.Value] = None) -> celpy.Result:
+    if not isinstance(val, (celtypes.BytesType, celtypes.StringType)):
         msg = "invalid argument, expected string or bytes"
         raise celpy.EvalError(msg)
     try:
@@ -79,7 +80,7 @@ def is_ip(val: celtypes.Value, version: celtypes.Value | None = None) -> celpy.R
 
 
 def is_ip_prefix(val: celtypes.Value, *args) -> celpy.Result:
-    if not isinstance(val, celtypes.BytesType | celtypes.StringType):
+    if not isinstance(val, (celtypes.BytesType, celtypes.StringType)):
         msg = "invalid argument, expected string or bytes"
         raise celpy.EvalError(msg)
     version = None
@@ -147,7 +148,7 @@ def is_nan(val: celtypes.Value) -> celpy.Result:
     return celtypes.BoolType(math.isnan(val))
 
 
-def is_inf(val: celtypes.Value, sign: None | celtypes.Value = None) -> celpy.Result:
+def is_inf(val: celtypes.Value, sign: typing.Optional[celtypes.Value] = None) -> celpy.Result:
     if not isinstance(val, celtypes.DoubleType):
         msg = "invalid argument, expected double"
         raise celpy.EvalError(msg)
@@ -172,7 +173,7 @@ def unique(val: celtypes.Value) -> celpy.Result:
     return celtypes.BoolType(len(val) == len(set(val)))
 
 
-def make_extra_funcs(locale: str) -> dict[str, celpy.CELFunction]:
+def make_extra_funcs(locale: str) -> typing.Dict[str, celpy.CELFunction]:
     string_fmt = string_format.StringFormat(locale)
     return {
         # Missing standard functions

protovalidate/internal/extra_func.py

@@ -152,7 +152,7 @@ def format_string(self, arg: celtypes.Value) -> celpy.Result:
         return celtypes.StringType(arg)
 
     def format_value(self, arg: celtypes.Value) -> celpy.Result:
-        if isinstance(arg, celtypes.StringType | str):
+        if isinstance(arg, (celtypes.StringType, str)):
             return celtypes.StringType(quote(arg))
         if isinstance(arg, celtypes.UintType):
             return celtypes.StringType(arg)

protovalidate/internal/string_format.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import typing
+
 from google.protobuf import message
 
 from buf.validate import expression_pb2  # type: ignore
@@ -98,7 +100,7 @@ def __init__(self, msg: str, violations: expression_pb2.Violations):
         super().__init__(msg)
         self.violations = violations
 
-    def errors(self) -> list[expression_pb2.Violation]:
+    def errors(self) -> typing.List[expression_pb2.Violation]:
         """
         Returns the validation errors as a simple Python list, rather than the
         Protobuf-specific collection type used by Violations.

protovalidate/validator.py

@@ -8,7 +8,7 @@ description = "Protocol Buffer Validation for Python"
 readme = "README.md"
 license = { file = "LICENSE" }
 keywords = ["validate", "protobuf", "protocol buffer"]
-requires-python = ">=3.10"
+requires-python = ">=3.8"
 classifiers = [
   "Programming Language :: Python :: 3",
   "License :: OSI Approved :: Apache Software License",
@@ -26,11 +26,11 @@ Issues = "https://github.com/bufbuild/protovalidate-python/issues"
 source = "vcs"
 
 [tool.black]
-target-version = ["py310"]
+target-version = ["py38"]
 line-length = 120
 
 [tool.ruff]
-target-version = "py310"
+target-version = "py38"
 line-length = 120
 select = [
   "A",

pyproject.toml

@@ -49,7 +49,7 @@
 from buf.validate.conformance.harness import harness_pb2
 
 
-def run_test_case(tc: typing.Any, result: harness_pb2.TestResult | None = None) -> harness_pb2.TestResult:
+def run_test_case(tc: typing.Any, result: typing.Optional[harness_pb2.TestResult] = None) -> harness_pb2.TestResult:
     if result is None:
         result = harness_pb2.TestResult()
     # Run the validator
@@ -67,7 +67,7 @@ def run_test_case(tc: typing.Any, result: harness_pb2.TestResult | None = None)
 def run_any_test_case(
     pool: descriptor_pool.DescriptorPool,
     tc: any_pb2.Any,
-    result: harness_pb2.TestResult | None = None,
+    result: typing.Optional[harness_pb2.TestResult] = None,
 ) -> harness_pb2.TestResult:
     type_name = tc.type_url.split("/")[-1]
     desc: descriptor.Descriptor = pool.FindMessageTypeByName(type_name)