}

Flask API POST Request: Complete Guide to Handling POST Data

22 min read

Introduction

POST requests are fundamental to web APIs—they allow clients to send data to your server for processing, storage, or any operation that modifies state. In this Flask API tutorial, you'll learn everything about handling POST requests, from basic JSON handling to building a complete REST API.

By the end of this tutorial, you'll know how to: - Create POST endpoints with @app.route - Access different types of POST data (JSON, form data, files) - Return proper JSON responses with status codes - Validate input and handle errors gracefully - Test your endpoints with curl, Postman, and Python requests

Prerequisites

  • Python 3.8 or higher installed
  • Basic understanding of Python
  • Familiarity with HTTP concepts (optional but helpful)

Setting Up a Basic Flask App

First, let's create a new project and install Flask:

# Create project directory
mkdir flask-post-tutorial
cd flask-post-tutorial

# Create and activate virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install Flask
pip install flask

Create a file named app.py with a minimal Flask application:

from flask import Flask

app = Flask(__name__)

@app.route('/')
def index():
    return {'message': 'Flask API is running!'}

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

Run the application:

python app.py

Your API is now running at http://127.0.0.1:5000.

Creating POST Endpoints with @app.route

To create a POST endpoint, use the methods parameter in the @app.route decorator:

from flask import Flask, request

app = Flask(__name__)

@app.route('/api/data', methods=['POST'])
def receive_data():
    return {'status': 'Data received'}, 201

You can also allow multiple HTTP methods on a single endpoint:

@app.route('/api/items', methods=['GET', 'POST'])
def items():
    if request.method == 'POST':
        return {'action': 'Creating item'}, 201
    return {'action': 'Listing items'}

Accessing POST Data

Flask provides several ways to access POST data through the request object. Let's explore each one.

Receiving JSON Data (request.json)

The most common way to receive data in modern APIs is JSON. Use request.json to parse JSON bodies:

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/api/users', methods=['POST'])
def create_user():
    # Get JSON data from request body
    data = request.json

    if not data:
        return jsonify({'error': 'No JSON data provided'}), 400

    # Access individual fields
    name = data.get('name')
    email = data.get('email')

    return jsonify({
        'message': 'User created',
        'user': {
            'name': name,
            'email': email
        }
    }), 201

Important: The client must send the Content-Type: application/json header for request.json to work.

For more control over JSON parsing, use request.get_json():

@app.route('/api/data', methods=['POST'])
def receive_json():
    # force=True parses JSON even without correct Content-Type
    # silent=True returns None instead of raising an error on invalid JSON
    data = request.get_json(force=False, silent=True)

    if data is None:
        return jsonify({'error': 'Invalid or missing JSON'}), 400

    return jsonify({'received': data}), 200

Receiving Form Data (request.form)

For HTML form submissions or application/x-www-form-urlencoded data, use request.form:

@app.route('/api/login', methods=['POST'])
def login():
    # Access form fields
    username = request.form.get('username')
    password = request.form.get('password')

    if not username or not password:
        return jsonify({'error': 'Username and password required'}), 400

    # Validate credentials (simplified example)
    if username == 'admin' and password == 'secret':
        return jsonify({'message': 'Login successful'}), 200

    return jsonify({'error': 'Invalid credentials'}), 401

You can also access form data like a dictionary:

@app.route('/api/form', methods=['POST'])
def handle_form():
    # Get all form data as a dictionary
    form_data = request.form.to_dict()

    # Access with default values
    name = request.form.get('name', 'Anonymous')

    # Access lists (for multi-select fields)
    tags = request.form.getlist('tags')

    return jsonify({
        'form_data': form_data,
        'name': name,
        'tags': tags
    })

Receiving Raw Data (request.data)

For raw request bodies (XML, plain text, custom formats), use request.data:

@app.route('/api/raw', methods=['POST'])
def receive_raw():
    # Get raw bytes
    raw_data = request.data

    # Decode to string if needed
    text_data = raw_data.decode('utf-8')

    return jsonify({
        'received_bytes': len(raw_data),
        'content': text_data
    })

For specific content types, you can also use:

@app.route('/api/xml', methods=['POST'])
def receive_xml():
    # Check content type
    if request.content_type != 'application/xml':
        return jsonify({'error': 'Expected XML content'}), 415

    xml_data = request.data.decode('utf-8')
    # Parse XML here...

    return jsonify({'message': 'XML received'})

Handling File Uploads (request.files)

Flask makes file uploads straightforward with request.files:

import os
from flask import Flask, request, jsonify
from werkzeug.utils import secure_filename

