HTTP-серверы на Python
http.server — встроенный сервер
Python имеет встроенный HTTP-сервер для отладки:
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
class APIHandler(BaseHTTPRequestHandler):
"""Simple API handler."""
def do_GET(self) -> None:
if self.path == "/api/health":
self._json_response({"status": "ok"})
elif self.path.startswith("/api/users"):
users = [{"id": 1, "name": "Алексей"}, {"id": 2, "name": "Мария"}]
self._json_response(users)
else:
self._json_response({"error": "Не найдено"}, status=404)
def do_POST(self) -> None:
content_length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_length)
data = json.loads(body)
if self.path == "/api/users":
response = {"id": 3, "name": data.get("name", "Unknown"), "created": True}
self._json_response(response, status=201)
else:
self._json_response({"error": "Не найдено"}, status=404)
def _json_response(self, data: dict | list, status: int = 200) -> None:
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.end_headers()
self.wfile.write(json.dumps(data, ensure_ascii=False).encode("utf-8"))
# server = HTTPServer(("localhost", 8000), APIHandler)
# server.serve_forever()
Важно: http.server предназначен только для разработки.
WSGI и ASGI
WSGI — синхронный стандарт. ASGI — асинхронный:
| WSGI | ASGI | |
|---|---|---|
| Модель | Синхронная | Асинхронная |
| WebSocket | Нет | Да |
| Серверы | Gunicorn, uWSGI | Uvicorn, Hypercorn |
| Фреймворки | Flask, Django | FastAPI, Starlette |
Flask — лёгкий WSGI-фреймворк
pip install flask
from flask import Flask, request, jsonify
app = Flask(__name__)
users: list[dict] = [
{"id": 1, "name": "Алексей", "email": "[email protected]"},
{"id": 2, "name": "Мария", "email": "[email protected]"},
]
next_id = 3
@app.route("/api/users", methods=["GET"])
def get_users():
return jsonify(users)
@app.route("/api/users/<int:user_id>", methods=["GET"])
def get_user(user_id: int):
user = next((u for u in users if u["id"] == user_id), None)
if user is None:
return jsonify({"error": "Не найден"}), 404
return jsonify(user)
@app.route("/api/users", methods=["POST"])
def create_user():
global next_id
data = request.get_json()
if not data or "name" not in data:
return jsonify({"error": "Поле 'name' обязательно"}), 400
user = {"id": next_id, "name": data["name"], "email": data.get("email", "")}
users.append(user)
next_id += 1
return jsonify(user), 201
@app.route("/api/users/<int:user_id>", methods=["DELETE"])
def delete_user(user_id: int):
global users
users = [u for u in users if u["id"] != user_id]
return "", 204
# if __name__ == "__main__":
# app.run(debug=True, port=5000)
FastAPI — современный ASGI-фреймворк
pip install fastapi uvicorn
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field
app = FastAPI(title="Users API", version="1.0.0")
class UserCreate(BaseModel):
name: str = Field(min_length=1, max_length=100)
email: str
age: int = Field(ge=0, le=150)
class UserResponse(BaseModel):
id: int
name: str
email: str
age: int
users_db: dict[int, dict] = {}
next_id = 1
@app.get("/api/users", response_model=list[UserResponse])
async def get_users() -> list[dict]:
return list(users_db.values())
@app.get("/api/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int) -> dict:
if user_id not in users_db:
raise HTTPException(status_code=404, detail="Не найден")
return users_db[user_id]
@app.post("/api/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate) -> dict:
global next_id
user_data = {"id": next_id, "name": user.name, "email": user.email, "age": user.age}
users_db[next_id] = user_data
next_id += 1
return user_data
@app.delete("/api/users/{user_id}", status_code=204)
async def delete_user(user_id: int) -> None:
if user_id not in users_db:
raise HTTPException(status_code=404, detail="Не найден")
del users_db[user_id]
# Run: uvicorn main:app --reload
# Docs: http://localhost:8000/docs
Сравнение Flask и FastAPI
| Характеристика | Flask | FastAPI |
|---|---|---|
| Стандарт | WSGI | ASGI |
| Валидация | Ручная | Автоматическая (Pydantic) |
| Документация | Ручная | Автоматическая (OpenAPI) |
| Async/await | Ограниченно | Полная поддержка |
| WebSocket | Flask-SocketIO | Встроено |
| Зрелость | С 2010 | С 2018 |
Тестирование API
# FastAPI testing
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
@app.get("/api/greet/{name}")
async def greet(name: str) -> dict:
return {"message": f"Привет, {name}!"}
def test_greet():
client = TestClient(app)
response = client.get("/api/greet/Алексей")
assert response.status_code == 200
assert response.json()["message"] == "Привет, Алексей!"
Практический пример: API заметок
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from datetime import datetime
app = FastAPI(title="Notes API")
class NoteCreate(BaseModel):
title: str = Field(min_length=1, max_length=200)
content: str = Field(default="")
tags: list[str] = Field(default_factory=list)
class NoteResponse(BaseModel):
id: int
title: str
content: str
tags: list[str]
created_at: str
notes_db: dict[int, dict] = {}
next_id = 1
@app.post("/api/notes", response_model=NoteResponse, status_code=201)
async def create_note(note: NoteCreate) -> dict:
global next_id
data = {
"id": next_id, "title": note.title, "content": note.content,
"tags": note.tags, "created_at": datetime.now().isoformat(),
}
notes_db[next_id] = data
next_id += 1
return data
@app.get("/api/notes", response_model=list[NoteResponse])
async def list_notes(tag: str | None = None, search: str | None = None) -> list[dict]:
results = list(notes_db.values())
if tag:
results = [n for n in results if tag in n["tags"]]
if search:
q = search.lower()
results = [n for n in results if q in n["title"].lower() or q in n["content"].lower()]
return results
@app.delete("/api/notes/{note_id}", status_code=204)
async def delete_note(note_id: int) -> None:
if note_id not in notes_db:
raise HTTPException(404, "Заметка не найдена")
del notes_db[note_id]
# Run: uvicorn notes:app --reload