mkdir grpc-ticket-service
cd grpc-ticket-service
virtualenv --python=python3.14 .venv
source .venv/bin/activate
pip install --upgrade pip

grpc-ticket-service/
├── definitions/
│   ├── __init__.py
│   ├── service.proto           # Protobuf definitions
│   └── builds/
│       ├── __init__.py
│       ├── service_pb2.py      # Generated message classes
│       ├── service_pb2.pyi     # Generated type stubs
│       └── service_pb2_grpc.py # Generated service classes
├── server.py                   # gRPC server implementation
├── client.py                   # gRPC client example
└── requirements.txt

syntax = "proto3";

// Empty message for health check
message Null {}

// Message representing a ticket
message Ticket {
    string name = 1;
    string description = 2;
    uint32 story_points = 3;
}

// Response message with expected deadline
message Confirmation {
    string expected_dateline = 1;
}

// Service definition
service TicketService {
    // Health check endpoint
    rpc Health(Null) returns (Null);

    // Add a new ticket and get estimated completion date
    rpc AddTicket(Ticket) returns (Confirmation);
}

pip install grpcio grpcio-tools googleapis-common-protos mypy-protobuf

python -m grpc_tools.protoc \
    -I definitions/ \
    --python_out=definitions/builds/ \
    --grpc_python_out=definitions/builds/ \
    --mypy_out=definitions/builds/ \
    definitions/service.proto

# server.py
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone

import grpc

from definitions.builds.service_pb2 import Confirmation
from definitions.builds.service_pb2_grpc import (
    TicketServiceServicer,
    add_TicketServiceServicer_to_server
)


class TicketService(TicketServiceServicer):
    """Implementation of TicketService defined in .proto file"""

    def Health(self, request, context):
        """Health check endpoint - returns empty message"""
        print("Health check called")
        return request

    def AddTicket(self, request, context):
        """
        Add a ticket and calculate expected deadline
        based on story points (1 point = 1 day)
        """
        print(f"Adding ticket: {request.name}")
        print(f"  Description: {request.description}")
        print(f"  Story Points: {request.story_points}")

        # Calculate deadline: current time + story_points days
        expected_dateline = datetime.now(timezone.utc) + timedelta(days=request.story_points)

        return Confirmation(
            expected_dateline=expected_dateline.strftime("%Y-%m-%d %H:%M:%S")
        )


def serve():
    """Start the gRPC server"""
    # Create server with thread pool
    server = grpc.server(ThreadPoolExecutor(max_workers=10))

    # Register our service implementation
    add_TicketServiceServicer_to_server(TicketService(), server)

    # Listen on port 50051
    server.add_insecure_port("[::]:50051")

    # Start the server
    server.start()
    print("The server is up and running...")

    # Keep server running
    server.wait_for_termination()


if __name__ == "__main__":
    serve()

python server.py

The server is up and running...

# client.py
import grpc

from definitions.builds.service_pb2 import Null, Ticket
from definitions.builds.service_pb2_grpc import TicketServiceStub


def main():
    # Create a channel to the server
    with grpc.insecure_channel("localhost:50051") as channel:
        # Create a stub (client)
        client = TicketServiceStub(channel)

        # Call Health check
        print("Calling Health()...")
        client.Health(Null())
        print("✓ Health check passed\n")

        # Call AddTicket
        print("Adding a ticket...")
        confirmation = client.AddTicket(Ticket(
            name="Implement user authentication",
            description="Add JWT-based authentication to the API",
            story_points=3
        ))

        print(f"✓ Ticket added successfully!")
        print(f"  Expected deadline: {confirmation.expected_dateline}")


if __name__ == "__main__":
    main()

python client.py

Calling Health()...
✓ Health check passed

Adding a ticket...
✓ Ticket added successfully!
Expected deadline: 2026-12-31 00:00:00

The server is up and running...
Health check called
Adding ticket: Implement user authentication
  Description: Add JWT-based authentication to the API
  Story Points: 3

syntax = "proto3";

message TaskRequest {
    string project_id = 1;
}

message Task {
    string id = 1;
    string title = 2;
    string status = 3;
}

service TaskService {
    // Server streams all tasks for a project
    rpc ListTasks(TaskRequest) returns (stream Task);
}

import time
from definitions.builds.service_pb2 import Task
from definitions.builds.service_pb2_grpc import TaskServiceServicer

