Fully featured framework for fast, easy and documented API development with Flask
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Structured error handling with HTTP status codes, custom exception classes, and automatic error response formatting. Flask-RESTPlus provides comprehensive error handling capabilities that integrate with automatic documentation generation.
Function for terminating requests with HTTP errors and structured error responses.
def abort(code=500, message=None, **kwargs):
"""
Abort the current request with an HTTP error.
Args:
code (int): HTTP status code (default: 500)
message (str, optional): Error message
**kwargs: Additional error data to include in response
Raises:
HTTPException: Flask HTTP exception with error details
"""Custom exception classes for different types of API errors.
class RestError(Exception):
def __init__(self, msg):
"""
Base class for all Flask-RESTPlus errors.
Args:
msg (str): Error message
"""
self.msg = msg
def __str__(self):
"""
String representation of the error.
Returns:
str: Error message
"""
class ValidationError(RestError):
def __init__(self, msg):
"""
Error for input validation failures.
Args:
msg (str): Validation error message
"""
class SpecsError(RestError):
def __init__(self, msg):
"""
Error for API specification issues.
Args:
msg (str): Specification error message
"""HTTP status code constants and utilities for consistent error responses.
# Available through flask_restplus._http.HTTPStatus
class HTTPStatus:
# Informational responses
CONTINUE = 100
SWITCHING_PROTOCOLS = 101
PROCESSING = 102
# Successful responses
OK = 200
CREATED = 201
ACCEPTED = 202
NO_CONTENT = 204
# Redirection messages
MOVED_PERMANENTLY = 301
FOUND = 302
NOT_MODIFIED = 304
# Client error responses
BAD_REQUEST = 400
UNAUTHORIZED = 401
FORBIDDEN = 403
NOT_FOUND = 404
METHOD_NOT_ALLOWED = 405
NOT_ACCEPTABLE = 406
CONFLICT = 409
GONE = 410
UNPROCESSABLE_ENTITY = 422
# Server error responses
INTERNAL_SERVER_ERROR = 500
NOT_IMPLEMENTED = 501
BAD_GATEWAY = 502
SERVICE_UNAVAILABLE = 503from flask_restplus import Api, Resource, abort
api = Api()
@api.route('/users/<int:user_id>')
class User(Resource):
def get(self, user_id):
# Simulate user lookup
user = find_user_by_id(user_id)
if not user:
# Abort with 404 Not Found
abort(404, message=f"User {user_id} not found")
if not user.is_active:
# Abort with custom message and additional data
abort(403,
message="Access denied",
reason="User account is inactive",
user_id=user_id)
return {'id': user.id, 'name': user.name}
def delete(self, user_id):
user = find_user_by_id(user_id)
if not user:
abort(404, message="User not found")
if user.is_admin:
abort(400,
message="Cannot delete admin user",
user_type="admin")
# Delete user...
return {'message': f'User {user_id} deleted'}, 200from flask_restplus import Api, Resource, abort, fields
api = Api()
# Define error response model for documentation
error_model = api.model('Error', {
'message': fields.String(required=True, description='Error message'),
'code': fields.Integer(description='Error code'),
'details': fields.Raw(description='Additional error details')
})
@api.route('/orders/<int:order_id>')
class Order(Resource):
@api.response(404, 'Order not found', error_model)
@api.response(400, 'Invalid request', error_model)
def get(self, order_id):
order = find_order(order_id)
if not order:
abort(404,
message="Order not found",
code="ORDER_NOT_FOUND",
order_id=order_id)
if order.status == 'cancelled':
abort(400,
message="Cannot access cancelled order",
code="ORDER_CANCELLED",
details={
'order_id': order_id,
'cancelled_at': order.cancelled_at.isoformat(),
'reason': order.cancellation_reason
})
return order.to_dict()from flask_restplus import Api
from werkzeug.exceptions import HTTPException
import logging
api = Api()
@api.errorhandler(ValidationError)
def handle_validation_error(error):
"""Handle validation errors."""
return {
'message': 'Validation failed',
'error': str(error),
'type': 'validation_error'
}, 400
@api.errorhandler(ValueError)
def handle_value_error(error):
"""Handle value errors."""
return {
'message': 'Invalid value provided',
'error': str(error),
'type': 'value_error'
}, 400
@api.errorhandler(KeyError)
def handle_key_error(error):
"""Handle missing key errors."""
return {
'message': 'Required field missing',
'field': str(error).strip("'\""),
'type': 'missing_field'
}, 400
@api.errorhandler(Exception)
def handle_unexpected_error(error):
"""Handle unexpected errors."""
logging.exception("Unexpected error occurred")
return {
'message': 'An unexpected error occurred',
'type': 'internal_error'
}, 500
# Handle specific HTTP exceptions
@api.errorhandler(404)
def handle_not_found(error):
"""Handle 404 errors."""
return {
'message': 'Resource not found',
'type': 'not_found'
}, 404from flask_restplus import Api, Namespace, Resource
api = Api()
ns = api.namespace('users', description='User operations')
@ns.errorhandler(ValidationError)
def handle_user_validation_error(error):
"""Handle validation errors in user namespace."""
return {
'message': 'User validation failed',
'error': str(error),
'namespace': 'users'
}, 400
@ns.errorhandler(ValueError)
def handle_user_value_error(error):
"""Handle value errors in user namespace."""
return {
'message': 'Invalid user data',
'error': str(error),
'namespace': 'users'
}, 400
@ns.route('/')
class UserList(Resource):
def post(self):
# Validation errors in this namespace will be handled by
# the namespace-specific error handlers above
user_data = api.payload
if not user_data.get('email'):
raise ValidationError("Email is required")
if '@' not in user_data['email']:
raise ValueError("Invalid email format")
return {'message': 'User created'}, 201from flask_restplus import Api, Resource, reqparse, fields
from flask_restplus.errors import ValidationError
api = Api()
# Model with validation
user_model = api.model('User', {
'name': fields.String(required=True, min_length=2, max_length=50),
'email': fields.String(required=True),
'age': fields.Integer(min=0, max=150)
})
@api.route('/users')
class UserList(Resource):
@api.expect(user_model, validate=True)
def post(self):
# Validation is automatic when validate=True
# Invalid data will trigger ValidationError
data = api.payload
# Additional custom validation
if '@' not in data['email']:
abort(400,
message="Invalid email format",
field="email",
value=data['email'])
# Process valid data...
return {'message': 'User created'}, 201
# Custom validation handler
@api.errorhandler(ValidationError)
def handle_validation_error(error):
return {
'message': 'Input validation failed',
'errors': error.msg if hasattr(error, 'msg') else str(error)
}, 400from flask_restplus import Api, Resource, reqparse, inputs
api = Api()
parser = reqparse.RequestParser(bundle_errors=True)
parser.add_argument('name', type=str, required=True, help='Name is required')
parser.add_argument('email', type=inputs.email(), required=True, help='Valid email required')
parser.add_argument('age', type=inputs.int_range(0, 150), help='Age must be 0-150')
@api.route('/register')
class Register(Resource):
@api.expect(parser)
def post(self):
try:
args = parser.parse_args()
# Process registration...
return {'message': 'Registration successful'}, 201
except Exception as e:
# Parser errors are automatically formatted
# With bundle_errors=True, all errors are returned together:
# {
# "message": "Input payload validation failed",
# "errors": {
# "name": "Name is required",
# "email": "Valid email required",
# "age": "Age must be 0-150"
# }
# }
abort(400, message=str(e))from flask_restplus import Api, Resource, abort
api = Api()
class InsufficientFundsError(Exception):
def __init__(self, balance, amount):
self.balance = balance
self.amount = amount
super().__init__(f"Insufficient funds: balance {balance}, requested {amount}")
class AccountLockedError(Exception):
def __init__(self, account_id, locked_until):
self.account_id = account_id
self.locked_until = locked_until
super().__init__(f"Account {account_id} locked until {locked_until}")
@api.errorhandler(InsufficientFundsError)
def handle_insufficient_funds(error):
return {
'message': 'Insufficient funds',
'balance': error.balance,
'requested_amount': error.amount,
'shortfall': error.amount - error.balance,
'error_code': 'INSUFFICIENT_FUNDS'
}, 400
@api.errorhandler(AccountLockedError)
def handle_account_locked(error):
return {
'message': 'Account temporarily locked',
'account_id': error.account_id,
'locked_until': error.locked_until.isoformat(),
'error_code': 'ACCOUNT_LOCKED'
}, 423 # HTTP 423 Locked
@api.route('/accounts/<int:account_id>/withdraw')
class Withdraw(Resource):
def post(self, account_id):
account = find_account(account_id)
amount = api.payload.get('amount', 0)
if not account:
abort(404, message="Account not found")
if account.is_locked():
raise AccountLockedError(account_id, account.locked_until)
if account.balance < amount:
raise InsufficientFundsError(account.balance, amount)
# Process withdrawal...
return {'message': 'Withdrawal successful', 'new_balance': account.balance - amount}from flask_restplus import Api, Resource, fields
api = Api()
# Define standard error models for documentation
error_model = api.model('Error', {
'message': fields.String(required=True, description='Error message'),
'error_code': fields.String(description='Machine-readable error code'),
'details': fields.Raw(description='Additional error details')
})
validation_error_model = api.model('ValidationError', {
'message': fields.String(required=True, description='Validation error message'),
'errors': fields.Raw(required=True, description='Field-specific validation errors')
})
@api.route('/products/<int:product_id>')
class Product(Resource):
@api.response(200, 'Success')
@api.response(404, 'Product not found', error_model)
@api.response(400, 'Invalid request', error_model)
@api.response(500, 'Internal server error', error_model)
def get(self, product_id):
"""Get a product by ID"""
product = find_product(product_id)
if not product:
abort(404,
message="Product not found",
error_code="PRODUCT_NOT_FOUND",
product_id=product_id)
return product.to_dict()
@api.expect(product_model, validate=True)
@api.response(201, 'Product created')
@api.response(400, 'Validation error', validation_error_model)
def post(self, product_id):
"""Create or update a product"""
# Validation handled automatically
data = api.payload
# Process product...
return {'message': 'Product created'}, 201from flask_restplus import Api
from flask import jsonify
import traceback
import uuid
api = Api()
def format_error_response(message, code=None, details=None, trace_id=None):
"""Format consistent error responses."""
response = {
'success': False,
'message': message,
'timestamp': datetime.utcnow().isoformat(),
'trace_id': trace_id or str(uuid.uuid4())
}
if code:
response['error_code'] = code
if details:
response['details'] = details
return response
@api.errorhandler(Exception)
def handle_error(error):
"""Global error handler with consistent formatting."""
trace_id = str(uuid.uuid4())
# Log error with trace ID for debugging
logging.error(f"Error {trace_id}: {str(error)}")
logging.error(f"Traceback {trace_id}: {traceback.format_exc()}")
if isinstance(error, HTTPException):
return format_error_response(
message=error.description or "HTTP error occurred",
code=f"HTTP_{error.code}",
trace_id=trace_id
), error.code
# Handle other exceptions
return format_error_response(
message="An unexpected error occurred",
code="INTERNAL_ERROR",
details={"type": type(error).__name__},
trace_id=trace_id
), 500
@api.route('/test-error')
class TestError(Resource):
def get(self):
# This will trigger the global error handler
raise ValueError("This is a test error")from flask_restplus import Api, Resource, abort
from flask import g, request
import logging
api = Api()
@api.before_request
def before_request():
"""Set up request context for error handling."""
g.request_id = str(uuid.uuid4())
g.start_time = time.time()
def log_error_context(error, status_code):
"""Log error with request context."""
context = {
'request_id': getattr(g, 'request_id', 'unknown'),
'method': request.method,
'url': request.url,
'user_agent': request.headers.get('User-Agent'),
'remote_addr': request.remote_addr,
'status_code': status_code,
'error': str(error),
'duration_ms': (time.time() - getattr(g, 'start_time', 0)) * 1000
}
logging.error(f"Request failed: {context}")
@api.errorhandler(Exception)
def handle_error_with_context(error):
"""Error handler that logs full request context."""
status_code = 500
if isinstance(error, HTTPException):
status_code = error.code
log_error_context(error, status_code)
return {
'message': 'An error occurred',
'request_id': getattr(g, 'request_id', 'unknown'),
'status': status_code
}, status_codeInstall with Tessl CLI
npx tessl i tessl/pypi-flask-restplus