fastapi-advanced 2.0.1

High-performance FastAPI integration with msgspec for request/response serialization and automatic OpenAPI schema generation

fastapi-advanced

High-performance FastAPI integration with msgspec for fast serialization and automatic OpenAPI support.

Overview

FastAPI's default Pydantic serialization can be a bottleneck in high-throughput applications. This library integrates msgspec for fast response serialization while maintaining full OpenAPI compatibility through Pydantic for request validation.

Key Benefits

• 2-5x faster response serialization with msgspec
• 10-20x faster object creation compared to Pydantic
• Full OpenAPI support - Pydantic for requests, msgspec for responses
• Type-safe with perfect mypy strict compliance
• Minimal changes from existing Pydantic implementations
• Production-ready with Cython optimizations available

Installation

pip install fastapi-advanced

Pre-compiled wheels are available for macOS, Linux, and Windows (Python 3.10-3.13).

Development Installation

git clone https://github.com/your-org/fastapi-advanced.git
cd fastapi-advanced
make install-dev

# Optional: Enable Cython optimizations
make compile-cython

Quick Start

from typing import TYPE_CHECKING

import msgspec
from fastapi import FastAPI

from fastapi_advanced import (
    PaginatedResponseSchema,
    ResponseModelSchema,
    as_body,
    msgspec_to_pydantic,
    paginated_response,
    response,
    setup_msgspec,
)

app = FastAPI(title="My API", version="1.0.0")
setup_msgspec(app)

# Define models with msgspec for runtime performance
class User(msgspec.Struct):
    id: int
    username: str
    email: str
    is_active: bool = True

# Convert to Pydantic for type annotations (OpenAPI generation)
if TYPE_CHECKING:
    UserSchema = User
else:
    UserSchema = msgspec_to_pydantic(User)

# Create endpoints with proper typing
@app.get("/users/{user_id}")
async def get_user(user_id: int) -> ResponseModelSchema[UserSchema]:
    """Get user by ID with type-safe responses."""
    user = User(id=user_id, username="alice", email="[e-mail]")
    return response(data=user, message="User retrieved successfully")

@app.get("/users")
async def list_users(
    page: int = 1,
    page_size: int = 10
) -> PaginatedResponseSchema[UserSchema]:
    """List users with automatic pagination metadata."""
    users = [User(id=i, username=f"user{i}", email=f"user{i}@example.com") for i in range(50)]

    start = (page - 1) * page_size
    end = start + page_size
    page_items = users[start:end]

    return paginated_response(
        items=page_items,
        total_results=len(users),
        page=page,
        page_size=page_size,
    )

Run the application:

uvicorn example:app --reload

Usage Patterns

Type-Safe Schema Conversion

Use the TYPE_CHECKING pattern to eliminate type warnings while maintaining runtime performance:

from typing import TYPE_CHECKING

import msgspec
from fastapi_advanced import msgspec_to_pydantic

class User(msgspec.Struct):
    id: int
    name: str
    email: str

if TYPE_CHECKING:
    # Type checker sees the msgspec Struct type
    UserSchema = User
else:
    # Runtime uses Pydantic for OpenAPI
    UserSchema = msgspec_to_pydantic(User)

# No type: ignore needed in function signatures
@app.get("/users")
async def get_users() -> ResponseModelSchema[UserSchema]:
    users = User(id=1, name="Alice", email="[e-mail]")
    return response(data=users)

Request Body Validation

Use as_body() for request body schemas with OpenAPI documentation:

class CreateUserRequest(msgspec.Struct):
    username: str
    email: str
    full_name: str | None = None

CreateUserRequestBody = as_body(CreateUserRequest)

@app.post("/users", status_code=201)
async def create_user(
    data: CreateUserRequestBody,
) -> ResponseModelSchema[UserSchema]:
    user = User(id=1, username=data.username, email=data.email)
    return response(data=user, status_code=201)

Standard Responses

All responses follow a consistent structure:

# Success response
return response(
    data=user,
    message="Operation successful",
    status_code=200,
)

# Error response
return response(
    data=None,
    message="Resource not found",
    status="error",
    status_code=404,
)

Response structure:

