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
-
Start the server:
python api.py -
Test with
curlor 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
-