Skip to main content
mySpellChecker is tested and integrated with FastAPI. The key pattern: create a SpellChecker instance once at startup, then reuse it across requests using check_async() for non-blocking operation.

​
FastAPI

​
Basic Endpoint

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from myspellchecker import SpellChecker
from myspellchecker.core.constants import ValidationLevel

app = FastAPI(title="Myanmar Spell Checker API")
checker = SpellChecker()  # Create once, reuse


class CheckRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=10000)
    level: str = Field(default="syllable")


class ErrorInfo(BaseModel):
    text: str
    position: int
    error_type: str
    suggestions: list[str]
    confidence: float


class CheckResponse(BaseModel):
    text: str
    has_errors: bool
    corrected_text: str
    errors: list[ErrorInfo]


@app.post("/check", response_model=CheckResponse)
async def check_spelling(request: CheckRequest):
    level = ValidationLevel(request.level)
    result = await checker.check_async(request.text, level=level)

    return CheckResponse(
        text=result.text,
        has_errors=result.has_errors,
        corrected_text=result.corrected_text,
        errors=[
            ErrorInfo(
                text=e.text,
                position=e.position,
                error_type=e.error_type,
                suggestions=e.suggestions[:5],
                confidence=e.confidence,
            )
            for e in result.errors
        ],
    )

​
Testing with curl

# Check single text
curl -X POST http://localhost:8000/check \
  -H "Content-Type: application/json" \
  -d '{"text": "မြန်မာနိုင်ငံသည်", "level": "syllable"}'

# Batch check
curl -X POST http://localhost:8000/check/batch \
  -H "Content-Type: application/json" \
  -d '{"texts": ["မြန်မာ", "နိုင်ငံ"], "level": "syllable"}'

# Health check
curl http://localhost:8000/health

​
Batch Endpoint

class BatchRequest(BaseModel):
    texts: list[str]

@app.post("/check/batch")
async def check_batch(request: BatchRequest):
    results = await checker.check_batch_async(request.texts)
    return {
        "results": [
            {"text": r.text, "has_errors": r.has_errors, "error_count": len(r.errors)}
            for r in results
        ],
        "summary": {
            "total_texts": len(results),
            "texts_with_errors": sum(1 for r in results if r.has_errors),
        },
    }

​
With Lifespan Management

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.checker = SpellChecker()
    yield

app = FastAPI(lifespan=lifespan)

@app.post("/check")
async def check_spelling(request: Request, body: CheckRequest):
    checker = request.app.state.checker
    result = await checker.check_async(body.text)
    return {"has_errors": result.has_errors, "errors": [e.to_dict() for e in result.errors]}

​
CORS Configuration

from fastapi.middleware.cors import CORSMiddleware
import os

allowed_origins_env = os.getenv("ALLOWED_ORIGINS")
if allowed_origins_env:
    origins = [o.strip() for o in allowed_origins_env.split(",")]
    allow_credentials = True
else:
    origins = ["*"]
    allow_credentials = False

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=allow_credentials,
    allow_methods=["*"],
    allow_headers=["*"],
)

​
Production Deployment

# Development with auto-reload
uvicorn app:app --reload --port 8000

# Production with Gunicorn
gunicorn app:app \
  -w 4 \
  -k uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000

​
WebSocket

​
FastAPI WebSocket

from fastapi import FastAPI, WebSocket
import json

app = FastAPI()
checker = SpellChecker()

@app.websocket("/ws/check")
async def websocket_check(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)
            result = await checker.check_async(message["text"])

            await websocket.send_json({
                "has_errors": result.has_errors,
                "errors": [
                    {"text": e.text, "suggestions": e.suggestions[:3]}
                    for e in result.errors
                ],
            })
    except Exception:
        pass

​
Common Patterns

​
Error Handling

from myspellchecker.core.exceptions import MyanmarSpellcheckError

@app.post("/check")
async def check_spelling(request: CheckRequest):
    try:
        result = await checker.check_async(request.text)
        return {"status": "success", "result": result.to_dict()}
    except MyanmarSpellcheckError as e:
        return {"status": "error", "message": str(e)}, 500
    except Exception:
        return {
            "status": "degraded",
            "message": "Spell check unavailable",
            "original_text": request.text,
        }

​
Health Check

@app.get("/health")
async def health_check():
    try:
        result = checker.check("မြန်မာ")
        return {"status": "healthy", "checker": "ok"}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}, 500

​
Redis Caching

import redis
import hashlib
import json

redis_client = redis.Redis()

def cached_check(text: str, ttl: int = 3600) -> dict:
    key = f"spell:{hashlib.md5(text.encode()).hexdigest()}"
    cached = redis_client.get(key)
    if cached:
        return json.loads(cached)

    result = checker.check(text)
    output = {
        "has_errors": result.has_errors,
        "errors": [{"text": e.text} for e in result.errors],
    }
    redis_client.setex(key, ttl, json.dumps(output))
    return output

​
See Also