HTTP Server with Flask

In this chapter, we'll create our first HTTP server using Flask. We'll learn:

• Introduction to Flask
• Creating your first HTTP endpoint
• Testing HTTP handlers
• Request and response handling
• Routing basics

Introduction to Flask

Flask is a lightweight web framework for Python. It's perfect for building APIs and web applications:

pip install flask pytest

Our first endpoint

Let's start with a simple health check endpoint.

Write the test first

Create tests/test_api.py:

import pytest
from app import create_app


@pytest.fixture
def client():
    app = create_app()
    app.config['TESTING'] = True
    with app.test_client() as client:
        yield client


def test_health_check(client):
    response = client.get('/health')
    
    assert response.status_code == 200
    assert response.json == {'status': 'healthy'}

Run the test

pytest tests/test_api.py

It fails because we haven't created the app yet.

Create the application

Create app/__init__.py:

from flask import Flask


def create_app():
    app = Flask(__name__)
    
    @app.route('/health')
    def health():
        return {'status': 'healthy'}
    
    return app

Run the tests - they pass!

Understanding the test

Let's break down what's happening:

1. create_app(): Factory function that creates Flask app
2. app.test_client(): Creates a test client for making requests
3. client.get('/health'): Makes a GET request to /health
4. response.json: Parses JSON response body

Adding tasks endpoint

User requirement

As a user, I want to see a list of all tasks so I can know what needs to be done.

Write the test

def test_get_tasks_returns_empty_list(client):
    response = client.get('/api/tasks')
    
    assert response.status_code == 200
    assert response.json == []

Make it pass

Update app/__init__.py:

from flask import Flask


def create_app():
    app = Flask(__name__)
    
    tasks = []
    
    @app.route('/health')
    def health():
        return {'status': 'healthy'}
    
    @app.route('/api/tasks')
    def get_tasks():
        return tasks
    
    return app

Creating tasks

User requirement

As a user, I want to create a new task so I can track what needs to be done.

Write the test

def test_create_task(client):
    response = client.post('/api/tasks', json={
        'title': 'Learn Flask',
        'description': 'Build a REST API with TDD'
    })
    
    assert response.status_code == 201
    data = response.json
    assert data['title'] == 'Learn Flask'
    assert data['description'] == 'Build a REST API with TDD'
    assert 'id' in data


def test_created_task_appears_in_list(client):
    # Create a task
    client.post('/api/tasks', json={
        'title': 'Test Task'
    })
    
    # Get all tasks
    response = client.get('/api/tasks')
    
    assert len(response.json) == 1
    assert response.json[0]['title'] == 'Test Task'

Make it pass

from flask import Flask, request
import uuid


def create_app():
    app = Flask(__name__)
    
    tasks = []
    
    @app.route('/health')
    def health():
        return {'status': 'healthy'}
    
    @app.route('/api/tasks')
    def get_tasks():
        return tasks
    
    @app.route('/api/tasks', methods=['POST'])
    def create_task():
        data = request.json
        task = {
            'id': str(uuid.uuid4()),
            'title': data.get('title'),
            'description': data.get('description', ''),
            'completed': False
        }
        tasks.append(task)
        return task, 201
    
    return app

Problem: Shared state between tests

Our tests are failing intermittently! The tasks list persists between tests.

Fix with a fixture

@pytest.fixture
def client():
    app = create_app()
    app.config['TESTING'] = True
    
    # Clear tasks before each test
    with app.test_client() as client:
        # Reset state
        client.delete('/api/tasks/reset')  # We'll implement this
        yield client

Or better, refactor to use dependency injection:

# app/__init__.py
from flask import Flask, request
import uuid


def create_app(task_store=None):
    app = Flask(__name__)
    
    if task_store is None:
        task_store = []
    
    @app.route('/health')
    def health():
        return {'status': 'healthy'}
    
    @app.route('/api/tasks')
    def get_tasks():
        return task_store
    
    @app.route('/api/tasks', methods=['POST'])
    def create_task():
        data = request.json
        task = {
            'id': str(uuid.uuid4()),
            'title': data.get('title'),
            'description': data.get('description', ''),
            'completed': False
        }
        task_store.append(task)
        return task, 201
    
    return app

Update the test fixture:

@pytest.fixture
def client():
    # Each test gets a fresh task store
    app = create_app(task_store=[])
    app.config['TESTING'] = True
    with app.test_client() as client:
        yield client

Getting a single task

Write the test

def test_get_single_task(client):
    # Create a task
    create_response = client.post('/api/tasks', json={
        'title': 'My Task'
    })
    task_id = create_response.json['id']
    
    # Get the task
    response = client.get(f'/api/tasks/{task_id}')
    
    assert response.status_code == 200
    assert response.json['title'] == 'My Task'


def test_get_nonexistent_task(client):
    response = client.get('/api/tasks/nonexistent')
    
    assert response.status_code == 404

Make it pass

@app.route('/api/tasks/<task_id>')
def get_task(task_id):
    for task in task_store:
        if task['id'] == task_id:
            return task
    return {'error': 'Task not found'}, 404

Updating tasks

Write the test

