2 min read

Building a Simple REST API

This guide demonstrates how to build a basic REST API using Python's standard library and marshmallow for validation.

Modules Used:

  • http.server: To handle HTTP requests.
  • json: To parse and format data.
  • marshmallow: To validate incoming data and serialize objects.

The Code

Save this as api.py.

from http.server import HTTPServer, BaseHTTPRequestHandler
import json
from marshmallow import Schema, fields, ValidationError

# --- Database (In-Memory) ---
tasks = [
    {"id": 1, "title": "Learn Python", "done": True},
    {"id": 2, "title": "Build an API", "done": False}
]

# --- Schema ---
class TaskSchema(Schema):
    id = fields.Int(dump_only=True)
    title = fields.Str(required=True)
    done = fields.Bool()

task_schema = TaskSchema()
tasks_schema = TaskSchema(many=True)

# --- Request Handler ---
class APIHandler(BaseHTTPRequestHandler):

    def _send_response(self, data, status=200):
        self.send_response(status)
        self.send_header('Content-type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(data).encode('utf-8'))

    def _send_error(self, message, status=400):
        self._send_response({"error": message}, status)

    def do_GET(self):
        if self.path == '/tasks':
            # Return list of tasks
            self._send_response(tasks_schema.dump(tasks))
        else:
            self._send_error("Not Found", 404)

    def do_POST(self):
        if self.path == '/tasks':
            # Get content length to read body
            try:
                content_length = int(self.headers['Content-Length'])
                post_data = self.rfile.read(content_length)
                json_data = json.loads(post_data)

                # Validate data
                new_task = task_schema.load(json_data)

                # Save to "Database"
                new_task['id'] = len(tasks) + 1
                tasks.append(new_task)

                self._send_response(task_schema.dump(new_task), 201)

            except json.JSONDecodeError:
                self._send_error("Invalid JSON Body")
            except ValidationError as err:
                self._send_error(err.messages)
            except Exception as e:
                self._send_error(str(e), 500)
        else:
            self._send_error("Not Found", 404)

# --- Server Setup ---
def run(server_class=HTTPServer, handler_class=APIHandler, port=8000):
    server_address = ('', port)
    httpd = server_class(server_address, handler_class)
    print(f"Starting API on port {port}...")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    httpd.server_close()

if __name__ == '__main__':
    run()

Usage

  1. Start the server:

    python api.py
  2. Test with curl or Postman:

    • Get Tasks:

      curl http://localhost:8000/tasks
    • Create Task:

      curl -X POST -H "Content-Type: application/json" -d '{"title": "Read documentation"}' http://localhost:8000/tasks
    • Test Validation (Error):

      curl -X POST -H "Content-Type: application/json" -d '{"done": true}' http://localhost:8000/tasks

programming/python/python