GitHubEmailContact
Skip to main content

Getting Started with gRPC

Build your first production-ready gRPC service in 10 minutes.

📋 Prerequisites​

  • Django project with django-cfg installed
  • Python 3.8+
  • Basic understanding of Protocol Buffers

🚀 Quick Start​

Step 1: Install gRPC Dependencies​

# Install all gRPC packages at once
pip install django-cfg[grpc]

# Or install manually
pip install grpcio grpcio-tools grpcio-reflection grpcio-health-checking protobuf

Verify installation:

python -c "import grpc; print(f'gRPC {grpc.__version__}')"
# Output: gRPC 1.60.0

Step 2: Enable gRPC in Configuration​

# api/config.py
from typing import Optional
from django_cfg import (
    DjangoConfig,
    GRPCConfig,
    GRPCServerConfig,
    GRPCAuthConfig,
)

class MyConfig(DjangoConfig):
    # ... your existing config ...

    # Enable gRPC
    grpc: Optional[GRPCConfig] = GRPCConfig(
        enabled=True,
        server=GRPCServerConfig(
            host="[::]",              # Listen on all interfaces
            port=50051,               # Standard gRPC port
            max_workers=10,           # Thread pool size
            enable_reflection=True,   # Enable for grpcurl/testing
            enable_health_check=True, # Health check endpoint
        ),
        auth=GRPCAuthConfig(
            enabled=False,  # Disable auth for now (development)
            # Or enable API key auth:
            # enabled=True,
            # require_auth=False,
            # accept_django_secret_key=True,  # Allow SECRET_KEY for dev
        ),
        # Auto-discover services from these apps
        auto_register_apps=True,
        enabled_apps=[
            "apps.users",      # Your apps here
            "apps.products",
        ],
    )

Step 3: Run Migrations​

python manage.py migrate

This creates the GRPCRequestLog table for logging.

Step 4: Create Your First Service​

Create apps/users/grpc_services.py:

from django_cfg.apps.integrations.grpc.services import BaseService
from django.contrib.auth import get_user_model
from google.protobuf import empty_pb2
import grpc

User = get_user_model()

# Simple protobuf message classes (or import from generated .proto files)
class UserResponse:
    def __init__(self, id, username, email):
        self.id = id
        self.username = username
        self.email = email

class UserService(BaseService):
    """User management gRPC service."""

    def GetUser(self, request, context):
        """
        Get user by ID.

        Request: { user_id: int }
        Response: { id: int, username: str, email: str }
        """
        try:
            user = User.objects.get(id=request.user_id)

            # Return user data
            response = UserResponse(
                id=user.id,
                username=user.username,
                email=user.email or "",
            )
            return response

        except User.DoesNotExist:
            # Abort with NOT_FOUND status
            self.abort_not_found(context, f"User {request.user_id} not found")

    def ListUsers(self, request, context):
        """
        List all users (streaming response).

        Request: empty or { limit: int }
        Response: stream of User objects
        """
        # Get users
        users = User.objects.filter(is_active=True)

        # Apply limit if provided
        limit = getattr(request, 'limit', 100)
        users = users[:limit]

        # Stream users one by one
        for user in users:
            yield UserResponse(
                id=user.id,
                username=user.username,
                email=user.email or "",
            )

    def CreateUser(self, request, context):
        """
        Create new user (requires authentication).

        Request: { username: str, email: str, password: str }
        Response: { id: int, username: str, email: str }
        """
        # Require authentication (if auth enabled)
        # user = self.require_user(context)

        # Validate
        if not request.username:
            self.abort_invalid_argument(context, "Username is required")

        # Check if exists
        if User.objects.filter(username=request.username).exists():
            self.abort_already_exists(context, "Username already taken")

        # Create user
        user = User.objects.create_user(
            username=request.username,
            email=request.email,
            password=request.password,
        )

        return UserResponse(
            id=user.id,
            username=user.username,
            email=user.email,
        )

That's it! The service is automatically discovered and registered.

Step 5: Start gRPC Server​

python manage.py rungrpc

Expected output:

🚀 Starting gRPC server...
📡 Server running at [::]:50051
🔍 Reflection enabled
❤️  Health check enabled
✅ Registered 1 service:
   - api.users.UserService

Step 6: Test Your Service​

Option 1: Using grpcurl (recommended for testing)

# List services
grpcurl -plaintext localhost:50051 list

# Output:
# api.users.UserService
# grpc.health.v1.Health
# grpc.reflection.v1alpha.ServerReflection