class TaskService(TaskServiceServicer):
    def ListTasks(self, request, context):
        """Stream tasks one by one"""
        tasks = [
            {"id": "1", "title": "Design API", "status": "done"},
            {"id": "2", "title": "Implement auth", "status": "in_progress"},
            {"id": "3", "title": "Write tests", "status": "todo"},
        ]

        for task_data in tasks:
            # Simulate processing delay
            time.sleep(0.5)

            # Yield each task (streaming)
            yield Task(
                id=task_data["id"],
                title=task_data["title"],
                status=task_data["status"]
            )

from definitions.builds.service_pb2 import TaskRequest
from definitions.builds.service_pb2_grpc import TaskServiceStub

with grpc.insecure_channel("localhost:50051") as channel:
    client = TaskServiceStub(channel)

    # Call streaming RPC
    task_stream = client.ListTasks(TaskRequest(project_id="proj_123"))

    # Iterate over the stream
    print("Receiving tasks...")
    for task in task_stream:
        print(f"  [{task.status}] {task.title} (ID: {task.id})")

Receiving tasks...
[done] Design API (ID: 1)
[in_progress] Implement auth (ID: 2)
[todo] Write tests (ID: 3)

syntax = "proto3";

message ChatMessage {
    string username = 1;
    string text = 2;
    int64 timestamp = 3;
}

service ChatService {
    // Bidirectional streaming chat
    rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

from definitions.builds.service_pb2_grpc import ChatServiceServicer
import time

class ChatService(ChatServiceServicer):
    def __init__(self):
        self.clients = []

    def Chat(self, request_iterator, context):
        """Bidirectional streaming chat"""
        # Add this client to the list
        self.clients.append(context)

        try:
            # Read messages from client
            for message in request_iterator:
                print(f"{message.username}: {message.text}")

                # Broadcast to all connected clients
                for client_context in self.clients:
                    if client_context != context:
                        yield message

        finally:
            # Remove client on disconnect
            self.clients.remove(context)

import threading
import time
from definitions.builds.service_pb2 import ChatMessage
from definitions.builds.service_pb2_grpc import ChatServiceStub

def send_messages(stub):
    """Send messages to server"""
    messages = [
        ChatMessage(username="alice", text="Hello!", timestamp=int(time.time())),
        ChatMessage(username="alice", text="How are you?", timestamp=int(time.time())),
    ]

    for msg in messages:
        yield msg
        time.sleep(1)

def run_chat():
    with grpc.insecure_channel("localhost:50051") as channel:
        client = ChatServiceStub(channel)

        # Start bidirectional stream
        responses = client.Chat(send_messages(client))

        # Listen for responses
        for response in responses:
            print(f"Received: {response.username}: {response.text}")

run_chat()

# grpc_benchmark.py
from time import time
import grpc

from definitions.builds.service_pb2 import Null, Ticket
from definitions.builds.service_pb2_grpc import TicketServiceStub


def main():
    with grpc.insecure_channel("localhost:50051") as channel:
        client = TicketServiceStub(channel)

        # Warm up
        client.Health(Null())

        # Benchmark
        start = time()

        for _ in range(2000):
            client.AddTicket(Ticket(
                name="SomeTicket",
                description="...",
                story_points=2
            ))

        elapsed = time() - start
        print(f"gRPC: {elapsed:.4f} seconds")
        print(f"Throughput: {2000/elapsed:.2f} req/s")


if __name__ == "__main__":
    main()

# fastapi_server.py
import json
from datetime import datetime, timedelta, timezone

import uvicorn
from fastapi import FastAPI
from starlette.requests import Request

app = FastAPI()


@app.get("/health")
def health():
    return {"status": "ok"}


@app.post("/ticket")
async def manage_tickets(request: Request):
    body = json.loads(await request.body())
    points = body.get("story_points")
    expected_dateline = datetime.now(timezone.utc) + timedelta(days=points)
    return {"expected_dateline": expected_dateline.strftime("%Y-%m-%d %H:%M:%S")}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=3001)

python fastapi_server.py

# flask_server.py
import json
from datetime import datetime, timedelta, timezone

from flask import Flask, request

app = Flask(__name__)


@app.route("/")
def health():
    return "OK"


@app.route("/ticket", methods=["POST"])
def manage_tickets():
    points = json.loads(request.data).get("story_points")
    expected_dateline = datetime.now(timezone.utc) + timedelta(days=points)
    return {"expected_dateline": expected_dateline.strftime("%Y-%m-%d %H:%M:%S")}


if __name__ == "__main__":
    app.run(port=3001)

python flask_server.py

# rest_benchmark.py
import json
from time import time

import urllib3

http = urllib3.PoolManager()


