asyncgrpcclient

Async gRPC client wrapper над grpc.aio — callable URL и Metadata, error mapping (StatusCode → exceptions), endpoint-aliasing, чистый API в стиле async REST-клиента.

Установка

pip install asyncgrpcclient

Возможности

• ✅ Тонкая обёртка над grpc.aio.insecure_channel / secure_channel
• ✅ Endpoints как именованные алиасы методов на gRPC stub
• ✅ Per-call metadata override (поверх default)
• ✅ Error mapping: grpc.aio.AioRpcError → GrpcQueryError
• ✅ Options с clone() / merge() для гибкой настройки timeout / status codes / extra kwargs
• ✅ full_response режим (вернуть GrpcResponse или сам proto)

Быстрый старт

from asyncgrpcclient import AsyncGrpcClient

# stub сгенерирован из proto (grpc_tools.protoc → my_service_pb2_grpc.py)
from my_service_pb2_grpc import MyServiceStub
from my_service_pb2 import GetUserRequest

client = AsyncGrpcClient(
    stub=MyServiceStub,
    address="my-service:50051",
    endpoints={
        "get_user": "GetUser",
        "list_users": "ListUsers",
    },
    metadata={"x-tenant-id": "default"},
    timeout=5.0,
)

# через alias
user = await client("get_user", GetUserRequest(id=1))

# или напрямую через имя метода stub
user = await client("GetUser", GetUserRequest(id=1))

# с per-call metadata
user = await client(
    "get_user",
    GetUserRequest(id=1),
    metadata={"x-trace-id": "abc-123"},
)

# полный response объект
response = await client("get_user", GetUserRequest(id=1), full_response=True)
print(response.status_code, response.metadata)

await client.close()

Secure channel

import grpc

creds = grpc.ssl_channel_credentials(root_certificates=ca_pem)

client = AsyncGrpcClient(
    stub=MyServiceStub,
    address="my-service.example.com:443",
    secure=True,
    credentials=creds,
)

Error handling

from asyncgrpcclient import GrpcQueryError

try:
    user = await client("get_user", GetUserRequest(id=999))
except GrpcQueryError as exc:
    print(exc.code)   # grpc.StatusCode (NOT_FOUND, UNAVAILABLE, ...)
    print(exc.name)
    print(exc.message)

Options через ok_statuses

По умолчанию OK-статус: (grpc.StatusCode.OK,). Если нужно принять NOT_FOUND как валидный ответ:

import grpc

response = await client(
    "get_user",
    GetUserRequest(id=999),
    ok_statuses=[grpc.StatusCode.OK, grpc.StatusCode.NOT_FOUND],
    full_response=True,
)
if response.status_code == grpc.StatusCode.NOT_FOUND:
    user = None

API

Аналог

API параллелен sync REST-клиентам. Idiomatic для команд, у которых REST и gRPC живут вместе и хочется одинаковый стиль вызова.

License

MIT