RESTful API Design for Microservices
As we build more microservices, API design has become critical. A poorly designed API causes pain for months. A well-designed API makes integration smooth.
I’ve designed 5 microservice APIs in the past 6 months. Here are the patterns that worked (and the mistakes I made).
Table of Contents
REST Principles
I follow REST principles, but pragmatically:
- Resources, not actions -
/users, not/getUsers - HTTP methods - GET, POST, PUT, DELETE
- Status codes - Use them correctly
- Stateless - No server-side sessions
But I’m not dogmatic. If RPC makes more sense, I use it.
URL Structure
Consistent URL patterns:
GET /api/v1/users # List users
GET /api/v1/users/:id # Get user
POST /api/v1/users # Create user
PUT /api/v1/users/:id # Update user
DELETE /api/v1/users/:id # Delete user
# Nested resources
GET /api/v1/users/:id/posts # User's posts
GET /api/v1/users/:id/posts/:pid # Specific post
I always include /api/v1 prefix for:
- Versioning - Can deploy v2 alongside v1
- Routing - Easy to route to API servers
- Clarity - Obvious it’s an API endpoint
Request/Response Format
Always JSON. Always.
Request:
POST /api/v1/users
Content-Type: application/json
{
"name": "John Doe",
"email": "john@example.com",
"role": "admin"
}
Response:
HTTP/1.1 201 Created
Content-Type: application/json
Location: /api/v1/users/123
{
"id": 123,
"name": "John Doe",
"email": "john@example.com",
"role": "admin",
"created_at": "2016-09-30T16:10:00Z"
}
I include:
- id - Always return the ID
- created_at/updated_at - Timestamps are useful
- Location header - For created resources
Pagination
For list endpoints, always paginate:
GET /api/v1/users?page=2&per_page=20
Response:
{
"data": [
{ "id": 21, "name": "User 21" },
{ "id": 22, "name": "User 22" }
],
"pagination": {
"page": 2,
"per_page": 20,
"total": 150,
"total_pages": 8
}
}
I wrap the data in an object (not a bare array) so I can add metadata.
Filtering and Sorting
Support common query operations:
# Filtering
GET /api/v1/users?role=admin
GET /api/v1/users?created_after=2016-01-01
# Sorting
GET /api/v1/users?sort=created_at
GET /api/v1/users?sort=-created_at # Descending
# Field selection
GET /api/v1/users?fields=id,name,email
Implementation in Flask:
@app.route('/api/v1/users')
def list_users():
query = User.query
# Filtering
if request.args.get('role'):
query = query.filter_by(role=request.args['role'])
if request.args.get('created_after'):
date = datetime.fromisoformat(request.args['created_after'])
query = query.filter(User.created_at >= date)
# Sorting
sort = request.args.get('sort', 'id')
if sort.startswith('-'):
query = query.order_by(desc(getattr(User, sort[1:])))
else:
query = query.order_by(getattr(User, sort))
# Pagination
page = int(request.args.get('page', 1))
per_page = int(request.args.get('per_page', 20))
paginated = query.paginate(page, per_page)
return jsonify({
'data': [user.to_dict() for user in paginated.items],
'pagination': {
'page': page,
'per_page': per_page,
'total': paginated.total,
'total_pages': paginated.pages
}
})
Error Handling
Consistent error format:
{
"error": {
"code": "VALIDATION_ERROR",
"message": "Invalid input",
"details": [
{
"field": "email",
"message": "Email is required"
},
{
"field": "password",
"message": "Password must be at least 8 characters"
}
]
}
}
HTTP status codes I use:
- 200 OK - Success
- 201 Created - Resource created
- 204 No Content - Success, no response body
- 400 Bad Request - Invalid input
- 401 Unauthorized - Not authenticated
- 403 Forbidden - Authenticated but not authorized
- 404 Not Found - Resource doesn’t exist
- 409 Conflict - Duplicate resource
- 422 Unprocessable Entity - Validation failed
- 500 Internal Server Error - Server error
Error handler in Flask:
class APIError(Exception):
def __init__(self, message, code='ERROR', status=400, details=None):
self.message = message
self.code = code
self.status = status
self.details = details or []
@app.errorhandler(APIError)
def handle_api_error(error):
response = {
'error': {
'code': error.code,
'message': error.message
}
}
if error.details:
response['error']['details'] = error.details
return jsonify(response), error.status
@app.errorhandler(404)
def not_found(error):
return jsonify({
'error': {
'code': 'NOT_FOUND',
'message': 'Resource not found'
}
}), 404
@app.errorhandler(500)
def internal_error(error):
app.logger.error(f'Internal error: {error}')
return jsonify({
'error': {
'code': 'INTERNAL_ERROR',
'message': 'Internal server error'
}
}), 500
Usage:
@app.route('/api/v1/users', methods=['POST'])
def create_user():
data = request.get_json()
errors = []
if not data.get('email'):
errors.append({'field': 'email', 'message': 'Email is required'})
if not data.get('password'):
errors.append({'field': 'password', 'message': 'Password is required'})
if errors:
raise APIError(
message='Validation failed',
code='VALIDATION_ERROR',
status=422,
details=errors
)
# Create user...
Authentication
I use JWT tokens:
GET /api/v1/users
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...
Implementation:
from functools import wraps
import jwt
def require_auth(f):
@wraps(f)
def decorated(*args, **kwargs):
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if not token:
raise APIError('Missing token', 'UNAUTHORIZED', 401)
try:
payload = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
request.user_id = payload['user_id']
except jwt.ExpiredSignatureError:
raise APIError('Token expired', 'TOKEN_EXPIRED', 401)
except jwt.InvalidTokenError:
raise APIError('Invalid token', 'INVALID_TOKEN', 401)
return f(*args, **kwargs)
return decorated
@app.route('/api/v1/users/<int:id>')
@require_auth
def get_user(id):
# request.user_id is available
user = User.query.get_or_404(id)
return jsonify(user.to_dict())
Versioning
I version APIs from day one. Three strategies:
1. URL versioning (what I use):
/api/v1/users
/api/v2/users
Pros: Simple, explicit Cons: URL changes
2. Header versioning:
GET /api/users
Accept: application/vnd.myapp.v1+json
Pros: Clean URLs Cons: Less visible
3. Query parameter:
GET /api/users?version=1
Pros: Easy to test Cons: Easy to forget
I prefer URL versioning because it’s explicit and easy to route.
Rate Limiting
Prevent abuse with rate limiting:
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"]
)
@app.route('/api/v1/users')
@limiter.limit("10 per minute")
def list_users():
# ...
Response headers:
X-RateLimit-Limit: 10
X-RateLimit-Remaining: 7
X-RateLimit-Reset: 1475251200
When limit exceeded:
HTTP/1.1 429 Too Many Requests
Retry-After: 60
{
"error": {
"code": "RATE_LIMIT_EXCEEDED",
"message": "Rate limit exceeded. Try again in 60 seconds."
}
}
CORS
For browser clients, enable CORS:
from flask_cors import CORS
CORS(app, resources={
r"/api/*": {
"origins": ["https://example.com"],
"methods": ["GET", "POST", "PUT", "DELETE"],
"allow_headers": ["Content-Type", "Authorization"]
}
})
Documentation
I use Swagger/OpenAPI for documentation:
from flask_swagger_ui import get_swaggerui_blueprint
SWAGGER_URL = '/api/docs'
API_URL = '/api/swagger.json'
swaggerui_blueprint = get_swaggerui_blueprint(
SWAGGER_URL,
API_URL,
config={'app_name': "User Service API"}
)
app.register_blueprint(swaggerui_blueprint, url_prefix=SWAGGER_URL)
@app.route('/api/swagger.json')
def swagger_spec():
return jsonify({
"swagger": "2.0",
"info": {
"title": "User Service API",
"version": "1.0.0"
},
"paths": {
"/api/v1/users": {
"get": {
"summary": "List users",
"parameters": [
{
"name": "page",
"in": "query",
"type": "integer"
}
],
"responses": {
"200": {
"description": "Success"
}
}
}
}
}
})
Now developers can browse the API at /api/docs.
Health Checks
Every service needs health checks:
@app.route('/health')
def health():
return jsonify({'status': 'healthy'})
@app.route('/health/detailed')
def health_detailed():
checks = {
'database': check_database(),
'redis': check_redis(),
'disk_space': check_disk_space()
}
all_healthy = all(checks.values())
status_code = 200 if all_healthy else 503
return jsonify({
'status': 'healthy' if all_healthy else 'unhealthy',
'checks': checks
}), status_code
def check_database():
try:
db.session.execute('SELECT 1')
return True
except:
return False
Load balancers use /health to determine if a service is up.
Idempotency
For safety, make PUT and DELETE idempotent:
@app.route('/api/v1/users/<int:id>', methods=['PUT'])
def update_user(id):
user = User.query.get_or_404(id)
data = request.get_json()
# Update fields
user.name = data.get('name', user.name)
user.email = data.get('email', user.email)
db.session.commit()
return jsonify(user.to_dict())
# Calling this multiple times with same data has same effect
For POST (non-idempotent), use idempotency keys:
@app.route('/api/v1/users', methods=['POST'])
def create_user():
idempotency_key = request.headers.get('Idempotency-Key')
if idempotency_key:
# Check if we've seen this key before
existing = IdempotencyLog.query.filter_by(key=idempotency_key).first()
if existing:
return jsonify(existing.response), existing.status_code
# Create user...
user = User(...)
db.session.add(user)
db.session.commit()
# Store idempotency log
if idempotency_key:
log = IdempotencyLog(
key=idempotency_key,
response=user.to_dict(),
status_code=201
)
db.session.add(log)
db.session.commit()
return jsonify(user.to_dict()), 201
Testing
I write integration tests for all endpoints:
import unittest
from app import app, db
class APITestCase(unittest.TestCase):
def setUp(self):
app.config['TESTING'] = True
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
self.client = app.test_client()
with app.app_context():
db.create_all()
def test_create_user(self):
response = self.client.post('/api/v1/users', json={
'name': 'Test User',
'email': 'test@example.com'
})
self.assertEqual(response.status_code, 201)
data = response.get_json()
self.assertEqual(data['name'], 'Test User')
self.assertIn('id', data)
def test_create_user_validation(self):
response = self.client.post('/api/v1/users', json={})
self.assertEqual(response.status_code, 422)
data = response.get_json()
self.assertIn('error', data)
Lessons Learned
What worked:
- Consistent patterns - Same structure across all services
- Good error messages - Saves debugging time
- Comprehensive docs - Swagger makes integration easy
- Versioning from day one - Easier to evolve APIs
Mistakes I made:
- No pagination initially - Had to add later, breaking change
- Inconsistent error codes - Different services used different formats
- No rate limiting - Got abused by a buggy client
- Poor field naming -
user_idvsuserIdinconsistency
Conclusion
Good API design is hard but worth the effort. A well-designed API is a joy to use. A poorly designed API causes endless frustration.
Key principles:
- Be consistent
- Use HTTP correctly
- Provide good error messages
- Document everything
- Version from day one
- Make it easy to test
These patterns have served us well across 5 microservices. They’ll evolve as we learn more, but the fundamentals are solid.
Next challenge: GraphQL. I’m curious if it’s better than REST for our use case.