grpc-framework 0.3.2b0

Easy gRPC APIs Framework.

gRPC Framework — a modern gRPC framework with Pythonic APIs

Language： 🌐 [English](en) | 🇨🇳 [简体中文](zh)

Pressure Test： 🌐 [English](en) | 🇨🇳 [简体中文](zh)

---

Source Code: https://github.com/JokerCrying/grpc-framework

gRPC-Framework is a modern, highly compatible, and more Pythonic gRPC framework for rapidly building gRPC projects and writing gRPC APIs.

Key Features:

• Pythonic: Decorator-driven API design, comprehensive type annotations, multi-paradigm programming, native async support, and flexible extension mechanisms. It simplifies complex gRPC service development into elegant, Pythonic code, enabling developers to build high-performance gRPC projects in the most natural Python way.
• Modern: Embraces modern Python best practices, including native async/await, a complete typing system, domain data modeling, contextvars-based context management, and declarative API design via decorators — fully aligned with Python 3.7+ features and design philosophy.
• Performance: Native asynchronous I/O, configurable thread pool executor, efficient middleware chaining, smart argument parsing cache, and a grpc.aio-based implementation deliver excellent concurrency and low latency, while keeping development convenient.
• Compatibility/Adaptability: Seamlessly interoperates with traditional protoc-generated service code via simple calls. Supports multiple configuration formats (YAML, JSON, INI, Python module), pluggable serializers and codecs, and flexible interceptors and middleware, enabling easy migration and broad tech stack compatibility.
• Simplicity: Clean decorator syntax, zero-config defaults, intuitive class-based and function-based views — build complete gRPC services with just a few lines of code, making complex distributed communication feel like writing regular Python functions.
• gRPC Standards: Fully compliant with gRPC standards, supporting all four standard interaction patterns, protobuf serialization, service reflection, health checks, interceptors, compression algorithms — ensuring full interoperability with any standard gRPC clients and servers.
• Client Support: Feature-complete client, including intelligent connection pool management (supports both async and sync modes), convenient methods for all four gRPC call patterns, automatic connection maintenance, and warm-up mechanisms.

Dependencies

gRPC Framework is built using the following libraries:

• grpcio — standard gRPC communication.
• grpcio-reflection — standard gRPC reflection.
• grpcio-health-checking — standard gRPC health checking.
• protobuf — ProtobufMessage type support and parsing.

Installation

pip install --upgrade pip
pip install grpc-framework

Configuration

gRPC Framework uses a dedicated configuration class and supports YAML, JSON, INI, and Python modules. You can create it via GRPCFrameworkConfig.from_module, GRPCFrameworkConfig.from_file, or by instantiating directly.

Instantiate via Config Files or Python Modules

If your project uses YAML, JSON, INI files, or a Python module for configuration, you can build GRPCFrameworkConfig with helpers. For other formats (e.g., TOML), register a custom parser via GRPCFrameworkConfig.add_config_parser.

• Helpers: GRPCFrameworkConfig.from_module('config'), GRPCFrameworkConfig.from_file('config.yaml').
• Custom parsers: Provide filetype and a parser(filepath, options) that returns a Dict[str, Any]. options is ConfigParserOptions with ini_root_name default.

from grpc_framework import GRPCFrameworkConfig, ConfigParserOptions

# Using a Python module
config_from_module = GRPCFrameworkConfig.from_module('config')

# Using a config file

# Tips: If you need to pass in the 'serializer', 'codec', 'converter' parameters, 
# write it in the format 'python_module.python_file:class/func'
config_from_file = GRPCFrameworkConfig.from_file('config.yaml')


# Add a custom parser (e.g., for TOML)
def from_toml_file(filepath: str, options: ConfigParserOptions):
    import tomllib
    with open(filepath, 'rb') as f:
        return tomllib.load(f)


GRPCFrameworkConfig.add_config_parser('toml', from_toml_file)