def main():
    start = time()

    for _ in range(2000):
        http.request(
            "POST",
            "http://localhost:3001/ticket",
            headers={"Content-Type": "application/json"},
            body=json.dumps({
                "name": "x",
                "description": "...",
                "story_points": 3
            })
        )

    elapsed = time() - start
    print(f"REST: {elapsed:.4f} seconds")
    print(f"Throughput: {2000/elapsed:.2f} req/s")


if __name__ == "__main__":
    main()

import grpc
from concurrent.futures import ThreadPoolExecutor

def serve_with_tls():
    # Read TLS credentials
    with open("server.key", "rb") as f:
        private_key = f.read()
    with open("server.crt", "rb") as f:
        certificate_chain = f.read()

    # Create server credentials
    server_credentials = grpc.ssl_server_credentials(
        [(private_key, certificate_chain)]
    )

    # Create server
    server = grpc.server(ThreadPoolExecutor(max_workers=10))
    add_TicketServiceServicer_to_server(TicketService(), server)

    # Add secure port
    server.add_secure_port("[::]:50051", server_credentials)

    server.start()
    print("Secure gRPC server started on port 50051")
    server.wait_for_termination()

import grpc

# Read root certificates
with open("ca.crt", "rb") as f:
    trusted_certs = f.read()

# Create credentials
credentials = grpc.ssl_channel_credentials(root_certificates=trusted_certs)

# Create secure channel
with grpc.secure_channel("api.example.com:50051", credentials) as channel:
    client = TicketServiceStub(channel)
    response = client.AddTicket(ticket)

import grpc
from definitions.builds.service_pb2_grpc import TicketServiceServicer

class TicketService(TicketServiceServicer):
    def AddTicket(self, request, context):
        # Validate input
        if not request.name:
            context.abort(
                grpc.StatusCode.INVALID_ARGUMENT,
                "Ticket name is required"
            )

        if request.story_points < 1 or request.story_points > 13:
            context.abort(
                grpc.StatusCode.INVALID_ARGUMENT,
                "Story points must be between 1 and 13"
            )

        try:
            # Business logic
            deadline = calculate_deadline(request.story_points)
            return Confirmation(expected_dateline=deadline)

        except Exception as e:
            # Log the error
            print(f"Error: {e}")

            # Return generic error to client
            context.abort(
                grpc.StatusCode.INTERNAL,
                "An internal error occurred"
            )

import grpc

try:
    response = client.AddTicket(ticket)
    print(f"Success: {response.expected_dateline}")

except grpc.RpcError as e:
    # Check status code
    if e.code() == grpc.StatusCode.INVALID_ARGUMENT:
        print(f"Invalid input: {e.details()}")
    elif e.code() == grpc.StatusCode.NOT_FOUND:
        print(f"Not found: {e.details()}")
    elif e.code() == grpc.StatusCode.UNAVAILABLE:
        print("Service unavailable, retrying...")
        # Implement retry logic
    else:
        print(f"Error: {e.code()} - {e.details()}")

pip install grpcio-health-checking

import grpc
from grpc_health.v1 import health, health_pb2, health_pb2_grpc
from concurrent.futures import ThreadPoolExecutor

def serve_with_health_check():
    server = grpc.server(ThreadPoolExecutor(max_workers=10))

    # Add your service
    add_TicketServiceServicer_to_server(TicketService(), server)

    # Add health check service
    health_servicer = health.HealthServicer()
    health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server)

    # Set service status to SERVING
    health_servicer.set("TicketService", health_pb2.HealthCheckResponse.SERVING)

    server.add_insecure_port("[::]:50051")
    server.start()
    server.wait_for_termination()

import grpc

with grpc.insecure_channel("localhost:50051") as channel:
client = TicketServiceStub(channel)

try:
    # Set 5 second deadline
    response = client.AddTicket(
        ticket,
        timeout=5.0  # 5 seconds
    )
    print(f"Response: {response.expected_dateline}")

except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
        print("Request timed out after 5 seconds")

pip install prometheus-client py-grpc-prometheus

import grpc
from py_grpc_prometheus.prometheus_server_interceptor import PromServerInterceptor
from prometheus_client import start_http_server

def serve_with_metrics():
# Start Prometheus metrics server on port 8000
start_http_server(8000)

# Create interceptor
interceptor = PromServerInterceptor()

# Create server with interceptor
server = grpc.server(
    ThreadPoolExecutor(max_workers=10),
    interceptors=[interceptor]
)

add_TicketServiceServicer_to_server(TicketService(), server)
server.add_insecure_port("[::]:50051")

server.start()
print("gRPC server on :50051, metrics on :8000")
server.wait_for_termination()