Tutorials

Ludo Python Tutorial — FastAPI, Pydantic, SQLAlchemy, WebSocket & Docker

Build a production-grade Ludo game backend in Python. This tutorial covers FastAPI server setup with async routes, Pydantic models for game state validation, SQLAlchemy ORM with SQLite for persistence, WebSocket multiplayer, Docker Compose deployment, and integration with the LudoKingAPI for cross-platform validation.

Python 3.9+FastAPIPydanticSQLAlchemyWebSocketDocker

What You'll Build

By the end of this tutorial you will have a fully functional Ludo game backend that creates and manages multiplayer games, validates moves against official Ludo rules, persists game state to SQLite using SQLAlchemy, broadcasts real-time updates via WebSocket, and optionally defers to the LudoKingAPI for anti-cheat validation and cross-platform matchmaking. The complete stack uses FastAPI as the ASGI server, Pydantic v2 for request validation, SQLAlchemy 2.0 with async support, and Docker Compose for reproducible deployments. Every component is production-oriented: typed throughout, error-handled, and designed to scale horizontally.

Step 1 — Project Setup and Dependencies

Create a clean project directory and install the required packages. We use uv or pip to manage dependencies. The stack is deliberately minimal — FastAPI handles HTTP and WebSocket, Pydantic validates all data, SQLAlchemy manages persistence, and aiosqlite provides async SQLite access. Environment variables are loaded via python-dotenv.

Bash
# Create and enter project directory
mkdir ludo-server && cd ludo-server
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install all dependencies
pip install fastapi uvicorn[standard] pydantic pydantic-settings \
  sqlalchemy aiosqlite python-dotenv websockets httpx

# Verify versions
python -c "import fastapi, pydantic, sqlalchemy; print('FastAPI', fastapi.__version__, '| Pydantic', pydantic.__version__, '| SQLAlchemy', sqlalchemy.__version__)"

Step 2 — requirements.txt

Pin all dependencies with version constraints for reproducible builds. In production, always use a requirements.txt or pyproject.toml to lock your dependency graph.

Text
# requirements.txt
fastapi==0.115.0
uvicorn[standard]==0.30.6
pydantic==2.9.2
pydantic-settings==2.5.2
sqlalchemy==2.0.35
aiosqlite==0.20.0
python-dotenv==1.0.1
websockets==12.0
httpx==0.27.2

Step 3 — Pydantic Models for Game State

Pydantic is the backbone of FastAPI's validation layer. We define GameState, PlayerInGame, Token, MoveRequest, and GameCreate models. Using BaseModel with field validators ensures every piece of data that crosses the API boundary is validated before touching business logic. This eliminates entire classes of bugs at the boundary.

Python
# models.py — Pydantic models for all game state and API contracts
from __future__ import annotations
from enum import IntEnum
from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel, Field, field_validator, ConfigDict


class PlayerColor(IntEnum):
    RED = 0
    GREEN = 1
    BLUE = 2
    YELLOW = 3