• package: Required. The package name that hosts the gRPC app. Default grpc (using exactly grpc is not allowed).
• name: Application name. Default grpc-framework.
• version: Application version, recommended format x.x.x(.beta|alpha).
• host: Bind address. Use [::] to listen on all addresses.
• port: Service port. Default 50051.
• serializer: Global serializer that orchestrates the Codec and Converter to process request data.
• codec: Global Codec that converts request bytes to transport objects. Default ProtobufCodec.
• converter: Global Converter that converts transport objects to domain models. Default ProtobufConverter.
• reflection: Enable gRPC reflection. Default False.
• app_service_name: Service name for function-based views under the app. Default RootService.
• executor_type: 'threading' or 'process', this is to automatically create the corresponding worker in multi-worker mode, and only one will be created in the worker dimension, and it will be applied in the full cycle
• execute_workers: The maximum number of executors in the worker dimension is CPU cores * 2 - 1 by default
• grpc_handlers: Additional gRPC handlers. Default None.
• interceptors: gRPC interceptors. Default None (a request parsing interceptor is loaded during service setup).
• grpc_options: gRPC server options. Default None (converted to an empty dict during app init).
• maximum_concurrent_rpc: Max concurrent RPCs. Default None (unlimited).
• grpc_compression: gRPC compression type. Default None.

Dependency Injection

gRPC Framework introduces a modern dependency injection system (inspired by FastAPI), making dependency management simple and intuitive.

Core Features

• Declarative Injection: Declare dependencies using Depends in function parameters or class attributes.
• Scope Management: Defaults to Request Scope, ensuring dependencies are instantiated only once per request.
• Resource Management: Supports generator dependencies with yield syntax, automatically handling resource initialization (Setup) and cleanup (Teardown), such as database connections.
• Nested Dependencies: Dependencies can have their own dependencies, and the framework automatically resolves and builds the dependency tree.

Examples

1. Injection in Function-Based Views

from grpc_framework import Depends


# Define a dependency
def get_db():
    return "FakeDBConnection"


# Inject into Handler
@app.unary_unary
async def get_user(user_id: int, db: str = Depends(get_db)):
    return {"id": user_id, "db_status": db}

2. Resource Cleanup (Setup/Teardown)

async def get_db_session():
    print("Connecting DB...")
    db = "Session"
    yield db
    print("Closing DB...")


@app.unary_unary
async def query_data(db: str = Depends(get_db_session)):
    return {"data": "ok"}

3. Injection in Class-Based Views

class UserService(Service):
    # Method A: Attribute Injection
    db: str = Depends(get_db)

    @unary_unary
    async def get_info(self):
        return {"db": self.db}

    # Method B: Parameter Injection
    @unary_unary
    async def update_info(self, db: str = Depends(get_db)):
        return {"db": db}

4. Type-Based Injection & Global Registration

Besides passing functions directly, you can declare dependencies using types. Combined with global container registration, this enables elegant dependency management.

class RedisConnect:
    def __init__(self):
        self.host = "localhost"


# 1. Register dependency globally (usually during app startup)
# Register RedisConnect type as itself (can also be a factory function)
app.container.register(RedisConnect, RedisConnect)


# 2. Inject using Depends[Type]
# The framework automatically looks up the Provider for RedisConnect from the container
@app.unary_unary
async def use_redis(redis: Depends[RedisConnect]):
    return {"redis_host": redis.host}

Multi-Worker Mode

To overcome the limitations of the Python GIL and fully utilize multi-core CPUs, the framework supports multi-process Worker mode.

How to Enable

Simply set the workers parameter to greater than 1 in the configuration:

# config.py
workers = 4  # Recommended to set to the number of CPU cores

Or in code:

config = GRPCFrameworkConfig(workers=4)
app = GRPCFramework(config=config)

Key Advantages