app = Flask(__name__)

# Configure upload settings
UPLOAD_FOLDER = 'uploads'
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'pdf'}
MAX_CONTENT_LENGTH = 16 * 1024 * 1024  # 16 MB max

app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
app.config['MAX_CONTENT_LENGTH'] = MAX_CONTENT_LENGTH

# Create upload directory if it doesn't exist
os.makedirs(UPLOAD_FOLDER, exist_ok=True)

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

@app.route('/api/upload', methods=['POST'])
def upload_file():
    # Check if file was included in request
    if 'file' not in request.files:
        return jsonify({'error': 'No file part in request'}), 400

    file = request.files['file']

    # Check if file was selected
    if file.filename == '':
        return jsonify({'error': 'No file selected'}), 400

    # Validate and save file
    if file and allowed_file(file.filename):
        # Secure the filename to prevent directory traversal attacks
        filename = secure_filename(file.filename)
        filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
        file.save(filepath)

        return jsonify({
            'message': 'File uploaded successfully',
            'filename': filename,
            'size': os.path.getsize(filepath)
        }), 201

    return jsonify({'error': 'File type not allowed'}), 400

Handle multiple file uploads:

@app.route('/api/upload-multiple', methods=['POST'])
def upload_multiple():
    if 'files' not in request.files:
        return jsonify({'error': 'No files in request'}), 400

    files = request.files.getlist('files')
    uploaded = []

    for file in files:
        if file and allowed_file(file.filename):
            filename = secure_filename(file.filename)
            filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
            file.save(filepath)
            uploaded.append(filename)

    return jsonify({
        'message': f'{len(uploaded)} files uploaded',
        'files': uploaded
    }), 201

Returning JSON Responses

Flask provides multiple ways to return JSON responses:

from flask import Flask, jsonify, make_response

app = Flask(__name__)

@app.route('/api/example1', methods=['POST'])
def example1():
    # Method 1: Return a dictionary (Flask auto-converts to JSON)
    return {'message': 'Hello'}, 201

@app.route('/api/example2', methods=['POST'])
def example2():
    # Method 2: Use jsonify for more control
    return jsonify(message='Hello', status='success'), 201

@app.route('/api/example3', methods=['POST'])
def example3():
    # Method 3: Use make_response for full control
    response = make_response(jsonify({'message': 'Hello'}))
    response.status_code = 201
    response.headers['X-Custom-Header'] = 'Custom Value'
    return response

Error Handling and Status Codes

Proper error handling is crucial for a good API. Here's how to implement it:

from flask import Flask, request, jsonify
from werkzeug.exceptions import HTTPException

app = Flask(__name__)

# Custom error handler for all HTTP exceptions
@app.errorhandler(HTTPException)
def handle_http_exception(e):
    return jsonify({
        'error': e.name,
        'message': e.description
    }), e.code

# Handle specific errors
@app.errorhandler(400)
def bad_request(e):
    return jsonify({
        'error': 'Bad Request',
        'message': str(e.description)
    }), 400

@app.errorhandler(404)
def not_found(e):
    return jsonify({
        'error': 'Not Found',
        'message': 'The requested resource was not found'
    }), 404

@app.errorhandler(500)
def internal_error(e):
    return jsonify({
        'error': 'Internal Server Error',
        'message': 'An unexpected error occurred'
    }), 500

# Example endpoint with manual error handling
@app.route('/api/items', methods=['POST'])
def create_item():
    data = request.get_json(silent=True)

    if not data:
        return jsonify({'error': 'Invalid JSON'}), 400

    if 'name' not in data:
        return jsonify({'error': 'Missing required field: name'}), 400

    # Process the data...
    return jsonify({'item': data}), 201

Common HTTP status codes for POST requests:

Code Meaning When to Use
200 OK Successful update/processing
201 Created New resource created
400 Bad Request Invalid input data
401 Unauthorized Authentication required
403 Forbidden Not allowed
404 Not Found Resource doesn't exist
409 Conflict Resource already exists
415 Unsupported Media Type Wrong content type
422 Unprocessable Entity Validation failed
500 Internal Server Error Server-side error

Input Validation

Always validate input data before processing. Here are several approaches:

Manual Validation

