High-performance FastAPI integration with msgspec for fast serialization and automatic OpenAPI support.
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.
- 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
pip install fastapi-advancedPre-compiled wheels are available for macOS, Linux, and Windows (Python 3.10-3.13).
git clone https://github.com/your-org/fastapi-advanced.git
cd fastapi-advanced
make install-dev
# Optional: Enable Cython optimizations
make compile-cythonfrom 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="[email protected]")
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 --reloadUse 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="[email protected]")
return response(data=users)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)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"
}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 pagecurrent_page: Current page numbertotal_pages: Total number of pagestotal_results: Total number of itemspage_size: Items per pagehas_next: Boolean indicating if next page existshas_previous: Boolean indicating if previous page exists
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")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": "[email protected]",
# "isActive": true
# }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.
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"],
),
] = NoneThe OpenAPI schema will include:
{
"street": {
"type": "string",
"description": "Street address including number",
"examples": ["123 Main St", "456 Oak Avenue"]
}
}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 = NoneOpenAPI 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 |
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" fieldYou 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 = NoneAll nested structs are converted to their corresponding Pydantic schemas for OpenAPI documentation.
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.
Enable additional performance improvements:
make compile-cythonProvides 2-3x faster type conversion operations. The library automatically falls back to pure Python if Cython extensions are not available.
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).
Full mypy strict compliance:
make typecheckThe library uses stub files (.pyi) to provide perfect type inference without runtime overhead.
# Run all tests
make test
# Run with coverage
make test-cov
# Run performance benchmarks
make benchmark- 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- 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)- 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)- Setup msgspec integration:
from fastapi_advanced import setup_msgspec
app = FastAPI()
setup_msgspec(app) # One line!setup_msgspec(app: FastAPI) -> FastAPI: Initialize msgspec integrationresponse(data, message, status, status_code) -> ResponseModelSchema[T]: Create standard responsepaginated_response(items, total_results, page, page_size, ...) -> PaginatedResponseSchema[T]: Create paginated responsemsgspec_to_pydantic(struct_cls) -> type[BaseModel]: Convert msgspec Struct to Pydantic modelas_body(struct_cls) -> type[BaseModel]: Convert msgspec Struct for request body
ResponseModelSchema[T]: Type-safe response wrapper (Pydantic for OpenAPI)PaginatedResponseSchema[T]: Type-safe paginated response wrapperResponseModel[T]: Runtime response model (msgspec)PaginatedResponse[T]: Runtime paginated response model (msgspec)
MsgspecJSONResponse: FastAPI JSONResponse subclass using msgspec serialization
Contributions are welcome. Please ensure:
- Code passes all checks:
make check - Tests pass:
make test - Type checking passes:
make typecheck - Code is formatted:
make format
MIT License - see LICENSE file for details.
Built on top of: