Skip to main content

gRPC with autogen by Pydantic models

Project description

Pydantic & gRPC

py-grpcio is a microframework and high-level wrapper of grpcio to simplify work with the original library using abstractions, useful python objects and pydantic models.

Examples of use are given below and also duplicated in the example directory.


Install latest

pip install py-grpcio

Example

Models

Pydantic models that describe messages for client-server interaction.

from uuid import UUID, uuid4
from datetime import datetime

from pydantic import Field

from py_grpcio import Message

from example.server.service.enums import Names


class PingRequest(Message):
    id: UUID = Field(default_factory=uuid4)


class PingResponse(Message):
    id: UUID
    timestamp: datetime = Field(default_factory=datetime.now)


class ComplexModel(Message):
    name: Names


class ComplexRequest(Message):
    id: UUID
    model: ComplexModel


class ComplexResponse(Message):
    id: UUID
    model: ComplexModel

Server

Basic implementation of gRPC services on the server side.

You need to describe the service abstractly and duplicate this service on the client side.

from abc import abstractmethod

from py_grpcio import BaseService

from example.server.service.models import PingRequest, PingResponse, ComplexRequest, ComplexResponse


class BaseExampleService(BaseService):
    @abstractmethod
    async def ping(self: 'BaseExampleService', request: PingRequest) -> PingResponse:
        ...

    @abstractmethod
    async def complex(self: 'BaseExampleService', request: ComplexRequest) -> ComplexResponse:
        ...

Full implementation of the gRPC service with methods.

from example.server.service.base import BaseExampleService
from example.server.service.models import PingRequest, PingResponse, ComplexRequest, ComplexResponse


class ExampleService(BaseExampleService):
    async def ping(self: 'ExampleService', request: PingRequest) -> PingResponse:
        return PingResponse(id=request.id)

    async def complex(self: 'BaseExampleService', request: ComplexRequest) -> ComplexResponse:
        return ComplexResponse(**request.model_dump())

Run the ExampleService on Server.

from py_grpcio import BaseServer

from example.server.service import ExampleService


if __name__ == '__main__':
    server: BaseServer = BaseServer()
    server.add_service(service=ExampleService)
    server.run()

Note that on the client side, this class must be named the same as it is named in the full server-side implementation.

That is, if on the server we call the base class as BaseExampleService and the class with the implementation of methods as ExampleService, then on the client side the abstract service should be called ExampleService.

Client

from abc import abstractmethod

from py_grpcio import BaseService

from example.server.service.models import PingRequest, PingResponse, ComplexRequest, ComplexResponse


class ExampleService(BaseService):
    @abstractmethod
    async def ping(self: 'ExampleService', request: PingRequest) -> PingResponse:
        ...

    @abstractmethod
    async def complex(self: 'ExampleService', request: ComplexRequest) -> ComplexResponse:
        ...

Calling the ExampleService endpoints by Client.

from uuid import uuid4
from asyncio import run

from loguru import logger

from example.client.services.example.enums import Names
from example.client.services.example import (
    ExampleService, PingRequest, PingResponse, ComplexModel, ComplexRequest, ComplexResponse
)

service: ExampleService = ExampleService(host='127.0.0.1')


async def main() -> None:
    response: PingResponse = await service.ping(request=PingRequest())
    logger.info(f'ping response: {response}')

    response: ComplexResponse = await service.complex(
        request=ComplexRequest(id=uuid4(), model=ComplexModel(name=Names.NAME_1))
    )
    logger.info(f'complex response: {response}')


if __name__ == '__main__':
    run(main())

Notes

  • You can use the library on the client side even if the server is implemented differently by simply describing it as an abstract service

  • The client can also be implemented using other libraries, the server that uses py-grpcio will still be able to accept such requests

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

py_grpcio-1.4.4.tar.gz (10.8 kB view details)

Uploaded Source

Built Distribution

py_grpcio-1.4.4-py3-none-any.whl (14.0 kB view details)

Uploaded Python 3

File details

Details for the file py_grpcio-1.4.4.tar.gz.

File metadata

  • Download URL: py_grpcio-1.4.4.tar.gz
  • Upload date:
  • Size: 10.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.0 CPython/3.9.19

File hashes

Hashes for py_grpcio-1.4.4.tar.gz
Algorithm Hash digest
SHA256 a1f3068f4af02ae4cca90a48d58e5ec3e6a8db081a7c4b62937127bca5ac503b
MD5 8f185048c4b27cc3fde631ae7b1d0ed6
BLAKE2b-256 1c96f0b6cdf836695835ce75de7ba4cb8bdc005e8d0a5ea5a99a6ba58d07feee

See more details on using hashes here.

File details

Details for the file py_grpcio-1.4.4-py3-none-any.whl.

File metadata

  • Download URL: py_grpcio-1.4.4-py3-none-any.whl
  • Upload date:
  • Size: 14.0 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.0 CPython/3.9.19

File hashes

Hashes for py_grpcio-1.4.4-py3-none-any.whl
Algorithm Hash digest
SHA256 241c3f421d4555e2e1795c0c062531286ace0f63d988104eb8f47dd4b846b13b
MD5 d067518594aaf105e73c1aad096e93b1
BLAKE2b-256 dc31adcda124232a28ad90e3f912486f34982b01e029fb8c3f95a97d26f5ec5a

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page