• High Performance: Leverages SO_REUSEPORT to allow multiple processes to listen on the same port, with load balancing handled automatically by the OS kernel.
• High Throughput: In high-concurrency scenarios, throughput can increase significantly (approaching Go performance).
• Isolation: Each Worker process runs independently, ensuring higher stability without interference.

Note: Multi-Worker mode relies on the operating system's SO_REUSEPORT feature. Currently, it is only supported on Linux and macOS. On Windows, it will fallback to single-process mode.

Serializer

gRPC Framework provides a serializer that takes two parameters, a codec and a converter. Its main responsibility is converting request data through the pipeline: request data (HTTP/2 data stream) <> transport object <> domain model.

Some built-in codecs and converters are available from grpc_framework:

• JSONCodec: Convert bytes into Dict/List
• ProtobufCodec: Convert bytes into ProtobufMessage
• ORJSONCodec: High-performance JSON codec powered by orjson (*requires installing orjson), leveraging its speed.
• DataclassesCodec: Convert bytes into Dict/List
• ProtobufConverter: Convert between ProtobufMessage and domain model (binary Protobuf data).
• JsonProtobufConverter: Bidirectional conversion between JSON and ProtobufMessage.
• JsonConverter: Convert between JSON strings and domain models.
• DataclassesConverter: Convert between Dataclass and Dict/List (using JSON bytes).

Custom Data Conversion

If the data conversion provided by gRPC Framework does not meet your business needs, you can implement your own serializer. Implement either grpc_framework.TransportCodec or grpc_framework.ModelConverter:

Codec

• decode(self, data: BytesLike, into: OptionalT = None) -> Any: Implement decode to convert raw client bytes into a transport object.
• encode(self, obj: Any) -> BytesLike: Implement encode to convert the transport object back to bytes.