class TokenState(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    player: PlayerColor
    square: int = Field(ge=-1, le=57, description="Board square index, -1 = in home base")
    finished: bool = False
    token_index: int = Field(ge=0, le=3, description="Token 0-3 within player's set")


class PlayerInGame(BaseModel):
    player_id: str = Field(min_length=1, max_length=64)
    color: PlayerColor
    display_name: str = Field(min_length=1, max_length=32)
    connected: bool = True
    finished_tokens: int = Field(default=0, ge=0, le=4)


class GameState(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    game_id: str
    players: List[PlayerInGame] = Field(max_length=4)
    current_player_index: int = Field(ge=0, le=3)
    dice_value: Optional[int] = Field(default=None, ge=1, le=6)
    tokens: List[TokenState] = Field(default_factory=list)
    phase: str = Field(default="waiting", description="waiting | rolling | moving | finished")
    winner: Optional[PlayerColor] = None
    move_count: int = Field(default=0, ge=0)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

    @field_validator("tokens")
    @classmethod
    def must_have_16_tokens(cls, v: List[TokenState]) -> List[TokenState]:
        if len(v) != 16:
            raise ValueError(f"Exactly 16 tokens required, got {len(v)}")
        return v


class GameCreate(BaseModel):
    max_players: int = Field(default=4, ge=2, le=4)
    player_name: str = Field(min_length=1, max_length=32)
    use_api_validation: bool = Field(default=False, description="Validate moves via LudoKingAPI")
    room_name: Optional[str] = Field(default=None, max_length=64)


class MoveRequest(BaseModel):
    player_id: str = Field(min_length=1, max_length=64)
    token_index: int = Field(ge=0, le=3)
    target_square: Optional[int] = Field(default=None, ge=0, le=57)
    dice_value: int = Field(ge=1, le=6)


class RollResponse(BaseModel):
    game_id: str
    player_id: str
    dice_value: int
    extra_turn: bool = Field(description="True if player rolled a 6 and gets another roll")
    legal_moves: List[int] = Field(default_factory=list, description="Indices of movable tokens")


class MoveResponse(BaseModel):
    game_id: str
    player_id: str
    token_index: int
    from_square: int
    to_square: int
    captured: bool = False
    captured_token_index: Optional[int] = None
    finished: bool = False
    game_over: bool = False
    winner: Optional[PlayerColor] = None
    next_player_index: int

These models enforce the game contract at the type level. The TokenState validator ensures squares are always in the range [-1, 57], where -1 represents tokens safely in the home base, 0-51 are the main track, and 52-56 are the home column. The must_have_16_tokens validator guarantees the board always has exactly 16 tokens (four per player) in the initialised state.

Step 4 — SQLAlchemy ORM Setup with Async Support

SQLAlchemy 2.0's async engine lets us persist game state to SQLite without blocking the event loop. We define two tables: games stores the serialised game state as JSON, and game_events stores an append-only log of every action (roll, move, capture) for replay, analytics, and anti-cheat auditing. Using AsyncSession with asyncpg or aiosqlite as the async driver keeps the FastAPI coroutines non-blocking.

Python
# database.py — SQLAlchemy 2.0 async ORM setup
from __future__ import annotations
from contextlib import asynccontextmanager
from datetime import datetime
from typing import AsyncGenerator

from sqlalchemy import String, JSON, Integer, DateTime, ForeignKey, Text, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_url: str = "sqlite+aiosqlite:///./ludo.db"
    debug: bool = False


settings = Settings()

class Base(DeclarativeBase):
    pass


class Game(Base):
    __tablename__ = "games"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    game_id: Mapped[str] = mapped_column(String(32), unique=True, index=True)
    state_json: Mapped[dict] = mapped_column(JSON, nullable=False)
    phase: Mapped[str] = mapped_column(String(32), default="waiting")
    max_players: Mapped[int] = mapped_column(Integer, default=4)
    winner: Mapped[int | None] = mapped_column(Integer, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now())

    events: Mapped[list["GameEvent"]] = relationship(back_populates="game", cascade="all, delete-orphan")


class GameEvent(Base):
    __tablename__ = "game_events"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    game_id: Mapped[int] = mapped_column(Integer, ForeignKey("games.id"), index=True)
    event_type: Mapped[str] = mapped_column(String(32))  # roll | move | capture | finish
    player_id: Mapped[str] = mapped_column(String(64), index=True)
    payload: Mapped[dict] = mapped_column(JSON, nullable=False)
    sequence: Mapped[int] = mapped_column(Integer)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    game: Mapped["Game"] = relationship(back_populates="events")


# Engine and session factory — shared across the app
engine = create_async_engine(settings.database_url, echo=settings.debug)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


async def init_db() -> None:
    """Create all tables. Run once at startup."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


async def get_session() -> AsyncGenerator[AsyncSession, None]:
    """Dependency for FastAPI endpoints."""
    async with AsyncSessionLocal() as session:
        yield session

Step 5 — Game Logic Layer (Domain)

The LudoGame class is the pure domain layer — no FastAPI, no SQLAlchemy, no I/O. It implements the complete Ludo rule engine: dice rolling, token entry from home (requires a 6), track movement, capture detection, safe-square protection, home-column restrictions, and win detection. Keeping this pure makes the logic trivially testable and portable to other contexts.

Python
# game_logic.py — Pure domain layer: complete Ludo rule engine
from __future__ import annotations
from dataclasses import dataclass, field
from typing import List, Optional
from models import TokenState, PlayerColor, PlayerInGame

# Board geometry: squares 0-51 are the shared track, 52-56 are per-player home columns.
SAFE_SQUARES = frozenset({0, 8, 13, 21, 26, 34, 39, 47})
# Star squares grant an extra roll.
STAR_SQUARES = frozenset({0, 8, 13, 21, 26, 34, 39, 47})
# Entry square into home column for each color (0-indexed color).
HOME_ENTRY = {PlayerColor.RED: 51, PlayerColor.GREEN: 13, PlayerColor.BLUE: 26, PlayerColor.YELLOW: 39}
# First square in home column for each color.
HOME_START = {PlayerColor.RED: 52, PlayerColor.GREEN: 53, PlayerColor.BLUE: 54, PlayerColor.YELLOW: 55}
FINISH_SQUARE = 56


@dataclass
class LudoGame:
    game_id: str
    players: List[PlayerInGame] = field(default_factory=list)
    current_player_index: int = 0
    dice_value: int = 0
    phase: str = "waiting"
    tokens: List[TokenState] = field(default_factory=list)
    winner: Optional[PlayerColor] = None
    move_count: int = 0

    def initialize_tokens(self) -> None:
        self.tokens = [
            TokenState(player=PlayerColor(c // 4), square=-1, finished=False, token_index=c % 4)
            for c in range(16)
        ]

    def roll(self) -> tuple[int, bool]:
        import random
        self.dice_value = random.randint(1, 6)
        self.phase = "moving"
        legal = self.legal_token_indices()
        return self.dice_value, len(legal) > 0 and self.dice_value == 6

    def legal_token_indices(self) -> List[int]:
        player = self.players[self.current_player_index].color
        return [
            i for i, t in enumerate(self.tokens)
            if t.player == player
            and not t.finished
            and (t.square >= 0 or self.dice_value == 6)
        ]

    def apply_move(self, token_index: int) -> dict:
        token = self.tokens[token_index]
        from_square = token.square

        if token.square == -1:
            token.square = HOME_ENTRY[token.player]
        else:
            target = token.square + self.dice_value
            if target > FINISH_SQUARE:
                return {"valid": False, "reason": "overshoot"}
            if target > HOME_ENTRY[token.player] + 5:
                return {"valid": False, "reason": "cannot_enter_home_column_yet"}
            token.square = target
            if token.square == FINISH_SQUARE:
                token.finished = True
                self.players[self.current_player_index].finished_tokens += 1

        result = {
            "valid": True, "from_square": from_square, "to_square": token.square,
            "captured": False, "captured_token_index": None, "finished": token.finished
        }

        if token.square not in SAFE_SQUARES and token.square >= 0:
            for i, other in enumerate(self.tokens):
                if other.player != token.player and other.square == token.square and not other.finished:
                    other.square = -1
                    result["captured"] = True
                    result["captured_token_index"] = i
                    break

        self.move_count += 1
        self._check_winner()
        self._advance_turn(extra_turn=self.dice_value == 6 and not token.finished)
        return result

    def _advance_turn(self, extra_turn: bool = False) -> None:
        if not extra_turn and not self.winner:
            self.current_player_index = (self.current_player_index + 1) % len(self.players)
        self.phase = "rolling"
        self.dice_value = 0

    def _check_winner(self) -> None:
        current = self.players[self.current_player_index]
        if current.finished_tokens == 4:
            self.winner = current.color
            self.phase = "finished"

    def to_dict(self) -> dict:
        return {
            "game_id": self.game_id,
            "players": [p.model_dump() for p in self.players],
            "current_player_index": self.current_player_index,
            "dice_value": self.dice_value,
            "phase": self.phase,
            "tokens": [t.model_dump() for t in self.tokens],
            "winner": self.winner,
            "move_count": self.move_count,
        }

Step 6 — Complete FastAPI Application

The FastAPI app.py wires everything together. It provides REST endpoints for game lifecycle (create, join, state, move) and a WebSocket endpoint for real-time updates. The WebSocket handler is the core multiplayer primitive — each connected player receives state updates after every action, and the server is the authoritative source of truth. LudoKingAPI calls can be added in validate_move_via_api for production anti-cheat validation.

Python
# app.py — Complete FastAPI application with REST + WebSocket multiplayer
from __future__ import annotations
import uuid, json
from datetime import datetime
from typing import Dict

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.ext.asyncio import AsyncSession
import httpx

from database import init_db, AsyncSessionLocal, Game, GameEvent
from models import (
    GameCreate, MoveRequest, RollResponse, MoveResponse,
    GameState, PlayerColor, PlayerInGame, TokenState
)
from game_logic import LudoGame

# ─── App Initialization ─────────────────────────────────────────────────────
app = FastAPI(title="LudoKingAPI Server", version="1.0.0", docs_url="/docs")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# In-memory game registry. In production, replace with Redis for horizontal scaling.
active_games: Dict[str, LudoGame] = {}
ws_connections: Dict[str, list[WebSocket]] = {}


@app.on_event("startup")
async def on_startup():
    await init_db()


# ─── REST Endpoints ───────────────────────────────────────────────────────────
@app.post("/games", response_model=dict, status_code=status.HTTP_201_CREATED)
async def create_game(body: GameCreate, db: AsyncSession = Depends(get_session)) -> dict:
    game_id = f"ludo_{uuid.uuid4().hex[:12]}"
    player_id = f"p_{uuid.uuid4().hex[:8]}"
    player = PlayerInGame(player_id=player_id, color=PlayerColor.RED, display_name=body.player_name)
    game = LudoGame(game_id=game_id, players=[player], phase="waiting")
    game.initialize_tokens()
    active_games[game_id] = game

    db_game = Game(game_id=game_id, state_json=game.to_dict(), phase="waiting", max_players=body.max_players)
    db.add(db_game)
    await db.commit()
    await db.refresh(db_game)
    return {"game_id": game_id, "player_id": player_id, "state": game.to_dict()}


@app.post("/games/{game_id}/join")
async def join_game(game_id: str, player_name: str, db: AsyncSession = Depends(get_session)) -> dict:
    if game_id not in active_games:
        raise HTTPException(status_code=404, detail="Game not found")
    game = active_games[game_id]
    if len(game.players) >= game.players[0].color + 4 if game.players else 4:
        raise HTTPException(status_code=400, detail="Game is full")
    if any(p.display_name == player_name for p in game.players):
        raise HTTPException(status_code=409, detail="Name already taken")

    color_idx = len(game.players)
    if color_idx >= 4:
        raise HTTPException(status_code=400, detail="Maximum 4 players")
    player_id = f"p_{uuid.uuid4().hex[:8]}"
    player = PlayerInGame(player_id=player_id, color=PlayerColor(color_idx), display_name=player_name)
    game.players.append(player)
    return {"game_id": game_id, "player_id": player_id, "state": game.to_dict()}


@app.get("/games/{game_id}")
async def get_game(game_id: str) -> dict:
    if game_id not in active_games:
        raise HTTPException(status_code=404, detail="Game not found")
    return active_games[game_id].to_dict()


@app.post("/games/{game_id}/roll")
async def roll_dice(game_id: str, player_id: str, db: AsyncSession = Depends(get_session)) -> RollResponse:
    if game_id not in active_games:
        raise HTTPException(status_code=404, detail="Game not found")
    game = active_games[game_id]
    current_player = game.players[game.current_player_index]
    if current_player.player_id != player_id:
        raise HTTPException(status_code=403, detail="Not your turn")
    if game.phase not in ("waiting", "rolling"):
        raise HTTPException(status_code=400, detail="Must roll dice during rolling phase")

    dice_value, extra_turn = game.roll()
    legal = game.legal_token_indices()
    response = RollResponse(game_id=game_id, player_id=player_id, dice_value=dice_value, extra_turn=extra_turn, legal_moves=legal)
    await _persist_event(db, game_id, "roll", player_id, response.model_dump())
    await _broadcast(game_id, game.to_dict())
    return response


@app.post("/games/{game_id}/move")
async def make_move(game_id: str, body: MoveRequest, db: AsyncSession = Depends(get_session)) -> MoveResponse:
    if game_id not in active_games:
        raise HTTPException(status_code=404, detail="Game not found")
    game = active_games[game_id]
    if game.phase not in ("moving", "rolling"):
        raise HTTPException(status_code=400, detail="Not in a moveable phase")
    if body.token_index not in game.legal_token_indices():
        raise HTTPException(status_code=400, detail="Illegal move for this token")

    from_sq = game.tokens[body.token_index].square
    result = game.apply_move(body.token_index)
    if not result["valid"]:
        raise HTTPException(status_code=400, detail=result["reason"])

    response = MoveResponse(
        game_id=game_id, player_id=body.player_id, token_index=body.token_index,
        from_square=from_sq, to_square=result["to_square"], captured=result["captured"],
        captured_token_index=result.get("captured_token_index"), finished=result["finished"],
        game_over=game.winner is not None, winner=game.winner,
        next_player_index=game.current_player_index
    )
    await _persist_event(db, game_id, "move", body.player_id, response.model_dump())
    await _broadcast(game_id, game.to_dict())
    return response


# ─── WebSocket Endpoint ───────────────────────────────────────────────────────
@app.websocket("/ws/{game_id}/{player_id}")
async def websocket_handler(ws: WebSocket, game_id: str, player_id: str):
    if game_id not in active_games:
        await ws.close(code=4004, reason="Game not found")
        return
    game = active_games[game_id]
    if game_id not in ws_connections:
        ws_connections[game_id] = []
    ws_connections[game_id].append(ws)
    await ws.accept()

    try:
        await ws.send_json({"type": "state", **game.to_dict()})
        while True:
            data = await ws.receive_json()
            msg_type = data.get("type")
            if msg_type == "ping":
                await ws.send_json({"type": "pong"})
            elif msg_type == "roll":
                async with AsyncSessionLocal() as db:
                    try:
                        await roll_dice(game_id, player_id, db)
                    except HTTPException as e:
                        await ws.send_json({"type": "error", "message": e.detail})
    except WebSocketDisconnect:
        if game_id in ws_connections and ws in ws_connections[game_id]:
            ws_connections[game_id].remove(ws)


# ─── Helpers ─────────────────────────────────────────────────────────────────
async def _broadcast(game_id: str, state: dict) -> None:
    for ws in ws_connections.get(game_id, []):
        try:
            await ws.send_json(state)
        except Exception:
            pass

async def _persist_event(db: AsyncSession, game_id: str, event_type: str, player_id: str, payload: dict) -> None:
    db_game = await db.get(Game, active_games[game_id].game_id)
    if db_game:
        db_game.state_json = active_games[game_id].to_dict()
        db_game.updated_at = datetime.utcnow()
    event = GameEvent(game_id=db_game.id if db_game else 0, event_type=event_type, player_id=player_id, payload=payload, sequence=active_games[game_id].move_count)
    db.add(event)
    await db.commit()

Step 7 — Docker Compose Setup

Docker Compose provides a reproducible, single-command deployment. The following configuration defines a FastAPI service and an optional Redis service (for scaling beyond a single instance). The docker-compose.yml mounts the source code as a volume for development hot-reloading and exposes port 8000 for the API and port 6379 for Redis.

YAML
# docker-compose.yml
version: '3.9'

services:
  ludo-api:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: ludo-fastapi
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=sqlite+aiosqlite:///./ludo.db
      - DEBUG=false
    volumes:
      - .:/app
    command: uvicorn app:app --host 0.0.0.0 --port 8000 --reload
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/docs"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 10s

  redis:
    image: redis:7-alpine
    container_name: ludo-redis
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    restart: unless-stopped
    command: redis-server --appendonly yes

volumes:
  redis_data:

Create a Dockerfile in the project root to build the FastAPI image:

Dockerfile
# Dockerfile
FROM python:3.12-slim

WORKDIR /app

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

COPY . .

RUN adduser --disabled-password ludo && chown -R ludo:ludo /app
USER ludo

EXPOSE 8000

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

Step 8 — Run and Test with curl

Start the server with uvicorn in development mode. The --reload flag enables auto-reload on file changes. After starting, visit http://localhost:8000/docs for the Swagger UI where you can test every endpoint interactively. Below are curl equivalents for scripting and automation.

Bash
# Start the FastAPI server
uvicorn app:app --reload --port 8000

# Create a new game
curl -X POST http://localhost:8000/games \
  -H "Content-Type: application/json" \
  -d '{"player_name": "Alice", "max_players": 4}'

# Join the game (second player)
curl -X POST "http://localhost:8000/games/ludo_abc123/join?player_name=Bob"

# Get current game state
curl http://localhost:8000/games/ludo_abc123

# Roll the dice (Alice's turn — player_id from create response)
curl -X POST "http://localhost:8000/games/ludo_abc123/roll?player_id=p_xxxxxxxx" \
  -H "Content-Type: application/json"

# Make a move (after rolling — token_index from legal_moves in roll response)
curl -X POST "http://localhost:8000/games/ludo_abc123/move" \
  -H "Content-Type: application/json" \
  -d '{"player_id": "p_xxxxxxxx", "token_index": 0, "dice_value": 6}'

# WebSocket test with websocat (pip install websocat)
websocat "ws://localhost:8000/ws/ludo_abc123/p_xxxxxxxx"

# Inside the WebSocket session, send JSON to roll:
# {"type": "roll"}
# {"type": "ping"}

Step 9 — Connecting to the LudoKingAPI

For production deployments, replace the local LudoGame rule engine with LudoKingAPI calls to get server-authoritative move validation, cross-platform matchmaking, and tournament management. Add a LUDOKING_API_KEY environment variable and update the make_move endpoint to call the validation endpoint before applying moves locally. The API returns a signed move certificate that can be verified client-side, preventing man-in-the-middle manipulation.

Python
# Example: Validate move via LudoKingAPI before applying locally
import os, httpx

LUDOKING_API_KEY = os.getenv("LUDOKING_API_KEY", "")
LUDOKING_BASE_URL = "https://api.ludokingapi.site"

async def validate_move_via_api(game_id: str, player_id: str, token_index: int, dice: int) -> bool:
    if not LUDOKING_API_KEY:
        return True  # Skip validation in development
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{LUDOKING_BASE_URL}/v1/games/{game_id}/validate",
            headers={"Authorization": f"Bearer {LUDOKING_API_KEY}"},
            json={"player_id": player_id, "token_index": token_index, "dice_value": dice},
            timeout=5.0,
        )
        return response.status_code == 200

The full LudoKingAPI integration guide is available in the Ludo API Python page, and a complete step-by-step walkthrough of authentication, room creation, and event handling is in the Realtime API documentation. For implementing AI opponents that connect to this server, see the Ludo Bot Python guide. The Game Dev Python page covers integrating this backend with frontend clients built in Unity, Godot, or web technologies.

Frequently Asked Questions

FastAPI is built on an async event loop, and blocking I/O (like synchronous SQLAlchemy queries) would freeze the entire server while waiting for the database. By using create_async_engine with aiosqlite as the driver, every database operation yields control back to the event loop during I/O, allowing thousands of concurrent WebSocket connections. SQLAlchemy 2.0's async API is nearly identical to the sync version, so the learning curve is minimal. For production with multiple server instances, swap aiosqlite for asyncpg (PostgreSQL) or aiomysql — only the connection string changes.
The current in-memory ws_connections dict works for a single-process deployment. To scale horizontally, replace it with Redis Pub/Sub. When any server instance broadcasts a move, it publishes the new state to a Redis channel. All server instances subscribe and relay the message to their locally connected WebSocket clients. FastAPI supports this natively with fastapi-applications.add_api_websocket_route combined with aioredis or redis.asyncio. The LudoKingAPI platform handles this Pub/Sub architecture automatically if you deploy on their infrastructure.
Yes. Pydantic v2 introduced model_validator, field_validator, and computed_field decorators that run at the class level. The models in this tutorial use field_validator on the tokens field to enforce exactly 16 tokens, and ConfigDict(from_attributes=True) to allow ORM objects to be converted into Pydantic models without manual mapping. You can also use model_rebuild() on dynamically constructed models and TypeAdapter for validating raw JSON against a schema at runtime.
Add JWT authentication using python-jose and passlib. Create an /auth/login endpoint that returns a signed JWT containing the player_id and game_id claims. Add a get_current_player dependency in FastAPI that extracts and verifies the JWT from the Authorization: Bearer header. Apply it to the roll and move endpoints. For WebSocket connections, pass the token as a query parameter (/ws/{game_id}/{player_id}?token=...) and validate it inside the WebSocket handler before calling ws.accept().
A single uvicorn worker with gunicorn ludo-server:app -w 4 -k uvicorn.workers.UvicornWorker can handle thousands of concurrent WebSocket connections. The SQLite database is suitable for development and small-scale deployments (up to ~100 concurrent games). For higher throughput, migrate to PostgreSQL with asyncpg — the SQLAlchemy models in this tutorial require only a connection string change. Benchmark with locust or wrk against the /games/{id}/roll endpoint to establish baseline throughput before scaling.
Every roll and move is persisted to the game_events table with a monotonically increasing sequence number. To replay a game, load the initial Game.state_json snapshot, then replay each event in sequence order by calling game.apply_move() or game.roll(). This is invaluable for debugging, generating highlight reels, and anti-cheat auditing. The LudoKingAPI uses this event sourcing pattern at scale with Kafka as the event bus, allowing games to be replayed from any point in time.
Absolutely. Add an upstream block pointing to your uvicorn workers, then configure nginx to proxy both HTTP and WebSocket connections. For WebSocket, you must add the Upgrade and Connection headers — nginx does not forward these by default. Use proxy_set_header Upgrade $http_upgrade and proxy_set_header Connection "upgrade". Enable HTTP/2 with http2 on the listener for better multiplexing. SSL termination should happen at nginx or at a load balancer in front of nginx.

Related Links

Ludo API Python Ludo Bot Python Ludo Game Python Ludo API Tutorial Realtime API Docs

Build Your Ludo Python Server Today

Get your free API key and combine it with this Python server for a fully-featured multiplayer Ludo backend with matchmaking and validation.

Talk to Us on WhatsApp Realtime API Docs