Ludo API Python — Async HTTP, WebSockets, FastAPI & Docker

Python's async ecosystem has matured dramatically, making it a top-tier choice for high-performance Ludo game integrations. This guide covers the full spectrum: from basic synchronous requests to advanced async HTTP patterns with aiohttp and httpx, real-time WebSocket clients built on asyncio, a production-ready FastAPI server skeleton, pytest-based test suites, and Docker deployment configurations. By the end, you'll have a complete, deployable Python project ready for production traffic.

Environment Setup and Dependencies

A proper Python project starts with a virtual environment and a pyproject.toml or requirements.txt that pins exact versions. This prevents dependency drift across your team and CI pipeline. The following packages cover every aspect of this guide:

# Create and activate virtual environment
python3 -m venv ludo-api-env
source ludo-api-env/bin/activate

# Install all dependencies with pinned versions
# Async HTTP client with connection pooling
pip install aiohttp==3.9.3

# HTTP client with built-in retry and timeout handling
pip install httpx==0.27.0

# Async WebSocket client
pip install websockets==12.0

# FastAPI + uvicorn for the server
pip install fastapi==0.111.0 uvicorn[standard]==0.29.0

# Data validation
pip install pydantic==2.6.3

# Environment variable management
pip install python-dotenv==1.0.1

# Async file I/O for logging
pip install aiofiles==23.2.1

# Testing framework
pip install pytest==8.1.1 pytest-asyncio==0.23.6 pytest-cov==4.1.0
pip install aioresponses==0.7.6  # Mock aiohttp in tests

For a requirements.txt file, pin all versions and regenerate it with pip freeze > requirements.txt after installing. In production, always use a lock file generated by pip-tools or poetry to ensure reproducible builds.

Async HTTP Client with aiohttp

Synchronous requests is fine for low-volume scripts, but a production Ludo platform makes hundreds of concurrent API calls — room queries, player stats, leaderboard fetches, game history lookups. aiohttp provides a fully async HTTP client with connection pooling, automatic JSON serialization, and middleware support. The key advantage over the synchronous library is that while one request is waiting on network I/O, other coroutines can execute. In a game platform, this means your data fetches never block the game loop.

import asyncio
import os
import json
from typing import Optional, Any
import aiohttp
from dotenv import load_dotenv

load_dotenv()

