gRPC argument validator
gRPC argument validator is a library that provides decorators to automatically validate arguments in requests to rpc methods.
Getting Started
This is an example of how you may give instructions on setting up your project locally. To get a local copy up and running follow these simple example steps.
Installation
From PyPI
pip install grpc-argument-validator
From source
-
Install
poetry
-
Clone repo
git clone https://github.com/messagebird/python-grpc-argument-validator.git
- Install packages
cd python-grpc-argument-validator && poetry install
- Run the tests
cd src/tests
poetry run python -m unittest
Quick Example
from google.protobuf.descriptor import FieldDescriptor
from grpc_argument_validator import validate_args
from grpc_argument_validator import AbstractArgumentValidator, ValidationResult, ValidationContext
class PathValidator(AbstractArgumentValidator):
def check(self, name: str, value: Path, field_descriptor: FieldDescriptor, validation_context: ValidationContext) -> ValidationResult:
if len(value.points) > 5:
return ValidationResult(valid=True)
return ValidationResult(False, f"path for '{name}' should be at least five points long")
class RouteService(RouteCheckerServicer):
@validate_args(
non_empty=["tags", "tags[]", "path.points"],
validators={"path": PathValidator()},
)
def Create(self, request: Route, context: grpc.ServicerContext):
return BoolValue(value=True)
Documentation
We host the full API reference on GitHub pages.
Argument field syntax
To specify which argument field should be validated, grpc-argument-validator
expects strings that match the field names
as defined in the protobufs. To access nested fields, use a dot (.
).
Consider the following protobuf definition:
syntax = "proto3";
package routeguide;
import "google/protobuf/empty.proto";
import "google/protobuf/wrappers.proto";
message Point {
int32 x = 1;
int32 y = 2;
google.protobuf.StringValue name = 3;
}
message Rectangle {
Point lo = 1;
Point hi = 2;
}
message Area {
Rectangle rectangle = 1;
google.protobuf.StringValue message = 2;
google.protobuf.BytesValue uuid = 3;
}
message Path {
repeated Point points = 1;
}
enum Planet {
PLANET_INVALID = 0;
PLANET_EARTH = 1;
PLANET_MARS = 2;
}
message PlanetValue {
Planet value = 1;
}
message Route {
Path path = 1;
google.protobuf.StringValue name = 2;
PlanetValue planet = 3;
repeated string tags = 4;
}
service RouteService {
rpc CreateRoute(Route) returns (google.protobuf.Empty);
rpc CreateArea(Area) returns (google.protobuf.Empty);
}
- If you want to validate the field
planet
in aRoute
proto, simply specify"planet"
or equivalently".planet"
. - If you want to validate the
value
field within thename
field of aRoute
proto, use"name.value"
or equivalently".name.value"
. - If you want to apply a check to each element of a
repeated
field, append[]
to the name of the field. - If you want to apply a check to the 'root proto' (i.e. the request itself), use
"."
as the field path.
To clarify this, let's say that we know that both planet
and name.value
should have non-default values. We can then
decorate a method in our gRPC server as follows:
import grpc
from google.protobuf.empty_pb2 import Empty
from grpc_argument_validator import validate_args
from tests.route_guide_protos.route_guide_pb2 import Route
from tests.route_guide_protos.route_guide_pb2_grpc import RouteServiceServicer
class RouteServiceImpl(RouteServiceServicer):
@validate_args(non_empty=["planet", "name.value"])
def CreateRoute(self, request: Route, context: grpc.ServicerContext):
return Empty()
Calling the service with a default value for either planet
or name.value
will yield an INVALID_ARGUMENT
status code
with further details on which fields violate the validation.
Validators
There are two kinds of validators you might consider:
- There are predefined validators which we will cover shortly
- Another option is to define your own validators
In the examples below, we have used exactly one validator + field path per validate_args
decorator for clarity.
Fortunately, our API allows you to use multiple validators and fields!
'Has' validator
The simplest of all predefined validators is the 'has' validator which simply checks whether a HasField
evaluates to
True
. This of course works in combination with nested fields.
In the example below, calling the Create
endpoint without setting Route.name
would result in an INVALID_ARGUMENT
status.
class RouteServiceImpl(RouteServiceServicer):
@validate_args(has=["name"])
def CreateRoute(self, request: Route, context: grpc.ServicerContext):
return Empty()
Run this on a local machine and make a request with an invalid argument:
with grpc.insecure_channel("127.0.0.1:50051") as c:
route_client = RouteServiceStub(channel=c)
try:
route_client.CreateRoute(Route(tags=["tag"]))
except grpc.RpcError as e:
if isinstance(e, grpc.Call):
print(e.details())
The following will be printed:
must have 'name'
UUID validator
Another common use-case is the validation of UUIDs. You can enlist the fields that should be UUIDs (represented as
16 bytes) with the uuids
argument:
class RouteServiceImpl(RouteServiceServicer):
@validate_args(uuids=["uuid.value"])
def CreateArea(self, request: Area, context: grpc.ServicerContext):
return Empty()
The client side might violate the UUID requirement as follows:
with grpc.insecure_channel("127.0.0.1:50051") as c:
route_client = RouteServiceStub(channel=c)
try:
route_client.CreateArea(Area(uuid=BytesValue(value="not a uuid".encode())))
except grpc.RpcError as e:
if isinstance(e, grpc.Call):
print(e.details())
This will print 'uuid.value' must be a valid UUID
.
Non-default validator
For fields that should have a non-default value, such as
enums, we have provided the non_default
argument:
class RouteServiceImpl(RouteServiceServicer):
@validate_args(non_default=["planet.value"])
def CreateRoute(self, request: Route, context: grpc.ServicerContext):
return Empty()
The client side may violate this as follows:
with grpc.insecure_channel("127.0.0.1:50051") as c:
route_client = RouteServiceStub(channel=c)
try:
route_client.CreateRoute(Route(planet=PlanetValue()))
except grpc.RpcError as e:
if isinstance(e, grpc.Call):
print(e.details())
Which will print 'planet.value' must have non-default value
.
Non-empty validator
We provide a 'non-'empty validator which can be used to ensure that a repeated
field has more than zero elements.
class RouteServiceImpl(RouteServiceServicer):
@validate_args(non_empty=["tags"])
def CreateRoute(self, request: Route, context: grpc.ServicerContext):
return Empty()
Which can be violated as follows:
with grpc.insecure_channel("127.0.0.1:50051") as c:
route_client = RouteServiceStub(channel=c)
try:
route_client.CreateRoute(Route(tags=[]))
except grpc.RpcError as e:
if isinstance(e, grpc.Call):
print(e.details())
Which will print 'tags' must be non-empty
.
Regexp validator
Finally, we have the regexp validator that can be used to check whether a string field matches a regular expression.
class RouteServiceImpl(RouteServiceServicer):
@validate_args(validators={"message.value": RegexpValidator(pattern=r"\d+")})
def CreateArea(self, request: Area, context: grpc.ServicerContext):
return Empty()
with grpc.insecure_channel("127.0.0.1:50051") as c:
route_client = RouteServiceStub(channel=c)
try:
route_client.CreateArea(Area(message=StringValue(value="hello world")))
except grpc.RpcError as e:
if isinstance(e, grpc.Call):
print(e.details())
Which will print 'message.value' must match regexp pattern: \d+
.
Custom validators
You can also write custom validators to flexibily handle your use-case. You need to derive a class from
AbstractArgumentValidator
and implement its check
method. The example below shows how to implement a simple
validator for checking that a path has 5 points. You can provide such custom validators through a dict
that
maps a field path to a validator:
from grpc_argument_validator import AbstractArgumentValidator
from grpc_argument_validator import ValidationContext
from grpc_argument_validator import ValidationResult
from google.protobuf.descriptor import FieldDescriptor
from examples.route_guide_pb2 import Path
class PathValidator(AbstractArgumentValidator):
def check(
self, name: str, value: Path, field_descriptor: FieldDescriptor, validation_context: ValidationContext
) -> ValidationResult:
if len(value.points) > 5:
return ValidationResult(valid=True)
return ValidationResult(False, f"path for '{name}' should be at least five points long")
class RouteServiceImpl(RouteServiceServicer):
@validate_args(validators={"path": PathValidator()})
def CreateRoute(self, request: Route, context: grpc.ServicerContext):
return Empty()
Optional vs. required validators
For each of the built-in validators (except for the has
validator), validate_args
has not one but two keyword
arguments. One of those is prepended with optional_
. This means that apart from uuid
, non_default
and
non_empty
we also have optional_uuid
, optional_non_default
and optional_non_empty
. The behavior is slightly
different: for any of the optional_*
validators, it is OK if the field is not contained by the incoming request.
Sometimes fields are simply optional, and you only want to validate them if they are present.
Since it is also common that fields are not optional, we also provide the required validators (without optional_*
)
for which HasField
must evaluate to True
for that field and all preceding fields in the protos hierarchy.
The custom validator counterparts are validators
and optional_validators
. Each takes a dict
with a mapping of
field paths to validators. These can be used for validators that might be preconfigured such as the RegexpValidator
or for customer validators.
Streaming requests
You can also use the validators for streaming requests. Since streaming requests might not all look the same in a
single stream (e.g. the first request might have metadata describing the remainder of the stream), we provide a
streaming request index in a ValidationContext
that is passed to an AbstractArgumentValidator
.
Here's an example of how that could be used:
class StreamingPathValidator(AbstractArgumentValidator):
def __init__(self, first_number_of_points: int, second_number_of_points: int):
self._first_number_of_points = first_number_of_points
self._second_number_of_points = second_number_of_points
def check(
self, name: str, value: Any, field_descriptor: FieldDescriptor, validation_context: ValidationContext
) -> ValidationResult:
if not validation_context.is_streaming:
return ValidationResult(False, "request must be a streaming request")
if validation_context.streaming_message_index == 0:
if len(value.points) != self._first_number_of_points:
return ValidationResult(False, f"first path should have {self._first_number_of_points} points")
if validation_context.streaming_message_index == 1:
if len(value.points) != self._second_number_of_points:
return ValidationResult(False, f"second path should have {self._second_number_of_points} points")
return ValidationResult(True)
Enabling rich error details
To enable richer error responses where each violation is
contained in a
BadRequest
proto, you can use
from grpc_argument_validator import ArgumentValidatorConfig
ArgumentValidatorConfig.set_rich_grpc_errors(enabled=True)
Now, your client-side can parse the error details as follows:
def extract_error_details(err):
status_proto = status_pb2.Status()
for metadatum in err.trailing_metadata():
if isinstance(metadatum, _Metadatum):
if metadatum.key == "grpc-status-details-bin":
status_proto.MergeFromString(metadatum.value)
unpacked = [_unpack_error_detail(det) for det in status_proto.details]
return unpacked
def _unpack_error_detail(grpc_detail):
val = error_details_pb2.BadRequest()
grpc_detail.Unpack(val)
return val
with grpc.insecure_channel("127.0.0.1:50051") as c:
route_client = RouteServiceStub(channel=c)
try:
route_client.CreateArea(Area(message=StringValue(value="hello world")))
except grpc.RpcError as e:
error_details = extract_error_details(e)
print(error_details)
Contributing
Contributions are what make the open source community such an amazing place to be learn, inspire, and create. Any contributions you make are greatly appreciated.
- Fork the Project
- Create your Feature Branch (
git checkout -b feature/AmazingFeature
) - Commit your Changes (
git commit -m 'Add some AmazingFeature'
) - Push to the Branch (
git push origin feature/AmazingFeature
) - Open a Pull Request
Generating HTML Documentation
Generate the docs by running:
pdoc --html -o docs src/grpc_argument_validator
License
Distributed under The BSD 3-Clause License. Copyright (c) 2021, MessageBird