# Describe service
grpcurl -plaintext localhost:50051 describe api.users.UserService

# Call GetUser
grpcurl -plaintext -d '{"user_id": 1}' \
  localhost:50051 api.users.UserService/GetUser

# Output (example):
# {
#   "id": 1,
#   "username": "admin",
#   "email": "admin@example.com"
# }

Option 2: Using Python client

# test_client.py
import grpc
from django_cfg.apps.integrations.grpc.services.grpc_client import DynamicGRPCClient

# Create client
client = DynamicGRPCClient(host="localhost", port=50051)

# Invoke method
response = client.invoke_method(
    service="api.users.UserService",
    method="GetUser",
    request_data={"user_id": 1}
)

print(response)
# {'id': 1, 'username': 'admin', 'email': 'admin@example.com'}

Option 3: Django Admin

View request logs in Django Admin:

http://localhost:8000/admin/integrations/grpc/grpcrequestlog/

🎯 Add Authentication​

API Key Authentication​

API keys are perfect for service-to-service communication and CLI tools.

Step 1: Enable API Key Auth​

# api/config.py
grpc: GRPCConfig = GRPCConfig(
    auth=GRPCAuthConfig(
        enabled=True,
        require_auth=False,  # Optional auth (some methods public)
        api_key_header="x-api-key",
        accept_django_secret_key=True,  # Allow SECRET_KEY for dev/internal
        public_methods=[
            "/grpc.health.v1.Health/Check",
        ],
    ),
)

Step 2: Create API Key​

Via Django Admin:

1. Open /admin/integrations/grpc/grpcapikey/
2. Click "Add gRPC API Key"
3. Fill in:
   - Name: "Analytics Service"
   - Description: "Internal analytics microservice"
   - User: Select user to authenticate as
   - Type: Service
   - Expires at: Optional (e.g., 1 year from now)
4. Click "Save"
5. Copy the generated API key from the detail page

Or programmatically:

from django_cfg.apps.integrations.grpc.models import GrpcApiKey
from django.contrib.auth import get_user_model

User = get_user_model()
admin_user = User.objects.filter(is_superuser=True).first()

# Create API key
api_key = GrpcApiKey.objects.create_for_user(
    user=admin_user,
    name="Analytics Service",
    description="Internal analytics microservice",
    key_type="service",
    expires_in_days=365,  # Expires in 1 year
)

print(f"API Key: {api_key.key}")
# Save this key securely!

Step 3: Use API Key​

With grpcurl:

grpcurl -plaintext \
  -H "x-api-key: <your_api_key_here>" \
  -d '{"user_id": 1}' \
  localhost:50051 api.users.UserService/GetUser

With Python client:

import grpc

channel = grpc.insecure_channel('localhost:50051')
metadata = [('x-api-key', 'your_api_key_here')]

stub = UserServiceStub(channel)
response = stub.GetUser(request, metadata=metadata)

For development/internal use with SECRET_KEY:

# Use Django SECRET_KEY as API key (convenient for dev)
grpcurl -plaintext \
  -H "x-api-key: $(python manage.py shell -c 'from django.conf import settings; print(settings.SECRET_KEY)')" \
  localhost:50051 api.users.UserService/GetUser

Step 4: Access User in Service​

class UserService(BaseService):
    def UpdateProfile(self, request, context):
        # Get authenticated user (from API key)
        user = self.require_user(context)

        # Get API key used for authentication
        api_key = getattr(context, 'api_key', None)
        if api_key:
            print(f"Request from API key: {api_key.name}")

        # Update profile
        user.bio = request.bio
        user.save()

        return UserResponse(...)

Step 5: Monitor Usage​

View API key usage in Django Admin:

/admin/integrations/grpc/grpcapikey/

Each API key shows:

  • Total request count
  • Last used timestamp
  • Status (active/expired/revoked)
  • Associated request logs

Or programmatically:

from django_cfg.apps.integrations.grpc.models import GrpcApiKey

# Get API key
api_key = GrpcApiKey.objects.get(name="Analytics Service")

print(f"Request count: {api_key.request_count}")
print(f"Last used: {api_key.last_used_at}")
print(f"Is valid: {api_key.is_valid}")

# View requests made with this key
logs = api_key.request_logs.all()[:10]

📊 Monitor Requests​

View in Django Admin​

All requests are automatically logged:

  1. Open Django Admin: http://localhost:8000/admin/
  2. Go to: Integrations → gRPC Request Logs
  3. Filter by:
    • Service name
    • Method name
    • Status (success/error)
    • User
    • Date range