class AsyncLudoClient:
    """Fully async HTTP client for the Ludo API with session management and retry logic."""

    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: str = "https://api.ludokingapi.site/v1",
        max_retries: int = 3,
        timeout: float = 15.0
    ):
        self.api_key = api_key or os.getenv("LUDO_API_KEY")
        self.base_url = base_url.rstrip("/")
        self.max_retries = max_retries
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        self._session: Optional[aiohttp.ClientSession] = None

    async def _get_session(self) -> aiohttp.ClientSession:
        """Lazily create and cache the aiohttp session for connection reuse."""
        if self._session is None or self._session.closed:
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
                "User-Agent": "LudoAsyncClient/1.0"
            }
            connector = aiohttp.TCPConnector(
                limit=100,            # Max concurrent connections
                limit_per_host=30,   # Max per host
                enable_cleanup_closed=True
            )
            self._session = aiohttp.ClientSession(
                headers=headers,
                timeout=self.timeout,
                connector=connector
            )
        return self._session

    async def _request(
        self,
        method: str,
        endpoint: str,
        params: Optional[dict] = None,
        json_data: Optional[dict] = None
    ) -> dict[str, Any]:
        """Internal request method with retry logic for transient failures."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        session = await self._get_session()

        for attempt in range(self.max_retries):
            try:
                async with session.request(
                    method, url, params=params, json=json_data
                ) as response:
                    response.raise_for_status()
                    return await response.json()
            except aiohttp.ClientResponseError as e:
                if e.status == 429:  # Rate limited — backoff and retry
                    wait = 2 ** attempt
                    print(f"Rate limited. Sleeping {wait}s before retry (attempt {attempt + 1})")
                    await asyncio.sleep(wait)
                elif e.status >= 500:  # Server error — retry
                    wait = 1 ** attempt
                    await asyncio.sleep(wait)
                else:
                    raise LudoAPIError(f"HTTP {e.status}: {e.message}")
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                if attempt == self.max_retries - 1:
                    raise LudoAPIError(f"Request failed after {self.max_retries} attempts: {e}")
                await asyncio.sleep(0.5 * (attempt + 1))

    async def get(self, endpoint: str, params: Optional[dict] = None) -> dict:
        return await self._request("GET", endpoint, params=params)

    async def post(self, endpoint: str, json_data: Optional[dict] = None) -> dict:
        return await self._request("POST", endpoint, json_data=json_data)

    async def get_rooms(self, status: Optional[str] = None) -> list:
        """Fetch available game rooms, optionally filtered by status."""
        params = {"status": status} if status else None
        result = await self.get("/rooms", params=params)
        return result.get("data", [])

    async def create_room(self, max_players: int = 4, game_mode: str = "classic") -> dict:
        return await self.post("/rooms", json_data={
            "maxPlayers": max_players,
            "gameMode": game_mode
        })

    async def get_game_state(self, game_id: str) -> dict:
        return await self.get(f"/games/{game_id}/state")

    async def close(self):
        """Explicitly close the session when done."""
        if self._session and not self._session.closed:
            await self._session.close()

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()


class LudoAPIError(Exception):
    """Custom exception for Ludo API errors."""
    pass


# Usage with async context manager — ensures clean session cleanup
async def main():
    async with AsyncLudoClient() as client:
        rooms = await client.get_rooms(status="waiting")
        for room in rooms:
            print(f"Room {room['code']}: {room['playerCount']}/{room['maxPlayers']} players")

        room_data = await client.create_room(max_players=4)
        print(f"Created room: {room_data['data']['code']}")

asyncio.run(main())

httpx Client with Retry and Timeout Configuration

httpx is the modern HTTP client for Python that bridges synchronous and asynchronous codebases. It offers first-class async support, built-in retry policies via httpx-retry, and a clean API that feels familiar to anyone coming from the requests library. For teams already using httpx in their stack, this is the recommended client. The following implementation adds intelligent retry logic, circuit breaker patterns, and structured logging — the three pillars of resilient HTTP integrations.

import httpx
import time
import logging
from typing import Optional

from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, before_sleep_log
)

logger = logging.getLogger(__name__)

# Configure retryable exceptions
RETRYABLE_EXCEPTIONS = (
    httpx.TimeoutException,
    httpx.NetworkError,
    httpx.RemoteProtocolError,
)

# Retry decorator: exponential backoff starting at 1s, up to 3 attempts
def retry_on_transient_error():
    return retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True
    )

class ResilientLudoClient:
    """Production-ready httpx client with retry logic, timeouts, and structured logging."""

    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.ludokingapi.site/v1",
        log_requests: bool = True
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.log_requests = log_requests

        # httpx AsyncClient with global timeout configuration
        self._client = httpx.AsyncClient(
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            },
            timeout=httpx.Timeout(
                connect=5.0,   # Connection establishment timeout
                read=10.0,    # Response read timeout
                write=5.0,    # Request write timeout
                pool=15.0     # Connection pool timeout
            ),
            limits=httpx.Limits(
                max_keepalive_connections=20,
                max_connections=100,
                keepalive_expiry=30.0
            )
        )

    @retry_on_transient_error()
    async def _request(self, method: str, endpoint: str, **kwargs) -> dict:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        start = time.monotonic()

        try:
            response = await self._client.request(method, url, **kwargs)
            duration = time.monotonic() - start

            if self.log_requests:
                logger.info(
                    f"{method} {endpoint} → {response.status_code} ({duration:.3f}s)"
                )

            response.raise_for_status()
            return response.json()

        except httpx.HTTPStatusError as e:
            duration = time.monotonic() - start
            logger.error(
                f"{method} {endpoint} → {e.response.status_code} ({duration:.3f}s): {e.response.text}"
            )
            raise

    async def get(self, endpoint: str, params: Optional[dict] = None) -> dict:
        return await self._request("GET", endpoint, params=params)

    async def post(self, endpoint: str, json_data: Optional[dict] = None) -> dict:
        return await self._request("POST", endpoint, json=json_data)

    async def close(self):
        await self._client.aclose()

WebSocket Client with asyncio

Real-time gameplay demands WebSocket connections for sub-second event delivery. Python's websockets library pairs natively with asyncio, letting you write event-driven game loops that are readable and maintainable. The pattern below uses a robust reconnection strategy with exponential backoff — critical for mobile clients on unreliable connections — and a typed event dispatch system that routes incoming messages to registered handlers.

import asyncio
import json
import logging
import time
from typing import Callable, Optional
import websockets
from websockets.client import WebSocketClientProtocol

from dotenv import load_dotenv
import os

load_dotenv()
logger = logging.getLogger(__name__)

class LudoWebSocketClient:
    """Async WebSocket client with automatic reconnection and typed event handlers."""

    def __init__(
        self,
        api_key: str,
        ws_url: str = "wss://api.ludokingapi.site/ws",
        max_reconnect_attempts: int = 5
    ):
        self.api_key = api_key
        self.ws_url = ws_url
        self.max_reconnect_attempts = max_reconnect_attempts
        self._socket: Optional[WebSocketClientProtocol] = None
        self._handlers: dict[str, Callable] = {}
        self._running = False
        self.room_code: Optional[str] = None
        self.player_token: Optional[str] = None

    def on(self, event_type: str, handler: Callable[[dict], None]) -> None:
        """Register an event handler for a specific event type."""
        self._handlers[event_type] = handler

    async def _connect(self) -> WebSocketClientProtocol:
        """Establish WebSocket connection with authentication header."""
        headers = {"Authorization": f"Bearer {self.api_key}"}
        socket = await websockets.connect(
            self.ws_url,
            extra_headers=headers,
            ping_interval=20,   # Keep connection alive
            ping_timeout=10,
            close_timeout=5
        )
        logger.info("WebSocket connected")
        return socket

    async def _join_room(self):
        """Send room-join message after connection."""
        if not self.room_code or not self.player_token:
            raise ValueError("room_code and player_token must be set before connecting")

        message = {
            "type": "room:join",
            "roomCode": self.room_code,
            "playerToken": self.player_token
        }
        await self._socket.send(json.dumps(message))
        logger.info(f"Joined room {self.room_code}")

    async def send(self, event_type: str, data: dict) -> None:
        """Send a typed message to the server."""
        if self._socket and self._socket.open:
            message = {"type": event_type, **data}
            await self._socket.send(json.dumps(message))
        else:
            logger.warning("Cannot send — socket is not open")

    async def start(self, room_code: str, player_token: str) -> None:
        """Start the WebSocket client with reconnection loop."""
        self.room_code = room_code
        self.player_token = player_token
        self._running = True

        for attempt in range(self.max_reconnect_attempts):
            try:
                self._socket = await self._connect()
                await self._join_room()

                async for raw_message in self._socket:
                    event = json.loads(raw_message)
                    event_type = event.get("type", "unknown")
                    payload = event.get("data", {})

                    logger.debug(f"Event: {event_type}")

                    if event_type in self._handlers:
                        self._handlers[event_type](payload)

                    if event_type == "room:closed":
                        logger.info("Room closed — exiting listener")
                        self._running = False
                        break

            except websockets.ConnectionClosed as e:
                logger.warning(
                    f"Connection closed (code: {e.code}). Reconnecting in {2 ** attempt}s..."
                )
                await asyncio.sleep(2 ** attempt)

            except Exception as e:
                logger.error(f"WebSocket error: {e}")
                await asyncio.sleep(5)

        if self._running:
            logger.error("Max reconnection attempts reached")

    async def stop(self) -> None:
        self._running = False
        if self._socket and self._socket.open:
            await self._socket.close(code=1000, reason="Client disconnect")


# Usage: register handlers and start the client
async def main():
    client = LudoWebSocketClient(os.getenv("LUDO_API_KEY"))

    # Register typed event handlers
    client.on("game:dice-rolled", lambda d: print(f"Dice: {d.get('value')}"))
    client.on("game:piece-moved", lambda d: print(f"Piece moved: {d.get('pieceId')} → {d.get('position')}"))
    client.on("game:turn-change", lambda d: print(f"Turn: {d.get('currentPlayerId')}"))
    client.on("game:ended", lambda d: print(f"Winner: {d.get('winnerId')}"))

    try:
        await client.start(room_code="ABCD12", player_token="player_token_here")
    finally:
        await client.stop()

asyncio.run(main())

FastAPI Server Setup

When your Python backend needs to expose its own REST endpoints — for example, to receive webhooks from the Ludo API, serve game state to your frontend, or act as a proxy layer — FastAPI is the ideal framework. It delivers automatic request validation with Pydantic, OpenAPI documentation at /docs, native WebSocket support, and async request handlers out of the box. The server below implements a clean layered architecture: a service layer that calls the Ludo API, Pydantic models for request/response validation, and a router that maps HTTP verbs to business logic.

# app/main.py — FastAPI application entry point
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Depends, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
import httpx

from dotenv import load_dotenv
load_dotenv()

# ─── Pydantic Models ─────────────────────────────────────────────────────────

class RoomCreateRequest(BaseModel):
    max_players: int = Field(default=4, ge=2, le=4)
    game_mode: str = Field(default="classic", pattern="^(classic|quick|ranked)$")

class RoomResponse(BaseModel):
    code: str
    status: str
    player_count: int
    max_players: int
    game_id: Optional[str] = None

class MoveRequest(BaseModel):
    game_id: str
    piece_id: str
    target_position: int = Field(ge=0, le=57)

class APIResponse(BaseModel):
    success: bool
    data: Optional[dict] = None
    error: Optional[str] = None

# ─── Application Lifespan ─────────────────────────────────────────────────────

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: create a shared httpx client with connection pooling
    app.state.http_client = httpx.AsyncClient(
        timeout=httpx.Timeout(10.0),
        headers={"Authorization": f"Bearer {os.getenv('LUDO_API_KEY')}"}
    )
    yield
    # Shutdown: close the client
    await app.state.http_client.aclose()

# ─── FastAPI App ──────────────────────────────────────────────────────────────

app = FastAPI(
    title="Ludo Game Proxy API",
    description="Bridges your frontend to the LudoKingAPI with rate limiting and caching.",
    version="1.0.0",
    lifespan=lifespan
)

# CORS middleware — allow your frontend domain in production
app.add_middleware(
    CORSMiddleware,
    allow_origins=os.getenv("CORS_ORIGINS", "http://localhost:3000").split(","),
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

LUDO_API_BASE = "https://api.ludokingapi.site/v1"

# ─── REST Endpoints ───────────────────────────────────────────────────────────

@app.post("/rooms", response_model=APIResponse)
async def create_room(request: RoomCreateRequest, client: httpx.AsyncClient = Depends(lambda: None)):
    # Note: in FastAPI, inject app.state.http_client via dependency
    response = await client.post(
        f"{LUDO_API_BASE}/rooms",
        json={"maxPlayers": request.max_players, "gameMode": request.game_mode}
    )
    response.raise_for_status()
    data = response.json()
    return APIResponse(success=True, data=data.get("data"))

@app.get("/rooms/{room_code}", response_model=APIResponse)
async def get_room(room_code: str):
    response = await app.state.http_client.get(f"{LUDO_API_BASE}/rooms/{room_code}")
    response.raise_for_status()
    data = response.json()
    return APIResponse(success=True, data=data.get("data"))

@app.post("/games/move", response_model=APIResponse)
async def submit_move(request: MoveRequest):
    response = await app.state.http_client.post(
        f"{LUDO_API_BASE}/games/{request.game_id}/move",
        json={"pieceId": request.piece_id, "targetPosition": request.target_position}
    )
    response.raise_for_status()
    data = response.json()
    return APIResponse(success=True, data=data.get("data"))

# ─── WebSocket Endpoint ───────────────────────────────────────────────────────

@app.websocket("/ws/game/{room_code}")
async def ws_game(websocket: WebSocket, room_code: str):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            # Forward to upstream WebSocket and relay response back
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print(f"Client disconnected from room {room_code}")


# Run with: uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

pytest Testing Suite

Testing is non-negotiable for production integrations. The following test suite uses pytest with pytest-asyncio for async test functions, aioresponses to mock HTTP calls, and pytest fixtures to provide clean, isolated test data. Fixtures are the backbone of a maintainable test suite — they create consistent test objects and clean them up automatically, regardless of whether a test passes or fails.

# tests/test_ludo_client.py
import pytest
import pytest_asyncio
import asyncio
from aioresponses import aioresponses
from ludo_client import AsyncLudoClient  # Your client module

# ─── Fixtures ─────────────────────────────────────────────────────────────────

@pytest.fixture(scope="session")
def api_key():
    """Provide a test API key."""
    return "test_api_key_12345"

@pytest.fixture(scope="session")
def base_url():
    """Provide the base API URL for tests."""
    return "https://api.ludokingapi.site/v1"

@pytest_asyncio.fixture
async def client(api_key, base_url):
    """Provide an async client instance with automatic cleanup."""
    async_client = AsyncLudoClient(api_key=api_key, base_url=base_url)
    yield async_client
    await async_client.close()

@pytest.fixture
def mock_room_data():
    """Provide realistic mock room data."""
    return {
        "success": True,
        "data": [
            {"code": "ROOM1", "status": "waiting", "playerCount": 2, "maxPlayers": 4},
            {"code": "ROOM2", "status": "playing", "playerCount": 4, "maxPlayers": 4}
        ]
    }

# ─── Tests ────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_get_rooms_returns_list(client, base_url, mock_room_data):
    """GET /rooms should return a list of rooms."""
    with aioresponses() as mocked:
        mocked.get(f"{base_url}/rooms", payload=mock_room_data)
        rooms = await client.get_rooms()
        assert len(rooms) == 2
        assert rooms[0]["code"] == "ROOM1"

@pytest.mark.asyncio
async def test_create_room_returns_code(client, base_url):
    """POST /rooms should return room code on success."""
    create_response = {
        "success": True,
        "data": {"code": "TESTROOM", "status": "waiting"}
    }
    with aioresponses() as mocked:
        mocked.post(f"{base_url}/rooms", payload=create_response)
        result = await client.create_room(max_players=4)
        assert result["data"]["code"] == "TESTROOM"

@pytest.mark.asyncio
async def test_get_rooms_filters_by_status(client, base_url, mock_room_data):
    """GET /rooms?status=waiting should return only waiting rooms."""
    with aioresponses() as mocked:
        mocked.get(
            f"{base_url}/rooms",
            payload=mock_room_data,
            params={"status": "waiting"}
        )
        rooms = await client.get_rooms(status="waiting")
        assert all(r["status"] == "waiting" for r in rooms)

@pytest.mark.asyncio
async def test_api_error_raises_exception(client, base_url):
    """HTTP 401 should raise LudoAPIError."""
    with aioresponses() as mocked:
        mocked.get(f"{base_url}/rooms", status=401)
        with pytest.raises(LudoAPIError):
            await client.get_rooms()

@pytest.mark.asyncio
async def test_client_context_manager(api_key, base_url):
    """Using 'async with' should cleanly close the session."""
    async with AsyncLudoClient(api_key=api_key, base_url=base_url) as client:
        assert client._session is not None
    assert client._session.closed

Docker Deployment

Containerizing your Python Ludo integration ensures consistent behavior from development through production. A multi-stage Dockerfile keeps the final image lean by separating the build environment from the runtime environment. The uvicorn server runs as a non-root user for security, health checks allow orchestrators like Kubernetes or Docker Compose to verify liveness, and environment variables inject the API key at runtime rather than baking it into the image.

# Dockerfile — Multi-stage build for production

# Stage 1: Build
FROM python:3.12-slim AS builder

WORKDIR /app

# Install build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies into a virtual environment
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

COPY requirements.txt .
RUN pip install --no-cache-dir --upgrade pip \
    && pip install --no-cache-dir -r requirements.txt

# Stage 2: Production runtime
FROM python:3.12-slim AS runtime

WORKDIR /app

# Create non-root user for security
RUN groupadd --gid 1000 ludoapi \
    && useradd --uid 1000 --gid ludoapi --shell /bin/bash ludoapi

COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

COPY --chown=ludoapi:ludoapi . .

# Switch to non-root user
USER ludoapi

# Expose port
EXPOSE 8000

# Health check — uvicorn's /health endpoint
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

# Run uvicorn with gunicorn workers for production concurrency
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml — local development and production stack
version: "3.9"

services:
  ludo-api:
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - LUDO_API_KEY=${LUDO_API_KEY}
      - CORS_ORIGINS=http://localhost:3000,https://yourdomain.com
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
      interval: 30s
      timeout: 10s
      retries: 3

  ludo-worker:
    build: .
    command: python -m app.game_worker
    environment:
      - LUDO_API_KEY=${LUDO_API_KEY}
    depends_on:
      - ludo-api
    restart: unless-stopped

Key Architecture Patterns Covered

Connection pooling via aiohttp TCPConnector and httpx AsyncClient limits — prevents exhausting file descriptors under load
Exponential backoff for retry on 429, 500, and connection errors — handles API rate limits and transient outages gracefully
Typed event dispatch in WebSocket client — clean separation between event routing and handler logic
Pydantic validation on FastAPI endpoints — rejects malformed requests before they reach your business logic
Session reuse — the aiohttp ClientSession maintains persistent connections, dramatically reducing latency for repeated API calls
Multi-stage Docker builds — final image contains only runtime dependencies, keeping attack surface minimal

Frequently Asked Questions

What's the difference between aiohttp and httpx for Ludo API integration?

How do I handle thousands of concurrent game rooms in Python?

Should I use pytest fixtures or manual setup in test classes?

How do I secure the API key in a Docker container?

Can I run the WebSocket client and FastAPI server in the same process?

Why is my pytest async test failing with "event loop closed"?

How do I monitor the health of my FastAPI Ludo server in production?

Need Help Building Your Python Ludo Integration?

From async patterns to Docker deployment, our team can help you build a production-ready Python backend for your Ludo platform. Get in touch on WhatsApp.

Chat on WhatsApp