{
  "status": "ok",
  "data": {...},
  "message": "Operation successful"
}

Paginated Responses

Automatic pagination metadata calculation:

@app.get("/items")
async def list_items(
    page: int = 1,
    page_size: int = 20
) -> PaginatedResponseSchema[ItemSchema]:
    all_items = get_all_items()

    start = (page - 1) * page_size
    end = start + page_size
    page_items = all_items[start:end]

    return paginated_response(
        items=page_items,
        total_results=len(all_items),
        page=page,
        page_size=page_size,
    )

Response includes:

• items: List of items for the current page
• current_page: Current page number
• total_pages: Total number of pages
• total_results: Total number of items
• page_size: Items per page
• has_next: Boolean indicating if next page exists
• has_previous: Boolean indicating if previous page exists

Custom Validation

Add custom validation with __post_init__:

class CreateUserRequest(msgspec.Struct):
    username: str
    email: str
    age: int

    def __post_init__(self) -> None:
        if len(self.username) < 3:
            raise ValueError("Username must be at least 3 characters")
        if not "@" in self.email:
            raise ValueError("Invalid email format")
        if self.age < 0 or self.age > 150:
            raise ValueError("Age must be between 0 and 150")

Field Name Conversion

Automatic camelCase conversion for JSON responses:

class User(msgspec.Struct, rename="camel"):
    id: int
    full_name: str
    email_address: str
    is_active: bool

# JSON output:
# {
#   "id": 1,
#   "fullName": "John Doe",
#   "emailAddress": "[e-mail]",
#   "isActive": true
# }

Advanced Features

Enum Support

Enums are fully supported in OpenAPI documentation. When you use Python enums in your msgspec structs, the generated OpenAPI schema will show all valid enum values and validate input accordingly.

from enum import Enum

class UserRole(Enum):
    ADMIN = "admin"
    MODERATOR = "moderator"
    USER = "user"

class User(msgspec.Struct, rename="camel"):
    id: int
    username: str
    role: UserRole

# OpenAPI will show:
# {
#   "role": {
#     "type": "string",
#     "enum": ["admin", "moderator", "user"]
#   }
# }

Invalid enum values in requests will return a 422 validation error.

Field Metadata (Description and Examples)

You can add descriptions and examples to your fields using msgspec.Meta. These will appear in the OpenAPI documentation.

from typing import Annotated

class Address(msgspec.Struct, rename="camel", kw_only=True):
    street: Annotated[
        str,
        msgspec.Meta(
            description="Street address including number",
            examples=["123 Main St", "456 Oak Avenue"],
        ),
    ]
    city: Annotated[
        str,
        msgspec.Meta(
            description="City name",
            examples=["New York", "Los Angeles"],
        ),
    ]
    postal_code: Annotated[
        str | None,
        msgspec.Meta(
            description="Postal or ZIP code",
            examples=["10001", "90210"],
        ),
    ] = None

The OpenAPI schema will include:

{
  "street": {
    "type": "string",
    "description": "Street address including number",
    "examples": ["123 Main St", "456 Oak Avenue"]
  }
}

DateTime, Date, and UUID Types

The library automatically generates proper OpenAPI format annotations for common Python types. This means your API documentation will show the correct data format expectations.

from datetime import date, datetime
from uuid import UUID

class Event(msgspec.Struct, rename="camel"):
    id: UUID
    name: str
    start_time: datetime
    end_date: date | None = None

OpenAPI will show proper format annotations:

{
  "id": {"type": "string", "format": "uuid"},
  "startTime": {"type": "string", "format": "date-time"},
  "endDate": {"type": "string", "format": "date"}
}

Supported types and their OpenAPI formats:

Python Type	OpenAPI Format
datetime	date-time
date	date
time	time
timedelta	duration
UUID	uuid
bytes	binary
Decimal	decimal

Tagged Unions

Support for discriminated unions:

from typing import Literal

class Circle(msgspec.Struct, tag="circle"):
    type: Literal["circle"]
    radius: float

class Rectangle(msgspec.Struct, tag="rectangle"):
    type: Literal["rectangle"]
    width: float
    height: float

Shape = Circle | Rectangle

