gRPC API Client

This module aims to provide a client that handles compiled protocol buffer code for a generic endpoint. This module reads a *.pb2 file generated by the protoc CLI tool. For documentation on how to generate protocol buffer bindings for your gRPC API endpoint please visit the Protocol Buffers Python Tutorial. This was inspired in an effort to interact with the Dex gRPC API running in Kubernetes.

Your generated protocol buffer bindings must be in your Python path and available as an import. As protoc generates two files, they also must be able to resolve imports for each other.

Local gRPC Service

To build and run a local gRPC server for testing:

$ make build
$ make run
$ make run docker_port=8888 # Run on a different port

This will start a Docker container with a basic configuration for gRPC testing listening on port 35557, or a port of your choosing.

Protocol Buffer Definitions

The gRPC bindings are generated automatically from the protocol buffers definition file. This can vary greatly from project to project. This repository contains a sample protoc definition compiled into grpc_api_client.sample that aims to be comprehensive enough to cover most types of request/response scenarios.

Example Usage

You can connect to a gRPC server:

from grpc_api_client.client import gRPC_API_Client
from grpc_api_client.sample import gRPC_API_Test_Fields

# These bindings should be generated by the user and made available in the Python path
client = gRPC_API_Client(
    'my.protoc.bindings_pb2',
    'my.protoc.bindings_pb2_grpc'
)

# Connect to the gRPC server using the provided bindings
client.connect('localhost', '5555',
    ca_cert = '/my/grpc/ca.crt',
    client_cert = '/my/grpc/client.crt',
    client_key = '/my/grpc/client.key'
)

# Show the available API methods
for name, method in client.api:
    print('API Method: {}: {}'.format(name, method))
    print('  InputClass: {}: {}'.format(method.input.name, method.input))
    print('    > Fields:')
    for field_name, field_obj in method.input_fields():
        print('      > {}: {}'.format(field_name, field_obj))
    print('  OutputClass: {}: {}'.format(method.output.name, method.output))

    # Print randomly generated sample data
    print('  SampleFields: {}'.format(gRPC_API_Test_Fields.create(method)))

Tests

Testing is done with unittest and nose for discovery.

$ make test