Fully featured framework for fast, easy and documented API development with Flask
—
Flask-RESTX provides comprehensive error handling capabilities with HTTP status code management, custom exception types, and utilities for API responses. It also includes CORS support and Swagger documentation generation for complete API development support.
Core functions for handling API errors and exceptions.
def abort(code=500, message=None, **kwargs):
"""
Abort the current request with HTTP status code and message.
Parameters:
- code: HTTP status code (default: 500)
- message: Optional error message
- kwargs: Additional data to include in error response
Raises:
HTTPException: Flask HTTP exception with specified status code
"""Custom exception types for different error scenarios.
class RestError(Exception):
def __init__(self, msg):
"""
Base class for all Flask-RESTX errors.
Parameters:
- msg: Error message
"""
def __str__(self):
"""Return error message as string."""
class ValidationError(RestError):
def __init__(self, msg):
"""
Exception for input validation errors.
Parameters:
- msg: Validation error details
"""
class SpecsError(RestError):
def __init__(self, msg):
"""
Exception for API specification errors.
Parameters:
- msg: Specification error details
"""Cross-Origin Resource Sharing functionality for API access control.
def crossdomain(
origin=None,
methods=None,
headers=None,
expose_headers=None,
max_age=21600,
attach_to_all=True,
automatic_options=True,
credentials=False
):
"""
Decorator for enabling CORS on API endpoints.
Parameters:
- origin: Allowed origins ('*' for all, list for specific origins)
- methods: Allowed HTTP methods (list of strings)
- headers: Allowed request headers (list of strings)
- expose_headers: Headers to expose to client (list of strings)
- max_age: Preflight request cache duration in seconds
- attach_to_all: Whether to attach CORS headers to all responses
- automatic_options: Whether to automatically handle OPTIONS requests
- credentials: Whether to allow credentials in CORS requests
Returns:
Decorator function for CORS-enabled endpoints
"""Automatic API documentation generation with OpenAPI/Swagger specification.
class Swagger:
def __init__(self, api):
"""
Swagger documentation generator.
Parameters:
- api: Api instance to generate documentation for
"""
def as_dict(self):
"""
Generate Swagger specification as dictionary.
Returns:
dict: Complete Swagger/OpenAPI specification
"""
@property
def paths(self):
"""Dictionary of API paths and operations."""
@property
def definitions(self):
"""Dictionary of model definitions."""
@property
def tags(self):
"""List of API tags."""Comprehensive HTTP status code constants for consistent error handling.
class HTTPStatus:
"""
HTTP status code constants with names, phrases, and descriptions.
Complete enumeration of all standard HTTP status codes.
"""
# Informational 1xx
CONTINUE = 100
SWITCHING_PROTOCOLS = 101
PROCESSING = 102
# Success 2xx
OK = 200
CREATED = 201
ACCEPTED = 202
NON_AUTHORITATIVE_INFORMATION = 203
NO_CONTENT = 204
RESET_CONTENT = 205
PARTIAL_CONTENT = 206
MULTI_STATUS = 207
ALREADY_REPORTED = 208
IM_USED = 226
# Redirection 3xx
MULTIPLE_CHOICES = 300
MOVED_PERMANENTLY = 301
FOUND = 302
SEE_OTHER = 303
NOT_MODIFIED = 304
USE_PROXY = 305
TEMPORARY_REDIRECT = 307
PERMANENT_REDIRECT = 308
# Client Error 4xx
BAD_REQUEST = 400
UNAUTHORIZED = 401
PAYMENT_REQUIRED = 402
FORBIDDEN = 403
NOT_FOUND = 404
METHOD_NOT_ALLOWED = 405
NOT_ACCEPTABLE = 406
PROXY_AUTHENTICATION_REQUIRED = 407
REQUEST_TIMEOUT = 408
CONFLICT = 409
GONE = 410
LENGTH_REQUIRED = 411
PRECONDITION_FAILED = 412
REQUEST_ENTITY_TOO_LARGE = 413
REQUEST_URI_TOO_LONG = 414
UNSUPPORTED_MEDIA_TYPE = 415
REQUESTED_RANGE_NOT_SATISFIABLE = 416
EXPECTATION_FAILED = 417
UNPROCESSABLE_ENTITY = 422
LOCKED = 423
FAILED_DEPENDENCY = 424
UPGRADE_REQUIRED = 426
PRECONDITION_REQUIRED = 428
TOO_MANY_REQUESTS = 429
REQUEST_HEADER_FIELDS_TOO_LARGE = 431
# Server Error 5xx
INTERNAL_SERVER_ERROR = 500
NOT_IMPLEMENTED = 501
BAD_GATEWAY = 502
SERVICE_UNAVAILABLE = 503
GATEWAY_TIMEOUT = 504
HTTP_VERSION_NOT_SUPPORTED = 505
VARIANT_ALSO_NEGOTIATES = 506
INSUFFICIENT_STORAGE = 507
LOOP_DETECTED = 508
NOT_EXTENDED = 510
NETWORK_AUTHENTICATION_REQUIRED = 511
# Utility functions for HTTP status handling
def is_informational(status_code):
"""Check if status code is informational (1xx)."""
def is_success(status_code):
"""Check if status code indicates success (2xx)."""
def is_redirect(status_code):
"""Check if status code indicates redirection (3xx)."""
def is_client_error(status_code):
"""Check if status code indicates client error (4xx)."""
def is_server_error(status_code):
"""Check if status code indicates server error (5xx)."""Functions for creating and formatting API responses.
def output_json(data, code, headers=None):
"""
Create JSON response with proper headers.
Parameters:
- data: Response data to serialize as JSON
- code: HTTP status code
- headers: Optional additional headers
Returns:
Flask Response object with JSON content
"""
def make_response(data, code=200, headers=None):
"""
Create Flask response from data.
Parameters:
- data: Response data
- code: HTTP status code
- headers: Optional response headers
Returns:
Flask Response object
"""from flask_restx import Resource, abort, HTTPStatus
class UserResource(Resource):
def get(self, user_id):
user = find_user(user_id)
if not user:
abort(HTTPStatus.NOT_FOUND, message=f'User {user_id} not found')
if not user.is_active:
abort(HTTPStatus.FORBIDDEN, message='User account is disabled')
return user
def delete(self, user_id):
if not is_admin():
abort(HTTPStatus.UNAUTHORIZED, message='Admin access required')
try:
delete_user(user_id)
return '', HTTPStatus.NO_CONTENT
except UserNotFound:
abort(HTTPStatus.NOT_FOUND, message='User not found')
except DatabaseError:
abort(HTTPStatus.INTERNAL_SERVER_ERROR, message='Database error occurred')from flask_restx import Api
api = Api()
@api.errorhandler(ValidationError)
def handle_validation_error(error):
"""Handle validation errors with custom response format."""
return {
'error': 'Validation failed',
'message': str(error),
'code': 'VALIDATION_ERROR'
}, HTTPStatus.BAD_REQUEST
@api.errorhandler(RestError)
def handle_rest_error(error):
"""Handle general REST API errors."""
return {
'error': 'API Error',
'message': str(error),
'code': 'REST_ERROR'
}, HTTPStatus.INTERNAL_SERVER_ERROR
@api.errorhandler(404)
def handle_not_found(error):
"""Handle 404 errors with custom format."""
return {
'error': 'Resource not found',
'message': 'The requested resource could not be found',
'code': 'NOT_FOUND'
}, HTTPStatus.NOT_FOUNDclass OrderResource(Resource):
def post(self):
try:
data = api.payload
order = create_order(data)
return order, HTTPStatus.CREATED
except InsufficientStock as e:
abort(
HTTPStatus.CONFLICT,
message='Insufficient stock',
details={
'requested_quantity': e.requested,
'available_quantity': e.available,
'product_id': e.product_id
},
code='INSUFFICIENT_STOCK'
)
except PaymentFailed as e:
abort(
HTTPStatus.PAYMENT_REQUIRED,
message='Payment processing failed',
details={
'payment_method': e.method,
'failure_reason': e.reason
},
code='PAYMENT_FAILED'
)from flask_restx import cors
# Enable CORS for all origins
@api.route('/public-data')
class PublicData(Resource):
@cors.crossdomain(origin='*')
def get(self):
return {'data': 'public information'}
# Restrict CORS to specific origins
@api.route('/restricted-data')
class RestrictedData(Resource):
@cors.crossdomain(
origin=['https://example.com', 'https://app.example.com'],
methods=['GET', 'POST'],
headers=['Authorization', 'Content-Type']
)
def get(self):
return {'data': 'restricted information'}
@cors.crossdomain(
origin=['https://example.com'],
methods=['POST'],
headers=['Authorization', 'Content-Type'],
max_age=3600
)
def post(self):
return {'message': 'Data created'}, HTTPStatus.CREATEDfrom flask import Flask
from flask_restx import Api
from flask_cors import CORS
app = Flask(__name__)
# Enable CORS for entire Flask app
CORS(app, resources={
r"/api/*": {
"origins": ["https://example.com", "https://app.example.com"],
"methods": ["GET", "POST", "PUT", "DELETE"],
"allow_headers": ["Authorization", "Content-Type"]
}
})
api = Api(app, prefix='/api/v1')from werkzeug.exceptions import HTTPException
import logging
api = Api()
@api.errorhandler(Exception)
def handle_unexpected_error(error):
"""Handle all unexpected errors."""
logging.error(f"Unexpected error: {str(error)}", exc_info=True)
return {
'error': 'Internal server error',
'message': 'An unexpected error occurred'
}, HTTPStatus.INTERNAL_SERVER_ERROR
@api.errorhandler(HTTPException)
def handle_http_error(error):
"""Handle HTTP errors with consistent format."""
return {
'error': error.name,
'message': error.description,
'code': error.code
}, error.codefrom jsonschema import ValidationError as JSONValidationError
@api.errorhandler(JSONValidationError)
def handle_json_validation_error(error):
"""Handle JSON schema validation errors with detailed information."""
return {
'error': 'Validation failed',
'message': error.message,
'path': list(error.absolute_path),
'invalid_value': error.instance,
'schema_path': list(error.schema_path)
}, HTTPStatus.BAD_REQUEST
class ValidationResource(Resource):
@api.expect(user_model, validate=True)
def post(self):
# Validation automatically handled by Flask-RESTX
# Custom validation errors can still be raised
data = api.payload
if User.query.filter_by(email=data['email']).first():
abort(
HTTPStatus.CONFLICT,
message='Email already exists',
field='email',
code='DUPLICATE_EMAIL'
)
return create_user(data), HTTPStatus.CREATED# Custom Swagger configuration
api = Api(
app,
version='1.0',
title='My API',
description='Comprehensive REST API with detailed documentation',
terms_url='https://example.com/terms',
contact='support@example.com',
license='MIT',
license_url='https://opensource.org/licenses/MIT',
authorizations={
'Bearer': {
'type': 'apiKey',
'in': 'header',
'name': 'Authorization',
'description': 'JWT token in format: Bearer <token>'
},
'ApiKey': {
'type': 'apiKey',
'in': 'header',
'name': 'X-API-Key',
'description': 'API key for authentication'
}
}
)
# Access Swagger specification
@api.route('/swagger-spec')
class SwaggerSpec(Resource):
def get(self):
"""Return the complete Swagger specification."""
swagger = Swagger(api)
return swagger.as_dict()from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
limiter = Limiter(
app,
key_func=get_remote_address,
default_limits=["100 per hour"]
)
@api.errorhandler(429) # Too Many Requests
def handle_rate_limit_error(error):
"""Handle rate limiting errors."""
return {
'error': 'Rate limit exceeded',
'message': 'Too many requests. Please try again later.',
'retry_after': error.retry_after,
'code': 'RATE_LIMIT_EXCEEDED'
}, HTTPStatus.TOO_MANY_REQUESTS
class RateLimitedResource(Resource):
@limiter.limit("10 per minute")
def post(self):
return {'message': 'Request processed'}, HTTPStatus.OKInstall with Tessl CLI
npx tessl i tessl/pypi-flask-restx