class TransportCodec(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def decode(self, data: BytesLike, into: OptionalT = None) -> Any:
        """bytes -> transport object (e.g., protobuf.Message or dict)"""
        raise NotImplementedError

    @abc.abstractmethod
    def encode(self, obj: Any) -> BytesLike:
        """transport object -> bytes"""
        raise NotImplementedError

Converter

• to_model(self, transport_obj: Any, model_type: TypeT) -> T: Convert the transport object into a domain model.
• from_model(self, model: T) -> Any: Convert the domain model back into a transport object.

class ModelConverter(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def to_model(self, transport_obj: Any, model_type: TypeT) -> T:
        """transport object -> domain model"""
        raise NotImplementedError

    @abc.abstractmethod
    def from_model(self, model: T) -> Any:
        """domain model -> transport object"""
        raise NotImplementedError

Examples

*In the examples below, JSONCodec and JSONConverter are used.

Create and Run an Application

from grpc_framework import GRPCFrameworkConfig, GRPCFramework

config = GRPCFrameworkConfig.from_module('config')

app = GRPCFramework(config=config)

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

Function-Based Views

from grpc_framework import GRPCFrameworkConfig, GRPCFramework, Request

config = GRPCFrameworkConfig.from_module('config')

app = GRPCFramework(config=config)


# Approach 1
@app.unary_unary
def IsServerAlive():
    return {"success": True}


# Approach 2
from grpc_framework import Service

some_service = Service("SomeService")


@some_service.unary_unary
def GetSomeData():
    # You can access the current request information
    request = Request.current()
    print(request.metadata)
    return {"success": True, "data": {"id": 1}}


app.add_service(some_service)

Or use async def...

from grpc_framework import GRPCFrameworkConfig, GRPCFramework, Request

config = GRPCFrameworkConfig.from_module('config')
app = GRPCFramework(config=config)


# Approach 1
@app.unary_unary
async def IsServerAlive():
    return {"success": True}


# Approach 2
from grpc_framework import Service

some_service = Service("SomeService")


@some_service.unary_unary
async def GetSomeData():
    # You can access the current request information
    request = Request.current()
    print(request.metadata)
    return {"success": True, "data": {"id": 1}}


app.add_service(some_service)

Class-Based Views

from grpc_framework import GRPCFrameworkConfig, GRPCFramework, Service, unary_unary, stream_unary, StreamRequest

config = GRPCFrameworkConfig.from_module('config')
app = GRPCFramework(config=config)


class SomeService(Service):
    @unary_unary
    def GetSomeData(self):
        # You can access the current request information
        print(self.request.metadata)
        return {"success": True}

    @stream_unary
    async def sum_counter(self, data: StreamRequest[dict]):
        result = 0
        async for item in data:
            result += data['count']
        return {'result': result}


app.add_service(SomeService)

Or use async def...

from grpc_framework import GRPCFrameworkConfig, GRPCFramework, Service, unary_unary, stream_unary, StreamRequest

config = GRPCFrameworkConfig.from_module('config')
app = GRPCFramework(config=config)


class SomeService(Service):
    @unary_unary
    async def GetSomeData(self):
        # You can access the current request information
        print(self.request.metadata)
        return {"success": True}

    @stream_unary
    async def sum_counter(self, data: StreamRequest[dict]):
        result = 0
        async for item in data:
            result += data['count']
        return {'result': result}


app.add_service(SomeService)

Legacy Compatibility

gRPC Framework provides interfaces to be compatible with legacy projects compiled with protoc, allowing them to be seamlessly hosted within gRPC Framework. However, request context or middleware configured in the framework will not be available, as the legacy service is only hosted rather than fully managed.

Example

import example_pb2
import example_pb2_grpc


class Greeter(example_pb2_grpc.GreeterServicer):
    def say_hello(self, request):
        return example_pb2.HelloReply(message=f"Hello, {request.name}")


app.load_rpc_stub(Greeter(), example_pb2_grpc.add_GreeterServicer_to_server)

Client Support

gRPC Framework provides a client that makes calling gRPC services simple. It supports both calling via generated stubs and by specifying method paths directly. It also includes a gRPC channel pool that supports both async ecosystem channels and default channels.

Channel Pool Configuration

• pool_mode: Required. Supports async and default to manage async ecosystem channels and default channels.
• min_size: Minimum number of connections. Default 10.
• max_size: Maximum number of connections. Default 20.
• secure_mode: Whether to enable secure mode. Affects channel creation. Default False.
• credit: gRPC credentials. Required when secure_mode=True.
• maintenance_interval: Background task checks channel health at this interval. Default 5 seconds.
• auto_preheating: Whether to preheat the pool. Default True. When enabled, the pool warms up to min_size on instantiation.
• channel_options: Additional channel options.

Client Usage

from grpc_framework.client import GRPCChannelPool, GRPCClient, GRPCChannelPoolOptions

grpc_channel_pool = GRPCChannelPool(GRPCChannelPoolOptions(pool_mode='default'))

client = GRPCClient(
    channel_pool_manager=grpc_channel_pool,
    host='localhost',
    port=50051,
    request_serializer=lambda x: x,
    response_deserializer=lambda x: x,
    timeout=5,
)

# Stub-based call
import example_pb2_grpc as example_pb2_grpc
import example_pb2 as example_pb2

request = example_pb2.SimpleRequest(query='1', page_number=1, result_per_page=20)
channel = client.channel_pool_manager.get()
impl = example_pb2_grpc.SimpleServiceStub(channel)
resp = client.call_method(impl.GetSimpleResponse, request)
print(resp)

# Direct method call
response = client.call_method('/package.Service/Method', request_data=b'{"name":"jack"}')
print(response)

Roadmap

Status	Feature	Planned Version	Notes
✅	Dependency collection	v1.1.0	Not started
⬜	Multi-loop support	v1.1.0	Not started
⬜	Version support	v1.1.0	Not started
⬜	Service-level codec/converter	v1.2.0	Not started
⬜	Service-level request context	v1.2.0	Not started