class Drawing(msgspec.Struct):
    shapes: list[Shape]

# Automatic discrimination based on "type" field

Nested Structs

You can nest msgspec structs within each other. The library handles nested schema generation automatically.

class Address(msgspec.Struct, rename="camel"):
    street: str
    city: str
    country: str

class SocialLinks(msgspec.Struct, rename="camel"):
    twitter: str | None = None
    linkedin: str | None = None

class UserProfile(msgspec.Struct, rename="camel"):
    id: int
    username: str
    address: Address | None = None
    social_links: SocialLinks | None = None

All nested structs are converted to their corresponding Pydantic schemas for OpenAPI documentation.

Thread Safety

The schema conversion is thread-safe. The library uses a reentrant lock to handle concurrent schema generation in multi-threaded environments like production ASGI servers. This means:

• Multiple requests can safely trigger schema conversion simultaneously
• Nested structs are handled correctly without deadlocks
• Circular references between structs are detected and handled with forward references

The schema registry caches converted schemas, so each struct is only converted once regardless of how many threads access it.

Optional Cython Optimizations

Enable additional performance improvements:

make compile-cython

Provides 2-3x faster type conversion operations. The library automatically falls back to pure Python if Cython extensions are not available.

Performance

Benchmark results (see PERFORMANCE.md for details):

Operation	Pydantic	fastapi-advanced	Improvement
Response Serialization	1.00x	5.12x	5.1x faster
Object Creation	1.00x	12.45x	12.4x faster
Type Conversion	1.00x	2.0-3.0x	2-3x faster (with Cython)

Note: Request validation uses Pydantic (for OpenAPI docs). Response serialization uses msgspec (for performance).

Type Checking

Full mypy strict compliance:

make typecheck

The library uses stub files (.pyi) to provide perfect type inference without runtime overhead.

Testing

# Run all tests
make test

# Run with coverage
make test-cov

# Run performance benchmarks
make benchmark

Migration Guide

From Pure Pydantic

1. Replace Pydantic models with msgspec Structs:

# Before
from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str

# After
import msgspec

class User(msgspec.Struct):
    id: int
    name: str

2. Add schema conversion:

from typing import TYPE_CHECKING
from fastapi_advanced import msgspec_to_pydantic

if TYPE_CHECKING:
    UserSchema = User
else:
    UserSchema = msgspec_to_pydantic(User)

3. Update return types:

# Before
@app.get("/users")
async def get_users() -> list[User]:
    return [User(id=1, name="Alice")]

# After
@app.get("/users")
async def get_users() -> ResponseModelSchema[list[UserSchema]]:
    users = [User(id=1, name="Alice")]
    return response(data=users)

4. Setup msgspec integration:

from fastapi_advanced import setup_msgspec

app = FastAPI()
setup_msgspec(app)  # One line!

API Reference

Core Functions

• setup_msgspec(app: FastAPI) -> FastAPI: Initialize msgspec integration
• response(data, message, status, status_code) -> ResponseModelSchema[T]: Create standard response
• paginated_response(items, total_results, page, page_size, ...) -> PaginatedResponseSchema[T]: Create paginated response
• msgspec_to_pydantic(struct_cls) -> type[BaseModel]: Convert msgspec Struct to Pydantic model
• as_body(struct_cls) -> type[BaseModel]: Convert msgspec Struct for request body

Response Models

• ResponseModelSchema[T]: Type-safe response wrapper (Pydantic for OpenAPI)
• PaginatedResponseSchema[T]: Type-safe paginated response wrapper
• ResponseModel[T]: Runtime response model (msgspec)
• PaginatedResponse[T]: Runtime paginated response model (msgspec)

Response Classes

• MsgspecJSONResponse: FastAPI JSONResponse subclass using msgspec serialization

Contributing

Contributions are welcome. Please ensure:

1. Code passes all checks: make check
2. Tests pass: make test
3. Type checking passes: make typecheck
4. Code is formatted: make format

License

MIT License - see LICENSE file for details.

Acknowledgments

Built on top of:

• FastAPI - Modern web framework
• msgspec - Fast serialization library
• Pydantic - Data validation