Fully featured framework for fast, easy and documented API development with Flask
—
Flask-RESTX provides a comprehensive schema definition system for complex data structures with JSON Schema validation, inheritance support, and automatic documentation generation. Models enable declarative API contracts with comprehensive validation and serve as the foundation for request/response documentation.
Foundation classes for all model types with common functionality.
class ModelBase:
def __init__(self, name, *args, **kwargs):
"""
Base class for all models.
Parameters:
- name: Model name for documentation and references
- args, kwargs: Additional parameters for subclasses
"""
def inherit(self, name, *parents):
"""
Create inherited model using Swagger composition pattern.
Parameters:
- name: New model name
- parents: Parent models to inherit from
Returns:
New model instance inheriting from parents
"""
def validate(self, data, resolver=None, format_checker=None):
"""
Validate data against model schema.
Parameters:
- data: Data to validate
- resolver: JSON schema resolver for references
- format_checker: JSON schema format checker
Raises:
ValidationError: If validation fails
"""
def get_parent(self, name):
"""
Get parent model by name.
Parameters:
- name: Parent model name
Returns:
Parent model instance
Raises:
ValueError: If parent not found
"""
@property
def ancestors(self):
"""Set of all ancestor model names."""
@property
def __schema__(self):
"""JSON Schema representation of the model."""
@property
def name(self):
"""Model name."""Concrete model implementations for different use cases.
class Model(dict):
def __init__(self, name, fields_dict=None, mask=None, strict=False, **kwargs):
"""
Dictionary-based model with field definitions.
Parameters:
- name: Model name
- fields_dict: Dictionary of field name to field definition
- mask: Optional field mask for selective serialization
- strict: Whether to enforce strict validation
- kwargs: Additional model parameters
"""
def __getitem__(self, key):
"""Get field definition by name."""
def __setitem__(self, key, value):
"""Set field definition by name."""
@classmethod
def clone(cls, name, *parents):
"""
Create a clone with additional fields.
Parameters:
- name: New model name
- parents: Parent models or field dictionaries
Returns:
New model instance
"""
@classmethod
def inherit(cls, name, *parents):
"""
Create inherited model using composition pattern.
Parameters:
- name: New model name
- parents: Parent models to inherit from
Returns:
New model instance
"""
def extend(self, name, fields_dict):
"""
Extend model with additional fields (deprecated).
Parameters:
- name: New model name
- fields_dict: Dictionary of additional fields
Returns:
Extended model instance
"""
def validate(self, data, resolver=None, format_checker=None):
"""
Validate data against model schema.
Parameters:
- data: Data to validate
- resolver: JSON schema resolver
- format_checker: JSON schema format checker
Raises:
ValidationError: If validation fails
"""
@property
def __schema__(self):
"""JSON Schema representation."""
@property
def resolved(self):
"""Resolved field definitions."""
class OrderedModel(Model):
def __init__(self, name, fields_dict=None, mask=None, strict=False, **kwargs):
"""
Ordered dictionary-based model preserving field order.
Parameters:
- name: Model name
- fields_dict: Ordered dictionary of field definitions
- mask: Optional field mask
- strict: Whether to enforce strict validation
- kwargs: Additional model parameters
"""
class SchemaModel(ModelBase):
def __init__(self, name, schema=None, **kwargs):
"""
JSON Schema-based model.
Parameters:
- name: Model name
- schema: JSON Schema definition
- kwargs: Additional model parameters
"""
@property
def _schema(self):
"""Internal schema representation."""
@property
def __schema__(self):
"""JSON Schema representation."""Field filtering and partial response functionality.
class Mask:
def __init__(self, mask=None, skip=False, **kwargs):
"""
Field mask for selective serialization.
Parameters:
- mask: Mask definition (string, dict, or Mask instance)
- skip: Whether to skip missing fields
- kwargs: Additional mask parameters
"""
def parse(self, mask):
"""
Parse mask string into field selection structure.
Parameters:
- mask: Mask string in format '{field1,field2{nested1,nested2}}'
"""
def apply(self, data, skip=False):
"""
Apply mask to data structure.
Parameters:
- data: Data to filter
- skip: Whether to skip missing fields
Returns:
Filtered data structure
"""
def apply(data, mask, skip=False):
"""
Apply mask to data structure.
Parameters:
- data: Data to filter
- mask: Mask instance or definition
- skip: Whether to skip missing fields
Returns:
Filtered data according to mask
"""
def format_error(error):
"""
Format validation error for display.
Parameters:
- error: Validation error instance
Returns:
Formatted error message
"""Exception types for mask parsing and application errors.
class MaskError(RestError):
def __init__(self, msg):
"""
Base exception for mask operations.
Parameters:
- msg: Error message
"""
class ParseError(MaskError):
def __init__(self, msg):
"""
Exception raised when mask parsing fails.
Parameters:
- msg: Parse error details
"""from flask_restx import Api, fields
api = Api()
# Define a simple model
user_model = api.model('User', {
'id': fields.Integer(required=True, description='User ID'),
'name': fields.String(required=True, description='Full name'),
'email': fields.String(required=True, description='Email address'),
'active': fields.Boolean(default=True, description='Account status'),
'created_at': fields.DateTime(description='Account creation time')
})# Address model
address_model = api.model('Address', {
'street': fields.String(required=True),
'city': fields.String(required=True),
'state': fields.String,
'country': fields.String(required=True),
'postal_code': fields.String
})
# User model with nested address
user_with_address = api.model('UserWithAddress', {
'id': fields.Integer(required=True),
'name': fields.String(required=True),
'email': fields.String(required=True),
'address': fields.Nested(address_model),
'billing_address': fields.Nested(address_model, allow_null=True)
})# Base person model
person_model = api.model('Person', {
'name': fields.String(required=True),
'email': fields.String(required=True),
'phone': fields.String
})
# Employee model inheriting from person
employee_model = api.inherit('Employee', person_model, {
'employee_id': fields.String(required=True),
'department': fields.String(required=True),
'hire_date': fields.Date,
'salary': fields.Float(min=0)
})
# Manager model inheriting from employee
manager_model = api.inherit('Manager', employee_model, {
'team_size': fields.Integer(min=0),
'reports': fields.List(fields.Nested(employee_model))
})# Base product model
product_model = api.model('Product', {
'id': fields.Integer(required=True),
'name': fields.String(required=True),
'price': fields.Float(required=True, min=0)
})
# Extend with additional fields
detailed_product = api.clone('DetailedProduct', product_model, {
'description': fields.String,
'category': fields.String,
'tags': fields.List(fields.String),
'in_stock': fields.Boolean(default=True)
})
# Alternative using extend method
extended_product = product_model.extend('ExtendedProduct', {
'manufacturer': fields.String,
'warranty_months': fields.Integer(min=0)
})# JSON Schema definition
user_schema = {
"type": "object",
"properties": {
"name": {"type": "string", "minLength": 1},
"age": {"type": "integer", "minimum": 0, "maximum": 150},
"email": {"type": "string", "format": "email"}
},
"required": ["name", "email"]
}
# Create schema model
schema_user_model = api.schema_model('SchemaUser', user_schema)from flask_restx import Resource, ValidationError
@api.route('/users')
class UserList(Resource):
@api.expect(user_model, validate=True)
def post(self):
# Request payload automatically validated against model
data = api.payload
try:
# Manual validation if needed
user_model.validate(data)
except ValidationError as e:
return {'error': str(e)}, 400
# Process valid data
return {'message': 'User created', 'data': data}, 201from flask_restx import Mask
# Create mask for selective fields
mask = Mask('name,email,address{city,country}')
# Apply mask to model data
user_data = {
'id': 1,
'name': 'John Doe',
'email': 'john@example.com',
'phone': '555-1234',
'address': {
'street': '123 Main St',
'city': 'Anytown',
'state': 'CA',
'country': 'USA',
'postal_code': '12345'
}
}
# Result includes only masked fields
filtered_data = mask.apply(user_data)
# {'name': 'John Doe', 'email': 'john@example.com', 'address': {'city': 'Anytown', 'country': 'USA'}}@api.route('/users/<int:user_id>')
class User(Resource):
@api.marshal_with(user_model)
@api.doc('get_user')
def get(self, user_id):
"""Fetch a user by ID"""
return find_user(user_id)
@api.expect(user_model, validate=True)
@api.marshal_with(user_model)
@api.doc('update_user')
@api.response(200, 'User updated', user_model)
@api.response(400, 'Validation error')
@api.response(404, 'User not found')
def put(self, user_id):
"""Update a user"""
data = api.payload
updated_user = update_user(user_id, data)
return updated_userfrom collections import OrderedDict
# Preserve field order in responses
ordered_user = api.model('OrderedUser', OrderedDict([
('id', fields.Integer(required=True)),
('name', fields.String(required=True)),
('email', fields.String(required=True)),
('created_at', fields.DateTime),
('updated_at', fields.DateTime)
]))
# Or use OrderedModel directly
ordered_product = OrderedModel('Product', OrderedDict([
('id', fields.Integer),
('sku', fields.String),
('name', fields.String),
('price', fields.Float),
('availability', fields.Boolean)
]))# Model with complex validation rules
order_model = api.model('Order', {
'id': fields.Integer(required=True),
'customer_id': fields.Integer(required=True, min=1),
'items': fields.List(fields.Nested(api.model('OrderItem', {
'product_id': fields.Integer(required=True),
'quantity': fields.Integer(required=True, min=1),
'unit_price': fields.Float(required=True, min=0)
})), required=True, min_items=1),
'total_amount': fields.Float(required=True, min=0),
'order_date': fields.DateTime(required=True),
'status': fields.String(required=True, enum=['pending', 'confirmed', 'shipped', 'delivered'])
})
# Custom validation logic
@api.route('/orders')
class OrderList(Resource):
@api.expect(order_model, validate=True)
def post(self):
data = api.payload
# Additional business logic validation
calculated_total = sum(item['quantity'] * item['unit_price'] for item in data['items'])
if abs(calculated_total - data['total_amount']) > 0.01:
return {'error': 'Total amount does not match item calculations'}, 400
return {'message': 'Order created'}, 201Install with Tessl CLI
npx tessl i tessl/pypi-flask-restx