Небольшой, но полный API
Пора собрать всё вместе: API пользователей с пятью классическими конечными точками —
POST /users создать · GET /users список · GET /users/{id} прочитать одного · PUT /users/{id} заменить · DELETE /users/{id} удалить.
Хранилище — словарь в памяти, поэтому пример остаётся самодостаточным. (Настоящий сервер обрабатывает запросы параллельно, поэтому рабочая версия защищала бы словарь блокировкой и, скорее всего, заменила бы его базой данных — обработчики при этом не изменились бы.)
Создание и список
POST разбирает входящий JSON, присваивает пользователю id, сохраняет его и отвечает 201 с созданной записью — вместе с id, чтобы клиент его узнал. GET возвращает всех пользователей в виде JSON-массива.
POST /users и GET /users
Два создания, затем список: каждый POST отвечает 201 с сохранённым пользователем, а последний GET показывает обоих — состояние API росло с каждым запросом.
Вывод: полная программа на Python
import json
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
users = {}
next_id = 1
class UsersHandler(BaseHTTPRequestHandler):
def do_POST(self):
global next_id
length = int(self.headers.get("Content-Length", 0))
try:
data = json.loads(self.rfile.read(length))
except json.JSONDecodeError:
self.reply(400, {"error": "invalid JSON"})
return
user = {"id": next_id, "name": data["name"]}
users[next_id] = user
next_id += 1
self.reply(201, user)
def do_GET(self):
self.reply(200, [users[k] for k in sorted(users)])
def reply(self, status, data):
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write((json.dumps(data) + "\n").encode())
def log_message(self, fmt, *args):
pass
server = ThreadingHTTPServer(("127.0.0.1", 0), UsersHandler)
base = f"http://127.0.0.1:{server.server_address[1]}"
threading.Thread(target=server.serve_forever, daemon=True).start()
for name in ["Ana", "Ben"]:
payload = json.dumps({"name": name}).encode()
req = urllib.request.Request(base + "/users", data=payload,
headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req) as resp:
print("POST /users ->", resp.status, resp.read().decode().strip())
with urllib.request.urlopen(base + "/users") as resp:
print("GET /users ->", resp.status, resp.read().decode().strip())
server.shutdown()
Чтение одного, замена, удаление
Маршруты /users/{id} завершают набор: чтение отвечает 200 или 404, замена записывает новые данные под существующим id, удаление отвечает 204. После удаления тот же GET, который работал раньше, честно возвращает 404 — ресурса больше нет.
GET, PUT и DELETE для /users/{id}
Полный жизненный цикл одного ресурса: прочитать его, заменить (ответ содержит новое имя), удалить и не суметь прочитать — каждый шаг с правильным кодом состояния.
Вывод: полная программа на Python
import json
import threading
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
users = {1: {"id": 1, "name": "Ana"}}
class UserHandler(BaseHTTPRequestHandler):
def user_id(self):
return int(self.path.rsplit("/", 1)[-1]) # "/users/1" -> 1
def do_GET(self):
user = users.get(self.user_id())
if user is None:
self.reply(404, {"error": "user not found"})
else:
self.reply(200, user)
def do_PUT(self):
length = int(self.headers.get("Content-Length", 0))
data = json.loads(self.rfile.read(length))
user = {"id": self.user_id(), "name": data["name"]} # сохраняем id
users[self.user_id()] = user
self.reply(200, user)
def do_DELETE(self):
users.pop(self.user_id(), None)
self.send_response(204)
self.end_headers()
def reply(self, status, data):
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write((json.dumps(data) + "\n").encode())
def log_message(self, fmt, *args):
pass
server = ThreadingHTTPServer(("127.0.0.1", 0), UserHandler)
base = f"http://127.0.0.1:{server.server_address[1]}"
threading.Thread(target=server.serve_forever, daemon=True).start()
def call(method, path, payload=None):
data = payload.encode() if payload else None
req = urllib.request.Request(base + path, data=data, method=method,
headers={"Content-Type": "application/json"})
try:
with urllib.request.urlopen(req) as resp:
status, body = resp.status, resp.read().decode()
except urllib.error.HTTPError as err:
status, body = err.code, err.read().decode()
print(f"{method:<6} {path} -> {status} {body.strip()}".rstrip())
call("GET", "/users/1")
call("PUT", "/users/1", '{"name": "Ana Maria"}')
call("DELETE", "/users/1")
call("GET", "/users/1")
server.shutdown()