@app.route('/api/users', methods=['POST'])
def create_user():
    data = request.get_json(silent=True)
    errors = []

    if not data:
        return jsonify({'error': 'JSON body required'}), 400

    # Validate required fields
    if not data.get('name'):
        errors.append('Name is required')
    elif len(data['name']) < 2:
        errors.append('Name must be at least 2 characters')

    if not data.get('email'):
        errors.append('Email is required')
    elif '@' not in data['email']:
        errors.append('Invalid email format')

    if data.get('age') is not None:
        if not isinstance(data['age'], int) or data['age'] < 0:
            errors.append('Age must be a positive integer')

    if errors:
        return jsonify({'errors': errors}), 400

    # Process valid data...
    return jsonify({'user': data}), 201

Using a Validation Library (marshmallow)

For complex APIs, use a validation library like marshmallow:

pip install marshmallow

from flask import Flask, request, jsonify
from marshmallow import Schema, fields, validate, ValidationError

app = Flask(__name__)

# Define a schema
class UserSchema(Schema):
    name = fields.Str(required=True, validate=validate.Length(min=2, max=100))
    email = fields.Email(required=True)
    age = fields.Int(validate=validate.Range(min=0, max=150))
    role = fields.Str(validate=validate.OneOf(['admin', 'user', 'guest']))

user_schema = UserSchema()

@app.route('/api/users', methods=['POST'])
def create_user():
    data = request.get_json(silent=True)

    if not data:
        return jsonify({'error': 'JSON body required'}), 400

    try:
        # Validate and deserialize
        validated_data = user_schema.load(data)
    except ValidationError as err:
        return jsonify({'errors': err.messages}), 400

    # Process validated data...
    return jsonify({'user': validated_data}), 201

Testing POST Endpoints

Let's explore different ways to test your Flask POST endpoints.

Testing with curl

# Send JSON data
curl -X POST http://127.0.0.1:5000/api/users \
  -H "Content-Type: application/json" \
  -d '{"name": "John", "email": "[email protected]"}'

# Send form data
curl -X POST http://127.0.0.1:5000/api/login \
  -d "username=admin&password=secret"

# Upload a file
curl -X POST http://127.0.0.1:5000/api/upload \
  -F "file=@/path/to/image.jpg"

# Upload multiple files
curl -X POST http://127.0.0.1:5000/api/upload-multiple \
  -F "[email protected]" \
  -F "[email protected]"

