JSON and Routing

In this chapter, we'll expand our Task API with better JSON handling and advanced routing. We'll cover:

• Returning JSON responses properly
• Advanced routing with Flask
• Request parameters and query strings
• Request body parsing and validation
• Error handling in APIs
• Testing JSON endpoints

Proper JSON responses

In the previous chapter, Flask automatically converted dicts to JSON. Let's be more explicit:

from flask import jsonify


@app.route('/api/tasks')
def get_tasks():
    tasks = task_service.get_all()
    return jsonify(tasks)

jsonify provides:

• Proper Content-Type: application/json header
• Correct encoding of Python objects

Testing JSON responses

def test_response_has_json_content_type(client):
    response = client.get('/api/tasks')
    
    assert response.content_type == 'application/json'

Request validation

User requirement

As a user, I want clear error messages when I submit invalid data.

Write the test

def test_create_task_requires_title(client):
    response = client.post('/api/tasks', json={
        'description': 'Missing title'
    })
    
    assert response.status_code == 400
    assert 'title' in response.json['error'].lower()


def test_create_task_rejects_empty_title(client):
    response = client.post('/api/tasks', json={
        'title': ''
    })
    
    assert response.status_code == 400
    assert 'title' in response.json['error'].lower()

Implement validation

@app.route('/api/tasks', methods=['POST'])
def create_task():
    data = request.json
    
    if not data:
        return jsonify({'error': 'Request body is required'}), 400
    
    title = data.get('title', '').strip()
    if not title:
        return jsonify({'error': 'Title is required'}), 400
    
    task = task_service.create(title, data.get('description', ''))
    return jsonify(task), 201

Query parameters

User requirement

As a user, I want to filter tasks by completion status.

Write the test

def test_filter_tasks_by_completion(client):
    # Create tasks
    client.post('/api/tasks', json={'title': 'Task 1'})
    client.post('/api/tasks', json={'title': 'Task 2'})
    
    # Complete one task
    response = client.get('/api/tasks')
    task_id = response.json[0]['id']
    client.put(f'/api/tasks/{task_id}', json={'completed': True})
    
    # Filter by completed
    completed = client.get('/api/tasks?completed=true')
    assert len(completed.json) == 1
    
    incomplete = client.get('/api/tasks?completed=false')
    assert len(incomplete.json) == 1


def test_get_all_tasks_without_filter(client):
    client.post('/api/tasks', json={'title': 'Task 1'})
    client.post('/api/tasks', json={'title': 'Task 2'})
    
    response = client.get('/api/tasks')
    
    assert len(response.json) == 2

Implement filtering

from flask import request


@app.route('/api/tasks')
def get_tasks():
    tasks = task_service.get_all()
    
    completed_filter = request.args.get('completed')
    if completed_filter is not None:
        is_completed = completed_filter.lower() == 'true'
        tasks = [t for t in tasks if t['completed'] == is_completed]
    
    return jsonify(tasks)

URL path parameters

Nested resources

As a user, I want to add notes to tasks.

Write the tests

def test_add_note_to_task(client):
    # Create a task
    task_response = client.post('/api/tasks', json={'title': 'My Task'})
    task_id = task_response.json['id']
    
    # Add a note
    response = client.post(f'/api/tasks/{task_id}/notes', json={
        'content': 'This is a note'
    })
    
    assert response.status_code == 201
    assert response.json['content'] == 'This is a note'


def test_get_task_notes(client):
    # Create a task with notes
    task_response = client.post('/api/tasks', json={'title': 'My Task'})
    task_id = task_response.json['id']
    client.post(f'/api/tasks/{task_id}/notes', json={'content': 'Note 1'})
    client.post(f'/api/tasks/{task_id}/notes', json={'content': 'Note 2'})
    
    # Get notes
    response = client.get(f'/api/tasks/{task_id}/notes')
    
    assert response.status_code == 200
    assert len(response.json) == 2

Implement nested routes

First, update the TaskService:

class TaskService:
    def __init__(self, store=None):
        self.store = store if store is not None else []
    
    def create(self, title, description=''):
        task = {
            'id': str(uuid.uuid4()),
            'title': title,
            'description': description,
            'completed': False,
            'notes': []  # Add notes list
        }
        self.store.append(task)
        return task
    
    def add_note(self, task_id, content):
        task = self.get_by_id(task_id)
        if task:
            note = {
                'id': str(uuid.uuid4()),
                'content': content,
                'created_at': datetime.utcnow().isoformat()
            }
            task['notes'].append(note)
            return note
        return None
    
    def get_notes(self, task_id):
        task = self.get_by_id(task_id)
        if task:
            return task['notes']
        return None

Add the routes:

@app.route('/api/tasks/<task_id>/notes', methods=['POST'])
def add_note(task_id):
    data = request.json
    content = data.get('content', '').strip()
    
    if not content:
        return jsonify({'error': 'Content is required'}), 400
    
    note = task_service.add_note(task_id, content)
    if note:
        return jsonify(note), 201
    return jsonify({'error': 'Task not found'}), 404


@app.route('/api/tasks/<task_id>/notes')
def get_notes(task_id):
    notes = task_service.get_notes(task_id)
    if notes is not None:
        return jsonify(notes)
    return jsonify({'error': 'Task not found'}), 404

URL converters

Flask provides URL converters for type validation:

# Integer parameter
@app.route('/api/items/<int:item_id>')
def get_item(item_id):  # item_id is already an int
    pass

# Float parameter
@app.route('/api/coordinates/<float:lat>/<float:lng>')
def get_location(lat, lng):
    pass