Query Programmatically​

from django_cfg.apps.integrations.grpc.models import GRPCRequestLog

# Recent requests
recent = GRPCRequestLog.objects.recent(hours=1)

# Errors only
errors = GRPCRequestLog.objects.filter(status='error')

# Statistics
stats = GRPCRequestLog.objects.get_statistics(hours=24)
print(f"Success rate: {stats['success_rate']}%")
print(f"Avg duration: {stats['avg_duration_ms']}ms")

# Per-service stats
for service in ['UserService', 'ProductService']:
    logs = GRPCRequestLog.objects.by_service(service)
    print(f"{service}: {logs.count()} requests")

🛠️ Development Tips​

1. Enable Debug Logging​

# settings.py
LOGGING = {
    'loggers': {
        'django_cfg.apps.integrations.grpc': {
            'level': 'DEBUG',
            'handlers': ['console'],
        },
    },
}

2. Hot Reload in Development​

The server automatically reloads when you change code (like Django runserver).

3. Use Base Service Helpers​

class MyService(BaseService):
    def MyMethod(self, request, context):
        # Helpers available:
        self.get_user(context)           # Optional user
        self.require_user(context)       # Required user
        self.require_staff(context)      # Staff only
        self.require_superuser(context)  # Superuser only
        self.check_permission(context, "app.perm")

        # Abort methods:
        self.abort_not_found(context, "Not found")
        self.abort_permission_denied(context, "No access")
        self.abort_invalid_argument(context, "Bad request")
        self.abort_already_exists(context, "Duplicate")

4. Test with Different Ports​

# Run on different port
python manage.py rungrpc --port 50052

# Run with more workers
python manage.py rungrpc --workers 20

🎓 Common Patterns​

Pattern 1: CRUD Service​

class ProductService(BaseService):
    def GetProduct(self, request, context):
        # Read
        product = Product.objects.get(id=request.id)
        return product_pb2.Product(...)

    def CreateProduct(self, request, context):
        # Create
        user = self.require_user(context)
        product = Product.objects.create(
            name=request.name,
            price=request.price,
        )
        return product_pb2.Product(...)

    def UpdateProduct(self, request, context):
        # Update
        user = self.require_user(context)
        product = Product.objects.get(id=request.id)
        product.name = request.name
        product.save()
        return product_pb2.Product(...)

    def DeleteProduct(self, request, context):
        # Delete
        self.require_staff(context)
        Product.objects.get(id=request.id).delete()
        return empty_pb2.Empty()

Pattern 2: Streaming Service​

class EventService(BaseService):
    def StreamEvents(self, request, context):
        # Server-side streaming
        events = Event.objects.filter(user_id=request.user_id)

        for event in events:
            yield event_pb2.Event(
                id=event.id,
                type=event.type,
                data=event.data,
            )

Pattern 3: Read-Only Service​

from django_cfg.apps.integrations.grpc.services import ReadOnlyService

class CatalogService(ReadOnlyService):
    """Read-only catalog browsing."""

    def SearchProducts(self, request, context):
        # Only queries allowed (no writes)
        products = Product.objects.filter(
            category=request.category,
            active=True,
        )[:100]

        for product in products:
            yield product_pb2.Product(...)

🚨 Common Issues​

Issue: Port already in use

# Kill existing process
lsof -ti :50051 | xargs kill -9

# Or use different port
python manage.py rungrpc --port 50052

Issue: Service not found

  • Check file name: grpc_services.py (not grpc_service.py)
  • Check app in enabled_apps
  • Check class inherits from BaseService

Issue: Import errors

# Wrong import
from django_cfg.apps.grpc.services import BaseService  # ❌

# Correct import
from django_cfg.apps.integrations.grpc.services import BaseService  # ✅

📚 Next Steps​

Now that you have a working service:

  1. Concepts - Understand architecture and patterns
  2. Authentication - Deep dive into API keys
  3. Configuration - Explore all config options
  4. Dynamic Invocation - Test without proto files
  5. FAQ - Common questions and solutions

💡 Key Takeaways​

  • ✅ Services are auto-discovered from grpc_services.py
  • ✅ Use base classes for common patterns
  • ✅ Django ORM works out of the box
  • ✅ API key auth is simple and secure with admin interface
  • ✅ All requests are automatically logged with API key tracking
  • ✅ Server monitoring tracks uptime and registered services
  • ✅ Use grpcurl for testing during development

Congratulations! You've built your first production-ready gRPC service with Django-CFG. 🎉

Last updated on