Testing with Postman

  1. Open Postman and create a new request
  2. Set the method to POST
  3. Enter the URL (e.g., http://127.0.0.1:5000/api/users)
  4. Go to the "Body" tab
  5. Select "raw" and choose "JSON" from the dropdown
  6. Enter your JSON data:
{
    "name": "John Doe",
    "email": "[email protected]"
}
  1. Click "Send"

For file uploads, select "form-data" and add a file field.

Testing with Python requests

import requests

BASE_URL = 'http://127.0.0.1:5000'

# Send JSON data
response = requests.post(
    f'{BASE_URL}/api/users',
    json={'name': 'John', 'email': '[email protected]'}
)
print(response.status_code)
print(response.json())

# Send form data
response = requests.post(
    f'{BASE_URL}/api/login',
    data={'username': 'admin', 'password': 'secret'}
)
print(response.json())

# Upload a file
with open('image.jpg', 'rb') as f:
    response = requests.post(
        f'{BASE_URL}/api/upload',
        files={'file': f}
    )
print(response.json())

# Upload multiple files
files = [
    ('files', open('file1.jpg', 'rb')),
    ('files', open('file2.jpg', 'rb'))
]
response = requests.post(f'{BASE_URL}/api/upload-multiple', files=files)
print(response.json())

Writing Unit Tests

import unittest
import json
from app import app

class TestAPI(unittest.TestCase):
    def setUp(self):
        self.app = app.test_client()
        self.app.testing = True

    def test_create_user_success(self):
        response = self.app.post(
            '/api/users',
            data=json.dumps({'name': 'John', 'email': '[email protected]'}),
            content_type='application/json'
        )
        self.assertEqual(response.status_code, 201)
        data = json.loads(response.data)
        self.assertEqual(data['user']['name'], 'John')

    def test_create_user_missing_email(self):
        response = self.app.post(
            '/api/users',
            data=json.dumps({'name': 'John'}),
            content_type='application/json'
        )
        self.assertEqual(response.status_code, 400)

    def test_create_user_invalid_json(self):
        response = self.app.post(
            '/api/users',
            data='invalid json',
            content_type='application/json'
        )
        self.assertEqual(response.status_code, 400)

if __name__ == '__main__':
    unittest.main()

Complete Example: Simple REST API

Here's a complete example of a simple REST API for managing a todo list:

from flask import Flask, request, jsonify
from datetime import datetime
import uuid

app = Flask(__name__)

# In-memory storage (use a database in production)
todos = {}

@app.route('/api/todos', methods=['GET'])
def get_todos():
    """Get all todos"""
    return jsonify({
        'todos': list(todos.values()),
        'count': len(todos)
    })

@app.route('/api/todos', methods=['POST'])
def create_todo():
    """Create a new todo"""
    data = request.get_json(silent=True)

    if not data:
        return jsonify({'error': 'JSON body required'}), 400

    if not data.get('title'):
        return jsonify({'error': 'Title is required'}), 400

    todo_id = str(uuid.uuid4())
    todo = {
        'id': todo_id,
        'title': data['title'],
        'description': data.get('description', ''),
        'completed': False,
        'created_at': datetime.utcnow().isoformat(),
        'updated_at': datetime.utcnow().isoformat()
    }

    todos[todo_id] = todo

    return jsonify({'todo': todo}), 201

@app.route('/api/todos/<todo_id>', methods=['GET'])
def get_todo(todo_id):
    """Get a specific todo"""
    todo = todos.get(todo_id)

    if not todo:
        return jsonify({'error': 'Todo not found'}), 404

    return jsonify({'todo': todo})

@app.route('/api/todos/<todo_id>', methods=['PUT'])
def update_todo(todo_id):
    """Update a todo"""
    todo = todos.get(todo_id)

    if not todo:
        return jsonify({'error': 'Todo not found'}), 404

    data = request.get_json(silent=True)

    if not data:
        return jsonify({'error': 'JSON body required'}), 400

    # Update fields
    if 'title' in data:
        todo['title'] = data['title']
    if 'description' in data:
        todo['description'] = data['description']
    if 'completed' in data:
        todo['completed'] = bool(data['completed'])

    todo['updated_at'] = datetime.utcnow().isoformat()

    return jsonify({'todo': todo})

@app.route('/api/todos/<todo_id>', methods=['DELETE'])
def delete_todo(todo_id):
    """Delete a todo"""
    if todo_id not in todos:
        return jsonify({'error': 'Todo not found'}), 404

    deleted = todos.pop(todo_id)

    return jsonify({'message': 'Todo deleted', 'todo': deleted})

@app.route('/api/todos/<todo_id>/toggle', methods=['POST'])
def toggle_todo(todo_id):
    """Toggle todo completion status"""
    todo = todos.get(todo_id)

    if not todo:
        return jsonify({'error': 'Todo not found'}), 404

    todo['completed'] = not todo['completed']
    todo['updated_at'] = datetime.utcnow().isoformat()

    return jsonify({'todo': todo})

# Error handlers
@app.errorhandler(400)
def bad_request(e):
    return jsonify({'error': 'Bad Request', 'message': str(e)}), 400

@app.errorhandler(404)
def not_found(e):
    return jsonify({'error': 'Not Found'}), 404

@app.errorhandler(500)
def server_error(e):
    return jsonify({'error': 'Internal Server Error'}), 500

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

Test the API:

# Create a todo
curl -X POST http://127.0.0.1:5000/api/todos \
  -H "Content-Type: application/json" \
  -d '{"title": "Learn Flask", "description": "Complete the POST tutorial"}'

# Get all todos
curl http://127.0.0.1:5000/api/todos

# Toggle completion
curl -X POST http://127.0.0.1:5000/api/todos/<id>/toggle

# Update a todo
curl -X PUT http://127.0.0.1:5000/api/todos/<id> \
  -H "Content-Type: application/json" \
  -d '{"title": "Learn Flask API", "completed": true}'

# Delete a todo
curl -X DELETE http://127.0.0.1:5000/api/todos/<id>

Advanced: Domain-Driven Design (DDD) Architecture

As your Flask API grows, maintaining clean code becomes challenging. Domain-Driven Design (DDD) provides a structured approach to organizing your codebase, making it more maintainable, testable, and scalable.

Why DDD for APIs?

Separation of Concerns: Each layer has a single responsibility, making the code easier to understand and modify.

Testability: Business logic is isolated from Flask, allowing you to test domain rules without HTTP concerns.

Flexibility: Swap implementations (e.g., change from in-memory storage to PostgreSQL) without touching business logic.

Team Scalability: Different team members can work on different layers independently.

DDD Project Structure

Let's restructure our Todo API using DDD principles:

app/
├── domain/
│   ├── entities/
│   │   └── todo.py          # Todo entity with business rules
│   ├── value_objects/
│   │   └── todo_status.py   # Status as value object
│   └── repositories/
│       └── todo_repository.py  # Abstract repository interface
├── application/
│   └── use_cases/
│       ├── create_todo.py
│       ├── get_todo.py
│       └── update_todo.py
├── infrastructure/
│   └── repositories/
│       └── memory_todo_repository.py  # Concrete implementation
└── interface/
    └── api/
        └── todo_routes.py   # Flask routes

Domain Layer

The domain layer contains your business logic, independent of any framework.

Value Objects (domain/value_objects/todo_status.py)

Value objects are immutable and defined by their attributes:

from enum import Enum
from dataclasses import dataclass
from typing import Optional

class TodoStatusType(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    CANCELLED = "cancelled"

@dataclass(frozen=True)
class TodoStatus:
    """Value object representing a todo's status."""

    value: TodoStatusType

    @classmethod
    def pending(cls) -> 'TodoStatus':
        return cls(TodoStatusType.PENDING)

    @classmethod
    def in_progress(cls) -> 'TodoStatus':
        return cls(TodoStatusType.IN_PROGRESS)

    @classmethod
    def completed(cls) -> 'TodoStatus':
        return cls(TodoStatusType.COMPLETED)

    @classmethod
    def cancelled(cls) -> 'TodoStatus':
        return cls(TodoStatusType.CANCELLED)

    @classmethod
    def from_string(cls, value: str) -> 'TodoStatus':
        try:
            status_type = TodoStatusType(value)
            return cls(status_type)
        except ValueError:
            raise ValueError(f"Invalid status: {value}")

    def can_transition_to(self, new_status: 'TodoStatus') -> bool:
        """Business rule: define valid status transitions."""
        valid_transitions = {
            TodoStatusType.PENDING: [TodoStatusType.IN_PROGRESS, TodoStatusType.CANCELLED],
            TodoStatusType.IN_PROGRESS: [TodoStatusType.COMPLETED, TodoStatusType.CANCELLED, TodoStatusType.PENDING],
            TodoStatusType.COMPLETED: [],  # Terminal state
            TodoStatusType.CANCELLED: [TodoStatusType.PENDING],  # Can reopen
        }
        return new_status.value in valid_transitions.get(self.value, [])

    def __str__(self) -> str:
        return self.value.value

Entities (domain/entities/todo.py)

Entities have identity and contain business logic:

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import uuid

from domain.value_objects.todo_status import TodoStatus

@dataclass
class Todo:
    """Todo entity with business rules."""

    id: str
    title: str
    description: str
    status: TodoStatus
    created_at: datetime
    updated_at: datetime

    @classmethod
    def create(cls, title: str, description: str = "") -> 'Todo':
        """Factory method to create a new Todo."""
        if not title or not title.strip():
            raise ValueError("Title cannot be empty")

        if len(title) > 200:
            raise ValueError("Title cannot exceed 200 characters")

        now = datetime.utcnow()
        return cls(
            id=str(uuid.uuid4()),
            title=title.strip(),
            description=description.strip(),
            status=TodoStatus.pending(),
            created_at=now,
            updated_at=now
        )

    def update_title(self, title: str) -> None:
        """Update the todo title."""
        if not title or not title.strip():
            raise ValueError("Title cannot be empty")

        if len(title) > 200:
            raise ValueError("Title cannot exceed 200 characters")

        self.title = title.strip()
        self._touch()

    def update_description(self, description: str) -> None:
        """Update the todo description."""
        self.description = description.strip()
        self._touch()

    def change_status(self, new_status: TodoStatus) -> None:
        """Change status with business rule validation."""
        if not self.status.can_transition_to(new_status):
            raise ValueError(
                f"Cannot transition from {self.status} to {new_status}"
            )
        self.status = new_status
        self._touch()

    def complete(self) -> None:
        """Mark todo as completed."""
        self.change_status(TodoStatus.completed())

    def start(self) -> None:
        """Mark todo as in progress."""
        self.change_status(TodoStatus.in_progress())

    def cancel(self) -> None:
        """Cancel the todo."""
        self.change_status(TodoStatus.cancelled())

    def reopen(self) -> None:
        """Reopen a cancelled todo."""
        self.change_status(TodoStatus.pending())

    @property
    def is_completed(self) -> bool:
        return self.status.value.value == "completed"

    def _touch(self) -> None:
        """Update the updated_at timestamp."""
        self.updated_at = datetime.utcnow()

    def to_dict(self) -> dict:
        """Convert to dictionary for serialization."""
        return {
            'id': self.id,
            'title': self.title,
            'description': self.description,
            'status': str(self.status),
            'is_completed': self.is_completed,
            'created_at': self.created_at.isoformat(),
            'updated_at': self.updated_at.isoformat()
        }

Repository Interface (domain/repositories/todo_repository.py)

Define the abstract interface for persistence:

from abc import ABC, abstractmethod
from typing import List, Optional

from domain.entities.todo import Todo

class TodoRepository(ABC):
    """Abstract repository interface for Todo persistence."""

    @abstractmethod
    def save(self, todo: Todo) -> None:
        """Save a todo (create or update)."""
        pass

    @abstractmethod
    def find_by_id(self, todo_id: str) -> Optional[Todo]:
        """Find a todo by its ID."""
        pass

    @abstractmethod
    def find_all(self) -> List[Todo]:
        """Get all todos."""
        pass

    @abstractmethod
    def delete(self, todo_id: str) -> bool:
        """Delete a todo by ID. Returns True if deleted."""
        pass

    @abstractmethod
    def exists(self, todo_id: str) -> bool:
        """Check if a todo exists."""
        pass

Application Layer

The application layer contains use cases that orchestrate domain operations.

Create Todo Use Case (application/use_cases/create_todo.py)

from dataclasses import dataclass
from typing import Optional

from domain.entities.todo import Todo
from domain.repositories.todo_repository import TodoRepository

@dataclass
class CreateTodoRequest:
    """Input data for creating a todo."""
    title: str
    description: str = ""

@dataclass
class CreateTodoResponse:
    """Output data after creating a todo."""
    todo: Todo

class CreateTodoUseCase:
    """Use case for creating a new todo."""

    def __init__(self, repository: TodoRepository):
        self.repository = repository

    def execute(self, request: CreateTodoRequest) -> CreateTodoResponse:
        """Create a new todo and persist it."""
        # Domain entity handles validation and business rules
        todo = Todo.create(
            title=request.title,
            description=request.description
        )

        # Persist through repository
        self.repository.save(todo)

        return CreateTodoResponse(todo=todo)

Get Todo Use Case (application/use_cases/get_todo.py)

from dataclasses import dataclass
from typing import List, Optional

from domain.entities.todo import Todo
from domain.repositories.todo_repository import TodoRepository

@dataclass
class GetTodoResponse:
    """Response for a single todo."""
    todo: Optional[Todo]
    found: bool

@dataclass  
class GetAllTodosResponse:
    """Response for all todos."""
    todos: List[Todo]
    count: int

class GetTodoUseCase:
    """Use case for retrieving todos."""

    def __init__(self, repository: TodoRepository):
        self.repository = repository

    def execute(self, todo_id: str) -> GetTodoResponse:
        """Get a single todo by ID."""
        todo = self.repository.find_by_id(todo_id)
        return GetTodoResponse(
            todo=todo,
            found=todo is not None
        )

    def execute_all(self) -> GetAllTodosResponse:
        """Get all todos."""
        todos = self.repository.find_all()
        return GetAllTodosResponse(
            todos=todos,
            count=len(todos)
        )

Update Todo Use Case (application/use_cases/update_todo.py)

from dataclasses import dataclass
from typing import Optional

from domain.entities.todo import Todo
from domain.repositories.todo_repository import TodoRepository
from domain.value_objects.todo_status import TodoStatus

@dataclass
class UpdateTodoRequest:
    """Input data for updating a todo."""
    todo_id: str
    title: Optional[str] = None
    description: Optional[str] = None
    status: Optional[str] = None

@dataclass
class UpdateTodoResponse:
    """Output data after updating a todo."""
    todo: Optional[Todo]
    success: bool
    error: Optional[str] = None

class UpdateTodoUseCase:
    """Use case for updating an existing todo."""

    def __init__(self, repository: TodoRepository):
        self.repository = repository

    def execute(self, request: UpdateTodoRequest) -> UpdateTodoResponse:
        """Update a todo's properties."""
        todo = self.repository.find_by_id(request.todo_id)

        if not todo:
            return UpdateTodoResponse(
                todo=None,
                success=False,
                error="Todo not found"
            )

        try:
            if request.title is not None:
                todo.update_title(request.title)

            if request.description is not None:
                todo.update_description(request.description)

            if request.status is not None:
                new_status = TodoStatus.from_string(request.status)
                todo.change_status(new_status)

            self.repository.save(todo)

            return UpdateTodoResponse(todo=todo, success=True)

        except ValueError as e:
            return UpdateTodoResponse(
                todo=None,
                success=False,
                error=str(e)
            )

class DeleteTodoUseCase:
    """Use case for deleting a todo."""

    def __init__(self, repository: TodoRepository):
        self.repository = repository

    def execute(self, todo_id: str) -> bool:
        """Delete a todo by ID."""
        return self.repository.delete(todo_id)

Infrastructure Layer

The infrastructure layer provides concrete implementations.

Memory Repository (infrastructure/repositories/memory_todo_repository.py)

from typing import Dict, List, Optional

from domain.entities.todo import Todo
from domain.repositories.todo_repository import TodoRepository

class MemoryTodoRepository(TodoRepository):
    """In-memory implementation of TodoRepository."""

    def __init__(self):
        self._storage: Dict[str, Todo] = {}

    def save(self, todo: Todo) -> None:
        """Save a todo to memory."""
        self._storage[todo.id] = todo

    def find_by_id(self, todo_id: str) -> Optional[Todo]:
        """Find a todo by ID."""
        return self._storage.get(todo_id)

    def find_all(self) -> List[Todo]:
        """Get all todos."""
        return list(self._storage.values())

    def delete(self, todo_id: str) -> bool:
        """Delete a todo by ID."""
        if todo_id in self._storage:
            del self._storage[todo_id]
            return True
        return False

    def exists(self, todo_id: str) -> bool:
        """Check if a todo exists."""
        return todo_id in self._storage

You can easily swap this for a database implementation:

# infrastructure/repositories/sqlalchemy_todo_repository.py
from sqlalchemy.orm import Session
from domain.entities.todo import Todo
from domain.repositories.todo_repository import TodoRepository

class SQLAlchemyTodoRepository(TodoRepository):
    """SQLAlchemy implementation of TodoRepository."""

    def __init__(self, session: Session):
        self.session = session

    def save(self, todo: Todo) -> None:
        # Convert domain entity to ORM model and save
        pass

    # ... other methods

Interface Layer (Flask Routes)

The interface layer handles HTTP concerns and delegates to use cases.

Todo Routes (interface/api/todo_routes.py)

from flask import Blueprint, request, jsonify

from domain.repositories.todo_repository import TodoRepository
from application.use_cases.create_todo import CreateTodoUseCase, CreateTodoRequest
from application.use_cases.get_todo import GetTodoUseCase
from application.use_cases.update_todo import UpdateTodoUseCase, UpdateTodoRequest, DeleteTodoUseCase

def create_todo_routes(repository: TodoRepository) -> Blueprint:
    """Factory function to create routes with dependency injection."""

    bp = Blueprint('todos', __name__, url_prefix='/api/todos')

    # Initialize use cases with the repository
    create_todo = CreateTodoUseCase(repository)
    get_todo = GetTodoUseCase(repository)
    update_todo = UpdateTodoUseCase(repository)
    delete_todo = DeleteTodoUseCase(repository)

    @bp.route('', methods=['GET'])
    def list_todos():
        """Get all todos."""
        response = get_todo.execute_all()
        return jsonify({
            'todos': [t.to_dict() for t in response.todos],
            'count': response.count
        })

    @bp.route('', methods=['POST'])
    def create():
        """Create a new todo."""
        data = request.get_json(silent=True)

        if not data:
            return jsonify({'error': 'JSON body required'}), 400

        try:
            req = CreateTodoRequest(
                title=data.get('title', ''),
                description=data.get('description', '')
            )
            response = create_todo.execute(req)
            return jsonify({'todo': response.todo.to_dict()}), 201

        except ValueError as e:
            return jsonify({'error': str(e)}), 400

    @bp.route('/<todo_id>', methods=['GET'])
    def get(todo_id: str):
        """Get a specific todo."""
        response = get_todo.execute(todo_id)

        if not response.found:
            return jsonify({'error': 'Todo not found'}), 404

        return jsonify({'todo': response.todo.to_dict()})

    @bp.route('/<todo_id>', methods=['PUT'])
    def update(todo_id: str):
        """Update a todo."""
        data = request.get_json(silent=True)

        if not data:
            return jsonify({'error': 'JSON body required'}), 400

        req = UpdateTodoRequest(
            todo_id=todo_id,
            title=data.get('title'),
            description=data.get('description'),
            status=data.get('status')
        )

        response = update_todo.execute(req)

        if not response.success:
            status_code = 404 if response.error == "Todo not found" else 400
            return jsonify({'error': response.error}), status_code

        return jsonify({'todo': response.todo.to_dict()})

    @bp.route('/<todo_id>', methods=['DELETE'])
    def delete(todo_id: str):
        """Delete a todo."""
        if delete_todo.execute(todo_id):
            return jsonify({'message': 'Todo deleted'}), 200
        return jsonify({'error': 'Todo not found'}), 404

    @bp.route('/<todo_id>/complete', methods=['POST'])
    def complete(todo_id: str):
        """Mark a todo as completed."""
        req = UpdateTodoRequest(todo_id=todo_id, status='completed')
        response = update_todo.execute(req)

        if not response.success:
            status_code = 404 if response.error == "Todo not found" else 400
            return jsonify({'error': response.error}), status_code

        return jsonify({'todo': response.todo.to_dict()})

    return bp

Application Entry Point (app.py)

from flask import Flask

from infrastructure.repositories.memory_todo_repository import MemoryTodoRepository
from interface.api.todo_routes import create_todo_routes

def create_app() -> Flask:
    """Application factory with dependency injection."""
    app = Flask(__name__)

    # Create repository instance (easily swappable)
    todo_repository = MemoryTodoRepository()

    # Register routes with dependencies injected
    app.register_blueprint(create_todo_routes(todo_repository))

    @app.route('/')
    def index():
        return {'message': 'DDD Flask API is running!'}

    return app

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

Testing with DDD

One of the biggest benefits of DDD is testability. You can test each layer independently.

Testing Domain Logic (No Flask Required)

import pytest
from domain.entities.todo import Todo
from domain.value_objects.todo_status import TodoStatus

class TestTodo:
    def test_create_todo(self):
        todo = Todo.create("Buy groceries", "Milk, eggs, bread")

        assert todo.title == "Buy groceries"
        assert todo.description == "Milk, eggs, bread"
        assert str(todo.status) == "pending"

    def test_create_todo_empty_title_fails(self):
        with pytest.raises(ValueError, match="Title cannot be empty"):
            Todo.create("")

    def test_complete_todo(self):
        todo = Todo.create("Test task")
        todo.start()  # Must be in progress first
        todo.complete()

        assert todo.is_completed

    def test_invalid_status_transition(self):
        todo = Todo.create("Test task")

        with pytest.raises(ValueError, match="Cannot transition"):
            todo.complete()  # Can't go from pending to completed directly

class TestTodoStatus:
    def test_valid_transitions(self):
        pending = TodoStatus.pending()
        in_progress = TodoStatus.in_progress()

        assert pending.can_transition_to(in_progress)
        assert not pending.can_transition_to(TodoStatus.completed())

Testing Use Cases with Mock Repository

import pytest
from unittest.mock import Mock

from application.use_cases.create_todo import CreateTodoUseCase, CreateTodoRequest
from domain.repositories.todo_repository import TodoRepository

class TestCreateTodoUseCase:
    def test_create_todo_success(self):
        # Arrange
        mock_repo = Mock(spec=TodoRepository)
        use_case = CreateTodoUseCase(mock_repo)
        request = CreateTodoRequest(title="Test", description="Description")

        # Act
        response = use_case.execute(request)

        # Assert
        assert response.todo.title == "Test"
        mock_repo.save.assert_called_once()

    def test_create_todo_invalid_title(self):
        mock_repo = Mock(spec=TodoRepository)
        use_case = CreateTodoUseCase(mock_repo)
        request = CreateTodoRequest(title="")

        with pytest.raises(ValueError):
            use_case.execute(request)

        mock_repo.save.assert_not_called()

Benefits of DDD Architecture

  1. Clean Separation: Flask routes are thin controllers; business logic lives in the domain.

  2. Framework Independence: Domain and application layers have no Flask imports.

  3. Easy Testing: Test business rules without HTTP, test routes with mocked use cases.

  4. Swappable Infrastructure: Change from memory to PostgreSQL by implementing a new repository.

  5. Clear Boundaries: New developers immediately understand where code belongs.

  6. Maintainability: Changes to business rules don't affect HTTP handling and vice versa.

This architecture might seem like overkill for a simple todo app, but it pays dividends as your application grows. Start simple, refactor to DDD when complexity demands it.

Conclusion

You now have a solid understanding of how to handle POST requests in Flask. We covered:

  • Creating POST endpoints with @app.route
  • Receiving JSON data with request.json and request.get_json()
  • Handling form submissions with request.form
  • Processing raw data with request.data
  • Managing file uploads with request.files
  • Returning proper JSON responses with status codes
  • Implementing error handling
  • Validating input data
  • Testing with curl, Postman, and Python requests
  • Structuring larger applications with Domain-Driven Design (DDD)

For production applications, consider adding: - Authentication (JWT, OAuth) - Rate limiting - Request logging - Database integration (SQLAlchemy) - API documentation (Swagger/OpenAPI)

Happy coding!

api