# Path parameter (can include slashes)
@app.route('/api/files/<path:filepath>')
def get_file(filepath):
    pass

Error handling

Global error handlers

from flask import Flask, jsonify


def create_app(task_service=None):
    app = Flask(__name__)
    
    @app.errorhandler(400)
    def bad_request(error):
        return jsonify({'error': 'Bad request'}), 400
    
    @app.errorhandler(404)
    def not_found(error):
        return jsonify({'error': 'Not found'}), 404
    
    @app.errorhandler(500)
    def internal_error(error):
        return jsonify({'error': 'Internal server error'}), 500
    
    # ... rest of app

Testing error handlers

def test_404_returns_json(client):
    response = client.get('/nonexistent')
    
    assert response.status_code == 404
    assert response.content_type == 'application/json'
    assert 'error' in response.json

Custom exceptions

class TaskNotFoundError(Exception):
    pass


class ValidationError(Exception):
    def __init__(self, message):
        self.message = message


@app.errorhandler(TaskNotFoundError)
def handle_task_not_found(error):
    return jsonify({'error': 'Task not found'}), 404


@app.errorhandler(ValidationError)
def handle_validation_error(error):
    return jsonify({'error': error.message}), 400

Request body validation with schemas

For complex validation, use a schema:

def validate_task(data):
    errors = []
    
    if not data:
        return ['Request body is required']
    
    title = data.get('title', '').strip()
    if not title:
        errors.append('Title is required')
    elif len(title) > 200:
        errors.append('Title must be 200 characters or less')
    
    description = data.get('description', '')
    if len(description) > 1000:
        errors.append('Description must be 1000 characters or less')
    
    return errors


@app.route('/api/tasks', methods=['POST'])
def create_task():
    data = request.json
    errors = validate_task(data)
    
    if errors:
        return jsonify({'errors': errors}), 400
    
    task = task_service.create(
        data['title'].strip(),
        data.get('description', '')
    )
    return jsonify(task), 201

Testing validation

def test_task_title_max_length(client):
    long_title = 'a' * 201
    response = client.post('/api/tasks', json={'title': long_title})
    
    assert response.status_code == 400
    assert 'title' in str(response.json['errors']).lower()


def test_returns_multiple_errors(client):
    response = client.post('/api/tasks', json={
        'title': '',
        'description': 'x' * 1001
    })
    
    assert response.status_code == 400
    assert len(response.json['errors']) == 2

Pagination

User requirement

As a user, I want to paginate through large lists of tasks.

Write the test

def test_pagination(client):
    # Create 25 tasks
    for i in range(25):
        client.post('/api/tasks', json={'title': f'Task {i}'})
    
    # Get first page
    page1 = client.get('/api/tasks?page=1&per_page=10')
    assert len(page1.json['items']) == 10
    assert page1.json['total'] == 25
    assert page1.json['page'] == 1
    assert page1.json['pages'] == 3
    
    # Get second page
    page2 = client.get('/api/tasks?page=2&per_page=10')
    assert len(page2.json['items']) == 10
    
    # Get third page
    page3 = client.get('/api/tasks?page=3&per_page=10')
    assert len(page3.json['items']) == 5

Implement pagination

@app.route('/api/tasks')
def get_tasks():
    tasks = task_service.get_all()
    
    # Apply filters
    completed_filter = request.args.get('completed')
    if completed_filter is not None:
        is_completed = completed_filter.lower() == 'true'
        tasks = [t for t in tasks if t['completed'] == is_completed]
    
    # Pagination
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    per_page = min(per_page, 100)  # Max 100 per page
    
    total = len(tasks)
    start = (page - 1) * per_page
    end = start + per_page
    
    return jsonify({
        'items': tasks[start:end],
        'total': total,
        'page': page,
        'per_page': per_page,
        'pages': (total + per_page - 1) // per_page
    })

Sorting

Write the test

def test_sort_tasks_by_title(client):
    client.post('/api/tasks', json={'title': 'Zebra'})
    client.post('/api/tasks', json={'title': 'Apple'})
    client.post('/api/tasks', json={'title': 'Mango'})
    
    response = client.get('/api/tasks?sort=title')
    titles = [t['title'] for t in response.json['items']]
    
    assert titles == ['Apple', 'Mango', 'Zebra']


def test_sort_tasks_descending(client):
    client.post('/api/tasks', json={'title': 'Zebra'})
    client.post('/api/tasks', json={'title': 'Apple'})
    
    response = client.get('/api/tasks?sort=title&order=desc')
    titles = [t['title'] for t in response.json['items']]
    
    assert titles == ['Zebra', 'Apple']

Implement sorting

@app.route('/api/tasks')
def get_tasks():
    tasks = task_service.get_all()
    
    # Sorting
    sort_by = request.args.get('sort')
    if sort_by in ['title', 'created_at']:
        reverse = request.args.get('order', 'asc') == 'desc'
        tasks = sorted(tasks, key=lambda t: t.get(sort_by, ''), reverse=reverse)
    
    # ... rest of filtering and pagination

Wrapping up

We've covered:

• JSON responses - Use jsonify() for proper content types
• Request validation - Validate input before processing
• Query parameters - Use request.args for filtering
• Path parameters - URL patterns like /<task_id>
• Nested resources - Routes like /tasks/<id>/notes
• Error handling - Global error handlers for consistent responses
• Pagination - Return paginated results with metadata
• Sorting - Allow clients to sort results

Key takeaways

1. Validate all input and return clear error messages
2. Use consistent response formats
3. Handle errors gracefully with proper status codes
4. Support filtering, pagination, and sorting for lists
5. Test edge cases and error conditions

In the next chapter, we'll persist our data with database integration.