def test_update_task(client):
    # Create a task
    create_response = client.post('/api/tasks', json={
        'title': 'Original Title'
    })
    task_id = create_response.json['id']
    
    # Update it
    response = client.put(f'/api/tasks/{task_id}', json={
        'title': 'Updated Title',
        'completed': True
    })
    
    assert response.status_code == 200
    assert response.json['title'] == 'Updated Title'
    assert response.json['completed'] is True

Make it pass

@app.route('/api/tasks/<task_id>', methods=['PUT'])
def update_task(task_id):
    data = request.json
    for task in task_store:
        if task['id'] == task_id:
            task['title'] = data.get('title', task['title'])
            task['description'] = data.get('description', task['description'])
            task['completed'] = data.get('completed', task['completed'])
            return task
    return {'error': 'Task not found'}, 404

Deleting tasks

Write the test

def test_delete_task(client):
    # Create a task
    create_response = client.post('/api/tasks', json={
        'title': 'To Delete'
    })
    task_id = create_response.json['id']
    
    # Delete it
    response = client.delete(f'/api/tasks/{task_id}')
    assert response.status_code == 204
    
    # Verify it's gone
    get_response = client.get(f'/api/tasks/{task_id}')
    assert get_response.status_code == 404

Make it pass

@app.route('/api/tasks/<task_id>', methods=['DELETE'])
def delete_task(task_id):
    for i, task in enumerate(task_store):
        if task['id'] == task_id:
            task_store.pop(i)
            return '', 204
    return {'error': 'Task not found'}, 404

Refactoring to a service layer

Our app is growing. Let's separate concerns:

Create services/task_service.py

import uuid


class TaskService:
    def __init__(self, store=None):
        self.store = store if store is not None else []
    
    def get_all(self):
        return self.store
    
    def get_by_id(self, task_id):
        for task in self.store:
            if task['id'] == task_id:
                return task
        return None
    
    def create(self, title, description=''):
        task = {
            'id': str(uuid.uuid4()),
            'title': title,
            'description': description,
            'completed': False
        }
        self.store.append(task)
        return task
    
    def update(self, task_id, **updates):
        task = self.get_by_id(task_id)
        if task:
            for key, value in updates.items():
                if key in task:
                    task[key] = value
        return task
    
    def delete(self, task_id):
        for i, task in enumerate(self.store):
            if task['id'] == task_id:
                self.store.pop(i)
                return True
        return False

Test the service

# tests/test_task_service.py
import pytest
from app.services.task_service import TaskService


@pytest.fixture
def service():
    return TaskService()


def test_create_task(service):
    task = service.create('Test Task', 'Description')
    
    assert task['title'] == 'Test Task'
    assert task['description'] == 'Description'
    assert task['completed'] is False
    assert 'id' in task


def test_get_all_tasks(service):
    service.create('Task 1')
    service.create('Task 2')
    
    tasks = service.get_all()
    
    assert len(tasks) == 2


def test_get_task_by_id(service):
    created = service.create('My Task')
    
    task = service.get_by_id(created['id'])
    
    assert task['title'] == 'My Task'


def test_update_task(service):
    created = service.create('Original')
    
    updated = service.update(created['id'], title='Updated', completed=True)
    
    assert updated['title'] == 'Updated'
    assert updated['completed'] is True

Update the app to use the service

from flask import Flask, request
from app.services.task_service import TaskService


def create_app(task_service=None):
    app = Flask(__name__)
    
    if task_service is None:
        task_service = TaskService()
    
    @app.route('/health')
    def health():
        return {'status': 'healthy'}
    
    @app.route('/api/tasks')
    def get_tasks():
        return task_service.get_all()
    
    @app.route('/api/tasks', methods=['POST'])
    def create_task():
        data = request.json
        task = task_service.create(
            data.get('title'),
            data.get('description', '')
        )
        return task, 201
    
    @app.route('/api/tasks/<task_id>')
    def get_task(task_id):
        task = task_service.get_by_id(task_id)
        if task:
            return task
        return {'error': 'Task not found'}, 404
    
    @app.route('/api/tasks/<task_id>', methods=['PUT'])
    def update_task(task_id):
        data = request.json
        task = task_service.update(task_id, **data)
        if task:
            return task
        return {'error': 'Task not found'}, 404
    
    @app.route('/api/tasks/<task_id>', methods=['DELETE'])
    def delete_task(task_id):
        if task_service.delete(task_id):
            return '', 204
        return {'error': 'Task not found'}, 404
    
    return app

Running the server

Create run.py:

from app import create_app

app = create_app()

if __name__ == '__main__':
    app.run(debug=True)

Run it:

python run.py

Test with curl:

# Create a task
curl -X POST http://localhost:5000/api/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "Learn Flask"}'

# Get all tasks
curl http://localhost:5000/api/tasks

Wrapping up

We've covered:

• Flask basics - Creating an application with Flask(__name__)
• Routes - Mapping URLs to functions with @app.route
• HTTP methods - GET, POST, PUT, DELETE
• Testing - Using Flask's test client
• JSON handling - request.json and returning dicts
• Service layer - Separating business logic from HTTP handling

Key takeaways

1. Write tests first, then implement
2. Use the test client for HTTP testing
3. Inject dependencies for testability
4. Separate concerns (routes vs business logic)
5. Each endpoint should be simple and focused

In the next chapter, we'll expand our JSON handling and add more sophisticated routing.