# kv-store-grpc

> Guide for implementing gRPC-based key-value store services in Python. This skill should be used when building gRPC servers with protobuf definitions, implementing KV store operations (Get, Set, Delete), or troubleshooting gRPC service connectivity. Applicable to tasks involving grpcio, protobuf code generation, and background server processes.

- Author: nouamanecodes
- Repository: nouamanecodes/skills
- Version: 20260113202157
- Stars: 0
- Forks: 0
- Last Updated: 2026-02-07
- Source: https://github.com/nouamanecodes/skills
- Web: https://mule.run/skillshub/@@nouamanecodes/skills~kv-store-grpc:20260113202157

---

---
name: kv-store-grpc
description: Guide for implementing gRPC-based key-value store services in Python. This skill should be used when building gRPC servers with protobuf definitions, implementing KV store operations (Get, Set, Delete), or troubleshooting gRPC service connectivity. Applicable to tasks involving grpcio, protobuf code generation, and background server processes.
---

# gRPC Key-Value Store Implementation

## Overview

This skill provides procedural guidance for implementing gRPC-based key-value store services in Python. It covers the full workflow from protobuf definition to server implementation and verification, with emphasis on avoiding common pitfalls.

## Implementation Workflow

### Step 1: Install Dependencies

Install the required gRPC packages:

```bash
pip install grpcio grpcio-tools
```

**Verification**: After installation, confirm packages are available:
```bash
pip list | grep grpc
```

Do not proceed until installation is verified. Missing or incorrect package versions cause silent failures during code generation.

### Step 2: Define the Protocol Buffer

Create a `.proto` file with the service definition. Key considerations:

**Type Selection for Values:**
- Use `int32` for integer values (standard choice for most use cases)
- Use `int64` if values may exceed 32-bit range
- Use `string` for keys (allows flexible key naming)

**Message Design Pattern:**
```protobuf
syntax = "proto3";

message SetRequest {
    string key = 1;
    int32 value = 2;
}

message SetResponse {
    bool success = 1;
}

message GetRequest {
    string key = 1;
}

message GetResponse {
    int32 value = 1;
    bool found = 2;  // Consider adding to distinguish missing keys from zero values
}
```

**Edge Case Consideration**: When returning values for non-existent keys, returning a default value (e.g., 0) is ambiguous. Consider:
- Adding a `found` or `exists` boolean field to responses
- Using gRPC status codes to indicate missing keys
- Documenting the chosen behavior explicitly

### Step 3: Generate Python Code

Run the protobuf compiler to generate Python gRPC code:

```bash
python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. <filename>.proto
```

This generates two files:
- `<filename>_pb2.py` - Message classes
- `<filename>_pb2_grpc.py` - Service stubs and servicer base classes

**Verification**: Confirm both files were generated before proceeding:
```bash
ls -la *_pb2*.py
```

### Step 4: Implement the Server

Create the server implementation with these considerations:

**Server Class Structure:**
```python
class KVStoreServicer(kv_pb2_grpc.KVStoreServicer):
    def __init__(self):
        self.store = {}

    def SetVal(self, request, context):
        self.store[request.key] = request.value
        return kv_pb2.SetResponse(success=True)

    def GetVal(self, request, context):
        value = self.store.get(request.key, 0)
        # Document: Returns 0 for non-existent keys
        return kv_pb2.GetResponse(value=value)
```

**Server Initialization:**
```python
def serve(port):
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    kv_pb2_grpc.add_KVStoreServicer_to_server(KVStoreServicer(), server)
    server.add_insecure_port(f'[::]:{port}')
    server.start()
    print(f"Server started on port {port}", flush=True)  # flush=True is critical
    server.wait_for_termination()
```

**Critical: Use `flush=True`** for print statements when running servers in background processes. Without flushing, output may not appear immediately, making it difficult to verify server startup.

### Step 5: Verify Server Operation

**Prefer functional testing over system diagnostics.** Testing with an actual gRPC client is the definitive verification method.

**Avoid this approach** (unreliable, tool-dependent):
```bash
# These may not be available and don't confirm gRPC functionality
netstat -tlnp | grep <port>
ss -tlnp | grep <port>
lsof -i :<port>
```

**Use this approach** (direct functional test):
```python
import grpc
import kv_pb2
import kv_pb2_grpc

channel = grpc.insecure_channel('localhost:<port>')
stub = kv_pb2_grpc.KVStoreStub(channel)

# Test Set operation
response = stub.SetVal(kv_pb2.SetRequest(key="test", value=42))
print(f"Set success: {response.success}")

# Test Get operation
response = stub.GetVal(kv_pb2.GetRequest(key="test"))
print(f"Got value: {response.value}")
```

Run inline if possible to avoid creating temporary test files:
```bash
python -c "import grpc; ..."
```

## Common Pitfalls

### 1. Missing Key Ambiguity
Returning default values (0, empty string) for missing keys is indistinguishable from keys with those actual values. Design the protocol to handle this explicitly.

### 2. Output Buffering in Background Processes
Print statements without `flush=True` may never appear when running servers in background. Always flush output for startup confirmation messages.

### 3. Excessive Diagnostic Commands
Avoid chaining system utilities (netstat, ss, lsof, ps) to verify server status. These tools may not be installed and don't confirm gRPC protocol functionality. Use a real client instead.

### 4. Unverified Package Installation
Always verify pip installations succeeded before proceeding. Silent installation failures cause confusing errors during code generation or runtime.

### 5. Missing Error Handling
Consider adding try-except blocks around RPC method implementations to handle unexpected errors gracefully and return appropriate gRPC status codes.

## Verification Checklist

Before considering the task complete:

- [ ] Dependencies installed and verified with `pip list`
- [ ] Proto file compiles without errors
- [ ] Both `_pb2.py` and `_pb2_grpc.py` files generated
- [ ] Server starts and prints confirmation (with `flush=True`)
- [ ] Functional test with gRPC client succeeds for all operations
- [ ] Edge cases documented (missing keys, zero values)