mmar-pyhr 0.0.7

No project description provided

mmar-pyhr

Pydantic-HTTP-RPC - A lightweight RPC framework based on HTTP + FastAPI with Pydantic validation.

Overview

mmar-pyhr provides a simple RPC-style interface for service-to-service communication over HTTP, similar to mmar-ptag but using HTTP instead of gRPC. It was created to address Kubernetes compatibility issues with gRPC while maintaining the same developer-friendly interface.

Features

• Type-safe RPC using Pydantic for validation
• HTTP/JSON transport (Kubernetes-friendly, no service mesh required)
• Interface-based design using Python classes
• Built-in tracing with trace_id support via HTTP headers
• Health check endpoints for Kubernetes pod lifecycle management
• Smart address handling (port numbers, localhost, full URLs)
• Sync/async support for service methods

Installation

uv pip install mmar-pyhr

Quick Start

Server

from mmar_pyhr import PyhrServerConfig, deploy_pyhr_server


class UserService:
    def get_user(self, *, user_id: int, trace_id: str = "") -> dict:
        return {"id": user_id, "name": "John Doe"}

    def create_user(self, *, name: str, email: str, trace_id: str = "") -> dict:
        return {"id": 123, "name": name, "email": email}


if __name__ == "__main__":
    config = PyhrServerConfig(port=8000, host="0.0.0.0")
    deploy_pyhr_server(
        config_server=config,
        service=UserService()
    )

Client

from mmar_pyhr import pyhr_client


class UserServiceI:
    def get_user(self, *, user_id: int, trace_id: str = "") -> dict:
        raise NotImplementedError

    def create_user(self, *, name: str, email: str, trace_id: str = "") -> dict:
        raise NotImplementedError


# Create client
client = pyhr_client(UserServiceI, "localhost:8000")

# Make requests
user = client.get_user(user_id=42)
new_user = client.create_user(name="Alice", email="[e-mail]")

# With trace_id
user = client.get_user(user_id=42, trace_id="req-123")

Address Formats

The client accepts various address formats:

pyhr_client(Service, 9494)                      # → http://localhost:9494
pyhr_client(Service, ":9494")                    # → http://localhost:9494
pyhr_client(Service, "localhost:9494")           # → http://localhost:9494
pyhr_client(Service, "http://localhost:9494")    # → http://localhost:9494
pyhr_client(Service, "https://api.example.com")  # → https://api.example.com

API Routes

The server automatically creates the following routes:

RPC Endpoints

POST /api/{method_name}  → service.method_name(**kwargs)

Health Checks

GET  /health_check        → {"status": "ok"}
GET  /health/liveness     → {"status": "alive"}
GET  /health/readiness    → {"status": "ready"}

Tracing

Trace IDs are automatically propagated via the X-Trace-ID HTTP header:

# Client
client.get_user(user_id=42, trace_id="req-123")

# Server automatically receives trace_id in context
# and can use it for logging

Transport Abstraction

mmar-pyhr integrates with the mmar-mimpl transport registry:

from mmar_mimpl import get_transport

# Uses MMAR_TRANSPORT environment variable
transport = get_transport()  # "pyhr" or "mmar_pyhr"

# Create client
client = transport.make_client(UserServiceI, address="localhost:8000")

# Deploy server
transport.deploy_server(
    service=UserService(),
    config=None,
    config_server=PyhrServerConfig(port=8000)
)

Set the transport via environment variable:

export MMAR_TRANSPORT=pyhr
# or
export MMAR_TRANSPORT=mmar_pyhr

Configuration

Client Configuration

from mmar_pyhr import PyhrTransportConfig

config = PyhrTransportConfig(
    timeout=30.0,        # Request timeout in seconds
    retry_attempts=3,    # Number of retry attempts
)

Server Configuration

from mmar_pyhr import PyhrServerConfig

config = PyhrServerConfig(
    host="0.0.0.0",      # Bind host
    port=8000,           # Bind port
    max_workers=1,       # Number of worker processes
    log_level="info",    # Logging level
)

Comparison with mmar-ptag

Feature	mmar-ptag (gRPC)	mmar-pyhr (HTTP)
Transport	gRPC/HTTP2	HTTP/JSON
Serialization	JSON over Protobuf	Pure JSON
Kubernetes	Requires service mesh	Native support
Debugging	Requires grpcurl	Standard curl/httpie
Health checks	Custom implementation	Built-in HTTP endpoints
Load balancing	Connection-level (needs mesh)	Request-level (native K8s)

Development

# Run tests
uv run pytest

# Build package
uv build

License

MIT

See Also

• mmar-ptag - gRPC-based RPC framework
• Architecture Decision: gRPC to HTTP migration