A pluggable API specification generator for OpenAPI/Swagger specifications
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Official plugin for converting marshmallow schemas to OpenAPI format with automatic field type mapping, validation constraint extraction, and schema reference management. Enables seamless integration between marshmallow serialization and OpenAPI specification generation.
Main plugin class for marshmallow integration with automatic field-to-OpenAPI type conversion and schema processing.
class MarshmallowPlugin(BasePlugin):
Converter = OpenAPIConverter # Field converter class for OpenAPI type mapping
Resolver = SchemaResolver # Schema resolver class for reference resolution
def __init__(
self,
schema_name_resolver: Callable[[type[Schema]], str] | None = None
):
"""
Initialize marshmallow plugin.
Parameters:
- schema_name_resolver: Function to generate schema definition names from Schema classes
The resolver receives a Schema class and returns the name used in $ref within the generated spec.
For circular references, this function must not return None.
"""Convert marshmallow schemas to OpenAPI schema definitions with automatic field type mapping and constraint extraction.
def schema_helper(self, name, _, schema=None, **kwargs):
"""
Convert marshmallow Schema to OpenAPI schema definition.
Parameters:
- name: Schema name identifier
- schema: Marshmallow Schema class or instance
- **kwargs: Additional arguments (unused)
Returns:
OpenAPI schema definition dictionary or None if no schema provided
The plugin automatically:
- Maps marshmallow field types to OpenAPI types
- Extracts validation constraints (min/max, required, etc.)
- Handles nested schemas and references
- Processes field metadata for OpenAPI properties
"""Process parameters, responses, and headers that contain marshmallow schemas.
def parameter_helper(self, parameter, **kwargs):
"""
Process parameter definition containing marshmallow Schema.
Parameters:
- parameter: Parameter definition dictionary
- **kwargs: Additional arguments
Returns:
Processed parameter definition with resolved schema references
"""
def response_helper(self, response, **kwargs):
"""
Process response definition containing marshmallow Schema.
Parameters:
- response: Response definition dictionary
- **kwargs: Additional arguments
Returns:
Processed response definition with resolved schema references
"""
def header_helper(self, header: dict, **kwargs: Any):
"""
Process header definition containing marshmallow Schema.
Parameters:
- header: Header definition dictionary
- **kwargs: Additional arguments
Returns:
Processed header definition with resolved schema references
"""Process operations to resolve marshmallow schemas in request/response definitions.
def operation_helper(
self,
path: str | None = None,
operations: dict | None = None,
**kwargs: Any
) -> None:
"""
Process operations to resolve marshmallow schemas.
Parameters:
- path: API path
- operations: Dictionary of operations by HTTP method
- **kwargs: Additional arguments
Resolves schema references in operation parameters, request bodies, and responses.
"""Customize field type mappings for OpenAPI conversion.
def map_to_openapi_type(self, field_cls, *args):
"""
Set custom mapping for marshmallow field class to OpenAPI type.
Parameters:
- field_cls: Marshmallow field class to map
- *args: Either (type, format) pair or existing marshmallow field class
Examples:
- plugin.map_to_openapi_type(CustomField, 'string', 'custom-format')
- plugin.map_to_openapi_type(CustomField, fields.String) # Reuse String mapping
"""Default marshmallow field to OpenAPI type mappings used by the plugin.
DEFAULT_FIELD_MAPPING: dict[type, tuple[str | None, str | None]] = {
marshmallow.fields.Integer: ("integer", None),
marshmallow.fields.Number: ("number", None),
marshmallow.fields.Float: ("number", None),
marshmallow.fields.Decimal: ("number", None),
marshmallow.fields.String: ("string", None),
marshmallow.fields.Boolean: ("boolean", None),
marshmallow.fields.UUID: ("string", "uuid"),
marshmallow.fields.DateTime: ("string", "date-time"),
marshmallow.fields.Date: ("string", "date"),
marshmallow.fields.Time: ("string", None),
marshmallow.fields.TimeDelta: ("integer", None),
marshmallow.fields.Email: ("string", "email"),
marshmallow.fields.URL: ("string", "url"),
marshmallow.fields.Dict: ("object", None),
marshmallow.fields.Field: (None, None),
marshmallow.fields.Raw: (None, None),
marshmallow.fields.List: ("array", None),
marshmallow.fields.IP: ("string", "ip"),
marshmallow.fields.IPv4: ("string", "ipv4"),
marshmallow.fields.IPv6: ("string", "ipv6"),
}Each mapping is a tuple of (type, format) where:
type: OpenAPI data type (string, integer, number, boolean, array, object, or None)format: OpenAPI format specification (email, uuid, date, date-time, ipv4, ipv6, etc., or None)Utility functions for working with marshmallow schemas in OpenAPI context.
def resolve_schema_instance(
schema: type[Schema] | Schema | str
) -> Schema:
"""
Return schema instance for given schema (class, instance, or name).
Parameters:
- schema: Schema class, instance, or registered class name
Returns:
Schema instance ready for processing
"""
def resolve_schema_cls(
schema: type[Schema] | str | Schema
) -> type[Schema] | list[type[Schema]]:
"""
Return schema class for given schema (class, instance, or name).
Parameters:
- schema: Schema class, instance, or registered class name
Returns:
Schema class or list of classes for union types
"""def resolver(schema: type[Schema]) -> str:
"""
Default schema name resolver that strips 'Schema' suffix from class name.
Parameters:
- schema: Schema class
Returns:
Schema name for use in OpenAPI references
Example: UserSchema -> User, PersonSchema -> Person
"""def get_fields(
schema: type[Schema] | Schema,
*,
exclude_dump_only: bool = False
) -> dict:
"""
Get fields from marshmallow schema.
Parameters:
- schema: Schema class or instance
- exclude_dump_only: Whether to exclude dump-only fields
Returns:
Dictionary of field name to field instance mappings
"""
def make_schema_key(schema: Schema) -> tuple:
"""
Create unique key for schema instance for caching and reference tracking.
Parameters:
- schema: Schema instance
Returns:
Tuple representing unique schema key
"""
def get_unique_schema_name(components: Components, name: str, counter: int = 0) -> str:
"""
Generate unique schema name when conflicts occur.
Parameters:
- components: Components instance from APISpec
- name: Base name to make unique
- counter: Recursion counter for numbering
Returns:
Unique schema name not already in components.schemas
Automatically appends numbers when name conflicts occur.
"""Schema processing constants and modifiers.
MODIFIERS: list[str] # ['only', 'exclude', 'load_only', 'dump_only', 'partial']from apispec import APISpec
from apispec.ext.marshmallow import MarshmallowPlugin
from marshmallow import Schema, fields
# Create spec with marshmallow plugin
spec = APISpec(
title="Pet Store API",
version="1.0.0",
openapi_version="3.0.2",
plugins=[MarshmallowPlugin()]
)
# Define marshmallow schemas
class UserSchema(Schema):
id = fields.Int(dump_only=True)
username = fields.Str(required=True, validate=lambda x: len(x) >= 3)
email = fields.Email(required=True)
created_at = fields.DateTime(dump_only=True)
is_active = fields.Bool(load_default=True)
# Register schema - plugin automatically converts to OpenAPI
spec.components.schema("User", schema=UserSchema)from apispec.ext.marshmallow.common import resolve_schema_cls
def custom_schema_resolver(schema):
"""Custom schema name resolver."""
schema_cls = resolve_schema_cls(schema)
name = schema_cls.__name__
# Custom naming logic
if name.endswith('Schema'):
name = name[:-6]
# Add version suffix for versioned schemas
if hasattr(schema_cls, '__version__'):
name += f"V{schema_cls.__version__}"
return name
# Use custom resolver
plugin = MarshmallowPlugin(schema_name_resolver=custom_schema_resolver)
spec = APISpec("API", "1.0.0", "3.0.2", plugins=[plugin])from marshmallow import fields
# Custom field with specific OpenAPI mapping
class TimestampField(fields.Integer):
"""Unix timestamp field."""
pass
# Map custom field to OpenAPI type
plugin.map_to_openapi_type(TimestampField, 'integer', 'int64')
# Alternatively, reuse existing field mapping
plugin.map_to_openapi_type(TimestampField, fields.Integer)class ProductSchema(Schema):
name = fields.Str(
required=True,
metadata={
"description": "Product name",
"example": "Premium Widget"
}
)
price = fields.Decimal(
required=True,
validate=lambda x: x > 0,
metadata={
"description": "Product price in USD",
"example": 29.99
}
)
category = fields.Str(
metadata={
"description": "Product category",
"enum": ["electronics", "clothing", "books"]
}
)
# Metadata is automatically included in OpenAPI schema
spec.components.schema("Product", schema=ProductSchema)# Define request/response schemas
class CreateUserRequest(Schema):
username = fields.Str(required=True)
email = fields.Email(required=True)
password = fields.Str(required=True, load_only=True)
class UserResponse(Schema):
id = fields.Int()
username = fields.Str()
email = fields.Email()
created_at = fields.DateTime()
# Register schemas
spec.components.schema("CreateUserRequest", schema=CreateUserRequest)
spec.components.schema("UserResponse", schema=UserResponse)
# Use in path operations
spec.path("/users", operations={
"post": {
"summary": "Create new user",
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/CreateUserRequest"}
}
}
},
"responses": {
"201": {
"description": "User created successfully",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/UserResponse"}
}
}
}
}
}
})class AddressSchema(Schema):
street = fields.Str(required=True)
city = fields.Str(required=True)
state = fields.Str(required=True)
zip_code = fields.Str(required=True)
class UserProfileSchema(Schema):
id = fields.Int(dump_only=True)
name = fields.Str(required=True)
email = fields.Email(required=True)
address = fields.Nested(AddressSchema) # Automatically creates reference
addresses = fields.List(fields.Nested(AddressSchema)) # Array of references
# Plugin automatically handles nested schema references
spec.components.schema("UserProfile", schema=UserProfileSchema)
# This also registers AddressSchema automatically via nested referenceInstall with Tessl CLI
npx tessl i tessl/pypi-apispec