Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 19 additions & 6 deletions protovalidate/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,15 +12,28 @@
# See the License for the specific language governing permissions and
# limitations under the License.

from protovalidate import validator
"""The semantic validation library for Protobuf in Python."""

Validator = validator.Validator
CompilationError = validator.CompilationError
ValidationError = validator.ValidationError
Violations = validator.Violations
from __future__ import annotations

from protovalidate._validator import (
CompilationError,
ValidationError,
Validator,
Violation,
Violations,
)

_default_validator = Validator()
validate = _default_validator.validate
collect_violations = _default_validator.collect_violations

__all__ = ["CompilationError", "ValidationError", "Validator", "Violations", "collect_violations", "validate"]
__all__ = [
"CompilationError",
"ValidationError",
"Validator",
"Violation",
"Violations",
"collect_violations",
"validate",
]
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@

"""Which CEL backend is available."""

from __future__ import annotations


def _detect() -> bool:
try:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,17 +12,22 @@
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import threading
from typing import TYPE_CHECKING

import celpy
import celpy.celtypes

if TYPE_CHECKING:
from lark import Tree

_has_state = threading.local()


def in_has() -> bool:
"""
Returns true if inside of CEL interpreter `has` macro.
"""Returns true if inside of CEL interpreter `has` macro.

This enables working around an issue in cel-python where it is not possible
to implement protobuf semantics around the `has` macro.
Expand All @@ -35,12 +40,11 @@ def in_has() -> bool:
class InterpretedRunner(celpy.InterpretedRunner):
def evaluate(self, context: celpy.Context) -> celpy.celtypes.Value:
class Evaluator(celpy.Evaluator):
def macro_has_eval(self, exprlist) -> celpy.celtypes.BoolType:
def macro_has_eval(self, exprlist: Tree) -> celpy.celtypes.BoolType:
_has_state.in_has = True
result = super().macro_has_eval(exprlist)
_has_state.in_has = False
return result

e = Evaluator(ast=self.ast, activation=self.new_activation())
value = e.evaluate(context)
return value
return e.evaluate(context)
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@

"""The cel-expr-python (cel-cpp) validation engine."""

from __future__ import annotations

from .bridge import GoogleBridge
from .extra_func import make_extension
from .rules import RuleFactory
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations

from google.protobuf import descriptor as google_descriptor

# protobuf 7+ removed FieldDescriptor.label / LABEL_REPEATED in favour of is_repeated.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,30 +16,32 @@

from __future__ import annotations

import typing
from typing import TYPE_CHECKING

from google.protobuf import descriptor_pb2 as google_descriptor_pb2
from google.protobuf import descriptor_pool as google_descriptor_pool
from google.protobuf import message as google_message
from google.protobuf import message_factory as google_message_factory
from protobuf import DescFile, DescMessage
from google.protobuf import (
descriptor_pb2 as google_descriptor_pb2,
descriptor_pool as google_descriptor_pool,
message as google_message,
message_factory as google_message_factory,
)

if typing.TYPE_CHECKING:
if TYPE_CHECKING:
import protobuf
from protobuf import DescFile, DescMessage


class GoogleBridge:
"""Lazily mirrors protobuf-py descriptors into google's pool and bridges
protobuf-py message values to google dynamic messages."""
"""Bridges from protobuf-py messages to google.protobuf for use with cel-expr-python."""

def __init__(self) -> None:
self._pool: google_descriptor_pool.DescriptorPool = google_descriptor_pool.Default()
self._pool: google_descriptor_pool.DescriptorPool = (
google_descriptor_pool.Default()
)
self._mirrored: set[str] = set()
self._classes: dict[str, type[google_message.Message]] = {}

def _mirror_file(self, desc_file: DescFile) -> None:
"""Registers a protobuf-py DescFile (and its transitive deps) into the
google pool, dependencies first, skipping files already present."""
"""Registers a protobuf-py DescFile and deps into the google pool."""
if desc_file.name in self._mirrored:
return
self._mirrored.add(desc_file.name)
Expand All @@ -48,7 +50,9 @@ def _mirror_file(self, desc_file: DescFile) -> None:
try:
self._pool.FindFileByName(desc_file.name)
except KeyError:
proto = google_descriptor_pb2.FileDescriptorProto.FromString(desc_file.proto.to_binary())
proto = google_descriptor_pb2.FileDescriptorProto.FromString(
desc_file.proto.to_binary()
)
self._pool.Add(proto)

def google_class(self, desc: DescMessage) -> type[google_message.Message]:
Expand Down
Loading
Loading