Skip to content

API Reference⚓︎

fast_grpc ⚓︎

Classes⚓︎

FastGRPC ⚓︎

FastGRPC(*services, loop=get_event_loop(), port=50051, reflection=False, middlewares=())

Server application.

PARAMETER DESCRIPTION
*services

Tuple of Fast-gRPC services.

TYPE: tuple[FastGRPCService] DEFAULT: ()

loop

Async event loop for running server.

TYPE: AbstractEventLoop DEFAULT: get_event_loop()

port

Port for listen requests.

TYPE: int DEFAULT: 50051

reflection

Flag for enable/disable server gRPC reflection.

TYPE: bool DEFAULT: False

middlewares

Tuple of middlewares (interceptors).

TYPE: tuple[FastGRPCMiddleware | Callable] DEFAULT: ()

Example
from fast_grpc import FastGRPC, FastGRPCService, grpc_method

class ExampleService(FastGRPCService):
    ...

app = FastGRPC(ExampleService())
app.run()

Functions⚓︎

add_service ⚓︎
add_service(service)

Add service to server.

PARAMETER DESCRIPTION
service

gRPC service.

TYPE: FastGRPCService

Example
app = FastGRPC()
app.add_service(ExampleService())
run ⚓︎
run()

Run server.

run_async async ⚓︎
run_async()

Run server.

Example
app = FastGRPC()
await app.run_async()

StatusCode ⚓︎

Bases: IntEnum

Status codes enum for describing gRPC methods response codes.

Attributes⚓︎

OK class-attribute instance-attribute ⚓︎
OK = 0
CANCELLED class-attribute instance-attribute ⚓︎
CANCELLED = 1
UNKNOWN class-attribute instance-attribute ⚓︎
UNKNOWN = 2
INVALID_ARGUMENT class-attribute instance-attribute ⚓︎
INVALID_ARGUMENT = 3
DEADLINE_EXCEEDED class-attribute instance-attribute ⚓︎
DEADLINE_EXCEEDED = 4
NOT_FOUND class-attribute instance-attribute ⚓︎
NOT_FOUND = 5
ALREADY_EXISTS class-attribute instance-attribute ⚓︎
ALREADY_EXISTS = 6
PERMISSION_DENIED class-attribute instance-attribute ⚓︎
PERMISSION_DENIED = 7
RESOURCE_EXHAUSTED class-attribute instance-attribute ⚓︎
RESOURCE_EXHAUSTED = 8
FAILED_PRECONDITION class-attribute instance-attribute ⚓︎
FAILED_PRECONDITION = 9
ABORTED class-attribute instance-attribute ⚓︎
ABORTED = 10
OUT_OF_RANGE class-attribute instance-attribute ⚓︎
OUT_OF_RANGE = 11
UNIMPLEMENTED class-attribute instance-attribute ⚓︎
UNIMPLEMENTED = 12
INTERNAL class-attribute instance-attribute ⚓︎
INTERNAL = 13
UNAVAILABLE class-attribute instance-attribute ⚓︎
UNAVAILABLE = 14
DATA_LOSS class-attribute instance-attribute ⚓︎
DATA_LOSS = 15
UNAUTHENTICATED class-attribute instance-attribute ⚓︎
UNAUTHENTICATED = 16

FastGRPCMiddleware ⚓︎

Middleware.

FastGRPCService ⚓︎

Implementation of gRPC service.

Attributes⚓︎

is_proxy class-attribute instance-attribute ⚓︎
is_proxy = True

Functions⚓︎

get_service_name ⚓︎
get_service_name()

Metod for getting service full name from pb2.

RETURNS DESCRIPTION
str

Full name string of service from pb2.

get_proto classmethod ⚓︎
get_proto()

Render and return content of proto file for this gRPC service.

RETURNS DESCRIPTION
str

Protobuf file content string.

from_proto classmethod ⚓︎
from_proto(proto_file, grpc_path=cwd())

Create gRPC service interface from proto file.

RETURNS DESCRIPTION
type[Self]

New service class, based on FastGRPCService.

Functions⚓︎

grpc_method ⚓︎

grpc_method(function=None, /, name=None, request_model=None, response_model=None, middlewares=(), disable=False)

Decorator for setting method as gRPC.

PARAMETER DESCRIPTION
function

Original request handler.

TYPE: Callable | None DEFAULT: None

name

Name for gRPC method.

TYPE: str | None DEFAULT: None

request_model

Model for describe request data.

TYPE: type[BaseModel] | None DEFAULT: None

response_model

Model for describe response data.

TYPE: type[BaseModel] | None DEFAULT: None

middlewares

Iterable of middlewares.

TYPE: Iterable[FastGRPCMiddleware | Callable] DEFAULT: ()

disable

Flag for enable/disable gRPC method.

TYPE: bool DEFAULT: False

Example
from fast_grpc import FastGRPCService, grpc_method
from pydantic import BaseModel

class ExampleService(FastGRPCService):
    @grpc_method
    async def ping(self, request: BaseModel) -> BaseModel:
        ...

    @grpc_method(
        name="isHealth",
        request_model=BaseModel,
        response_model=BaseModel,
        middlewares=(),
        disable=False,
    )
    async def is_health(self, request, context):
        ...