Usage¶

To use gRPCAlchemy in a project:

from grpcalchemy.orm import Message
from grpcalchemy import Server, Context, grpcmethod

class HelloMessage(Message):
    __filename__ = 'hello'
    text: str

class HelloService(Server):
    @grpcmethod
    def Hello(self, request: HelloMessage, context: Context) -> HelloMessage:
        return HelloMessage(text=f'Hello {request.text}')

if __name__ == '__main__':
    HelloService.run()

Defining our Message¶

Any message which is used in RPC method must have a explicit schema. We can use py files generated by proto files by the grpc_tools straightly. However defining the schemas by our ORM can help to iron out bugs involving incorrect types or missing fields, and also allow us to define utility methods on our message in the same way that traditional ORMs do.

In our Tutorial Application we need to send several different types of message. We will need to have a collection of users, so that we may link posts to an individual. We also need to send our different types of posts (eg: text, image and link) in the RPC method. To aid navigation of our Tutorial Application, posts may have tags associated with them, so that the list of posts shown to the user may be limited to posts that have been assigned a specific tag. Finally, it would be nice if comments could be added to posts. We’ll start with users, as the other document models are slightly more involved.

Users¶

Just as if we were using a RPC Message with an ORM, we need to define which fields a User may have, and what types of data they might have:

from grpcalchemy.orm import Message
class User(Message):
    email: str
    first_name: str
    last_name: str

Posts, Comments and Tags¶

Now we’ll think about how to define the rest of the information. To associate the comments with individual posts, We’d also need a link message to provide the many-to-many relationship between posts and tags.

Posts¶

We can think of Post as a base class, and TextPost, ImagePost and LinkPost as subclasses of Post.

from grpcalchemy.orm import Message
class Post(Message):
    title: str
    author: str

class TextPost(Post):
    content: str

class ImagePost(Post):
    image_path: str

class LinkPost(Post):
    link_url: str

We are storing a reference to the author of the posts using a ReferenceField object. These are equal to use other message types in RPC message.

Tags¶

Now that we have our Post models figured out, how will we attach tags to them? RPC message allows us to define lists of items natively. So, for both efficiency and simplicity’s sake, we’ll define the tags as strings directly within the post. Let’s take a look at the code of our modified Post class:

from typing import List
from grpcalchemy.orm import Message, Repeated
class Post(Message):
    title: str
    author: User
    tags: Repeated[str]

The ListField object that is used to define a Post’s tags takes a field object as its first argument — this means that you can have lists of any type of field (including lists).

Note

We don’t need to modify the specialized post types as they all inherit from Post.

Comments¶

A comment is typically associated with one post.utility methods, in exactly the same way we do with regular documents:

from grpcalchemy.orm import Message
class Comment(Message):
    content: str
    name: str

We can then define a list of comment documents in our post message:

from typing import List
from grpcalchemy.orm import Message, Repeated
class Post(Message):
    title: str
    author: User
    tags: Repeated[str]
    comments: Repeated[Comment]

Defining our gRPC Method¶

grpcmethod is a decorator indicating gRPC methods.

The valid gRPC Method must be with explicit type hint to define the type of request and return value.

Using Iterator to define Stream gRPC Method:

from typing import Iterator

class HelloService(Server):

    @grpcmethod
    def UnaryUnary(self, request: HelloMessage, context: Context) -> HelloMessage:
        return HelloMessage(text=f'Hello {request.text}')

    @grpcmethod
    def UnaryStream(self, request: HelloMessage, context: Context) -> Iterator[HelloMessage]:
        yield HelloMessage(text=f'Hello {request.text}')

    @grpcmethod
    def StreamUnary(self, request: Iterator[HelloMessage], context: Context) -> HelloMessage:
        for r in request:
            pass
        return HelloMessage(text=f'Hello {r.text}')

    @grpcmethod
    def StreamStream(self, request: Iterator[HelloMessage], context: Context) -> Iterator[HelloMessage]:
        for r in request:
            yield HelloMessage(text=f'Hello {r.text}')

The above code is equal to an RPC service with a method:

syntax = "proto3";

service HelloService {
    rpc StreamStream (stream HelloMessage) returns (stream HelloMessage) {}

    rpc StreamUnary (stream HelloMessage) returns (HelloMessage) {}

    rpc UnaryStream (HelloMessage) returns (stream HelloMessage) {}

    rpc UnaryUnary (HelloMessage) returns (HelloMessage) {}
}

Using Blueprint to Build Your Large Application¶

gRPCAlchemy uses a concept of blueprints for making gRPC services and supporting common patterns within an application or across applications. Blueprint can greatly simplify how large applications work.

from typing import List, Type
from grpcalchemy.orm import Message
from grpcalchemy import Server, Context, Blueprint

class MyService(Server):
    @classmethod
    def get_blueprints(cls) -> List[Type[Blueprint]]:
        return [HelloService]

class HelloMessage(Message):
    __filename__ = 'hello'
    text: str

class HelloService(Blueprint):
    @grpcmethod
    def Hello(self, request: HelloMessage, context: Context) -> HelloMessage:
        return HelloMessage(text=f'Hello {request.text}')


if __name__ == '__main__':
    MyService.run()

Configuration¶

You can define your custom config by inherit from DefaultConfig which defined a list of configuration available in gRPCAlchemy and their default values.

Note

DefaultConfig is defined by configalchemy - https://configalchemy.readthedocs.io

from grpcalchemy import DefaultConfig

from hello import HelloService

class MyConfig(DefaultConfig):
    ...

config = MyConfig()

HelloService.run(config=config)

Middleware¶

Middleware is a framework of hooks into gRPCAlchemy’s request/response processing.

Costume middleware can implement by overriding Blueprint.before_request, Blueprint.after_request, Server.process_request and Server.process_response.