Async gRPC Server

┌─────────────────────────────────────────┐
│   python manage.py rungrpc              │
│                                         │
│  ┌─────────────────────────────────┐   │
│  │ grpc.aio.server() [ASYNC]       │   │
│  │  └─ asyncio event loop          │   │
│  │     ├─ API Key Auth (async)     │   │
│  │     ├─ Logging (async)          │   │
│  │     ├─ Streaming Services       │   │
│  │     └─ Unary Services           │   │
│  └─────────────────────────────────┘   │
└─────────────────────────────────────────┘

class UserService(BaseService):
    async def GetUser(self, request, context):
        user = await User.objects.aget(id=request.user_id)
        return UserResponse(id=user.id, name=user.name)

class EventService(BaseService):
    async def StreamEvents(self, request, context):
        """Stream events to client."""
        async for event in Event.objects.filter(
            user_id=request.user_id
        ).aiterator():
            yield EventResponse(
                id=event.id,
                type=event.type,
                data=event.data,
            )
            await asyncio.sleep(0)  # Yield to event loop

class LogService(BaseService):
    async def UploadLogs(self, request_iterator, context):
        """Receive stream of logs from client."""
        count = 0
        async for log_entry in request_iterator:
            await Log.objects.acreate(
                message=log_entry.message,
                level=log_entry.level,
            )
            count += 1
 
        return UploadResponse(logs_received=count)

class ChatService(BaseService):
    async def Chat(self, request_iterator, context):
        """Bidirectional chat stream."""
        async for message in request_iterator:
            # Process incoming message
            await Message.objects.acreate(
                text=message.text,
                user_id=message.user_id,
            )
 
            # Send response
            yield ChatResponse(
                text=f"Received: {message.text}",
                timestamp=int(time.time()),
            )

from django_cfg.apps.integrations.grpc.services.streaming import (
    BidirectionalStreamingService,
    BidirectionalStreamingConfig,
    StreamingMode,
    PingStrategy,
)
 
class BotService(BidirectionalStreamingService):
    config = BidirectionalStreamingConfig(
        streaming_mode=StreamingMode.ANEXT,
        ping_strategy=PingStrategy.INTERVAL,
        ping_interval=5.0,
        ping_timeout=180.0,
        connection_timeout=None,  # CRITICAL: Must be None!
    )
 
    async def on_client_connected(self, client_id: str, context):
        logger.info(f"Client {client_id} connected")
 
    async def on_message_received(self, client_id: str, message, context):
        # Handle message based on type
        return await self.dispatch_message(message)

# Async ORM methods
user = await User.objects.aget(id=1)
users = [user async for user in User.objects.filter(active=True).aiterator()]
await User.objects.acreate(name="John")
await user.asave()
await user.adelete()
 
# Async related objects
profile = await user.profile.aget()
orders = [order async for order in user.orders.all().aiterator()]

import asyncio
 
# Wrap sync ORM in to_thread
count = await asyncio.to_thread(User.objects.count)
exists = await asyncio.to_thread(User.objects.filter(email=email).exists)
 
# Complex queries
result = await asyncio.to_thread(
    lambda: User.objects.select_related('profile')
                        .prefetch_related('orders')
                        .get(id=user_id)
)

from django_cfg import DjangoConfig, GRPCConfig, GRPCServerConfig
 
class MyConfig(DjangoConfig):
    grpc = GRPCConfig(
        enabled=True,
        server=GRPCServerConfig(
            host="[::]",
            port=50051,
 
            # Async options
            max_concurrent_streams=1000,
            asyncio_debug=False,
 
            # Standard options
            enable_reflection=True,
            enable_health_check=True,
        ),
    )

from django_cfg.apps.integrations.grpc.services.streaming.config import (
    BidirectionalStreamingConfig,
    StreamingMode,
    PingStrategy,
)
 
config = BidirectionalStreamingConfig(
    streaming_mode=StreamingMode.ANEXT,
    ping_strategy=PingStrategy.INTERVAL,
    ping_interval=5.0,
    ping_timeout=180.0,
    connection_timeout=None,  # CRITICAL!
    max_queue_size=1000,
    enable_sleep_zero=True,
    enable_logging=True,
)

async def StreamData(self, request, context):
    for item in large_dataset:
        yield ItemResponse(data=item)
        await asyncio.sleep(0)  # Yield to event loop

async def StreamEvents(self, request, context):
    while not context.cancelled():
        events = await self.get_new_events()
        for event in events:
            yield EventResponse(event=event)
        await asyncio.sleep(1)

async def ProcessStream(self, request_iterator, context):
    async with self.get_async_connection() as conn:
        async for request in request_iterator:
            await conn.process(request)

async def UploadItems(self, request_iterator, context):
    batch = []
    async for item in request_iterator:
        batch.append(Item(data=item.data))
        if len(batch) >= 100:
            await Item.objects.abulk_create(batch)
            batch = []
 
    if batch:
        await Item.objects.abulk_create(batch)

# Development
python manage.py rungrpc
 
# Production
python manage.py rungrpc --host 0.0.0.0 --port 50051
 
# With debug mode
python manage.py rungrpc --asyncio-debug

GRPCServerConfig(
    asyncio_debug=True,  # Shows async warnings
)

python manage.py rungrpc --asyncio-debug