diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 046d2cc..07fcd7c 100644
--- a/.github/workflows/ci.yaml
+++ b/.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
diff --git a/.github/workflows/conformance.yaml b/.github/workflows/conformance.yaml
index 277915d..752e3fc 100644
--- a/.github/workflows/conformance.yaml
+++ b/.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
diff --git a/Pipfile b/Pipfile
index ae7cf1b..fa4894f 100644
--- a/Pipfile
+++ b/Pipfile
@@ -17,4 +17,4 @@ exceptiongroup = "*"
tomli = "*"
[requires]
-python_version = "3.10"
+python_version = "3.8"
diff --git a/Pipfile.lock b/Pipfile.lock
index 9d1d622..50799e5 100644
--- a/Pipfile.lock
+++ b/Pipfile.lock
@@ -1,11 +1,11 @@
{
"_meta": {
"hash": {
- "sha256": "4d528deff7028686ac5fac50920324fefd22fcf168e33a2494110a75fe03296d"
+ "sha256": "689a69b69eb47c8c0cef034d3ec46bd52c1c933a8b6a9c23b0ebba3bcb4ddf53"
},
"pipfile-spec": 6,
"requires": {
- "python_version": "3.10"
+ "python_version": "3.8"
},
"sources": [
{
@@ -186,6 +186,14 @@
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'",
"version": "==2.8.2"
},
+ "pytz": {
+ "hashes": [
+ "sha256:7b4fddbeb94a1eba4b557da24f19fdf9db575192544270a9101d8509f9f43d7b",
+ "sha256:ce42d816b81b68506614c11e8937d3aa9e41007ceb50bfdcb0749b921bf646c7"
+ ],
+ "markers": "python_version < '3.9'",
+ "version": "==2023.3.post1"
+ },
"pyyaml": {
"hashes": [
"sha256:04ac92ad1925b2cff1db0cfebffb6ffc43457495c9b3c39d3fcae417d7125dc5",
diff --git a/protovalidate/internal/constraints.py b/protovalidate/internal/constraints.py
index 864d498..283c3a8 100644
--- a/protovalidate/internal/constraints.py
+++ b/protovalidate/internal/constraints.py
@@ -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:
diff --git a/protovalidate/internal/extra_func.py b/protovalidate/internal/extra_func.py
index b9fb6ad..88f0de0 100644
--- a/protovalidate/internal/extra_func.py
+++ b/protovalidate/internal/extra_func.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
diff --git a/protovalidate/internal/string_format.py b/protovalidate/internal/string_format.py
index 29b92b3..bed78f5 100644
--- a/protovalidate/internal/string_format.py
+++ b/protovalidate/internal/string_format.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)
diff --git a/protovalidate/validator.py b/protovalidate/validator.py
index 90c122b..3b6f20d 100644
--- a/protovalidate/validator.py
+++ b/protovalidate/validator.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.
diff --git a/pyproject.toml b/pyproject.toml
index f979220..673ab59 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -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",
diff --git a/tests/conformance/runner.py b/tests/conformance/runner.py
index e704159..f5f41d2 100644
--- a/tests/conformance/runner.py
+++ b/tests/conformance/runner.py
@@ -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)