Перед тем как писать код, хороший разработчик внимательно читает ТЗ. Прочитай его целиком, прежде чем переходить к следующей части.
Нам поступил заказ на разработку упрощённой модели того, как устроен веб-фреймворк изнутри: часть системы, которая принимает запрос, находит подходящий обработчик, тот в свою очередь обращается к базе данных и возвращает ответ.
Это учебная модель, а не реальный сервер — она не работает по сети и не использует настоящую базу данных. Наша задача — воспроизвести внутреннюю механику на объектах, чтобы отработать магические методы Python на знакомом контексте (см. вводную лекцию к модулю).
Система должна уметь:
- представлять входящий запрос — метод (
GET,POSTи так далее), путь, заголовки, тело; - представлять исходящий ответ — статус-код и тело;
- регистрировать обработчики для конкретных пар (метод, путь);
- находить подходящий обработчик для входящего запроса и вызывать его;
- открывать и корректно закрывать соединение с базой данных при обработке запроса, даже если внутри произошла ошибка;
- вести историю всех обработанных запросов, по которой можно пройтись в цикле
for.
- Заголовки запроса должны быть доступны и как обычный словарь (
request.headers["content-type"]), и как атрибут объекта (request.content_type) — второй способ должен работать «сам по себе», без необходимости заранее объявлять каждый возможный заголовок как атрибут класса. - Если заголовок, к которому обратились как к атрибуту, отсутствует — возвращается
None, а не выбрасывается исключение. - Ответ считается успешным, если его статус-код лежит в диапазоне
[200, 300). Ответ должен уметь сообщать об этом в булевом контексте — то есть должно быть возможно написатьif response:вместоif 200 <= response.status_code < 300:. - Нельзя зарегистрировать два обработчика для одной и той же пары (метод, путь) — повторная регистрация должна завершаться понятной ошибкой.
- Если для входящего запроса не нашлось подходящего зарегистрированного обработчика — система должна вернуть ответ со статус-кодом
404, а не упасть с исключением. - Соединение с базой данных должно гарантированно закрываться после выполнения запроса — вне зависимости от того, произошла ли внутри ошибка.
- Историю обработанных запросов можно обойти в цикле
for— там должны быть доступны пары «запрос — ответ» в том порядке, в котором они были обработаны.
| Требование | Детали |
|---|---|
| Язык | Python 3.10+ |
| Хранилище данных | не требуется (сеть и база данных — имитация в памяти, без файлов) |
| Интерфейс | консоль, точка входа — скрипт с заранее заданными тестовыми запросами (без input()) |
| Сторонние библиотеки | не использовать (только стандартная библиотека) |
Здесь описана точная архитектура, которую нужно реализовать. Отступать от неё без согласования с преподавателем нельзя — архитектура специально спроектирована так, чтобы закрепить пройденные темы Модуля 3.
mini_framework/
│
├── models/
│ ├── __init__.py
│ ├── request.py # класс Request
│ └── response.py # класс Response
│
├── router/
│ ├── __init__.py
│ ├── route.py # класс Route
│ └── router.py # класс Router
│
├── db/
│ ├── __init__.py
│ └── database_connection.py # класс DatabaseConnection
│
├── app/
│ ├── __init__.py
│ ├── request_log.py # класс RequestLog
│ └── app.py # класс App
│
└── main.py # точка входа
Представляет входящий запрос.
Атрибуты:
| Атрибут | Тип | Описание |
|---|---|---|
method |
str |
HTTP-метод ("GET", "POST" и т. д.) |
path |
str |
путь запроса, например "/api/users" |
headers |
dict |
заголовки запроса, ключи приведены к нижнему регистру |
body |
любой тип | тело запроса (может быть None) |
Методы:
-
__init__(method, path, headers=None, body=None)— сохраняет метод и путь, приводит ключи заголовков к нижнему регистру (еслиheadersне передан — используется пустой словарь), сохраняет тело. -
__str__— возвращает строку вида"GET /api/users"(метод и путь через пробел). -
__repr__— возвращает техническую строку вида"Request(method='GET', path='/api/users')". -
__getattr__(name)— вызывается только тогда, когда атрибут не найден обычным способом (то есть не дляmethod,path,headers,body— они лежат в__dict__и находятся раньше). Должен превратить имя атрибута в имя заголовка (заменить_на-) и поискать его вself.headers. Если заголовок найден — вернуть его значение, если нет — вернутьNone.Обрати внимание на важную деталь безопасности: внутри
__getattr__нельзя писатьself.headers, если есть риск, что__getattr__может быть вызван раньше, чем атрибутheadersвообще появится в объекте, — это привело бы к бесконечной рекурсии (разобрано в Уроке 11). Безопасный способ — обратиться напрямую кself.__dict__.get("headers", {}), полностью в обход обычного поиска атрибута.
Представляет исходящий ответ.
Атрибуты:
| Атрибут | Тип | Описание |
|---|---|---|
status_code |
int |
статус-код ответа |
body |
любой тип | тело ответа |
Методы:
-
__init__(status_code, body=None)— сохраняет оба значения. -
__bool__— возвращаетTrue, если200 <= status_code < 300, иначеFalse. -
__str__— возвращает строку вида"[200 OK] {...тело...}"либо"[404 ERROR] {...тело...}"— используй результатbool(self)(то естьif self:), чтобы определить, писатьOKилиERROR, а не дублируй проверку диапазона status_code второй раз. -
__repr__— возвращает техническую строку вида"Response(status_code=200, body=...)".
Представляет один зарегистрированный маршрут — связку «метод + путь + функция-обработчик».
Атрибуты:
| Атрибут | Тип | Описание |
|---|---|---|
method |
str |
HTTP-метод маршрута |
path |
str |
путь маршрута |
handler |
функция | функция-обработчик, принимающая Request и возвращающая Response |
Методы:
-
__init__(method, path, handler)— сохраняет все три значения. -
__call__(request)— делает объектRouteвызываемым: вызываетself.handler(request)и возвращает результат. Снаружи это должно позволять писатьroute(request)вместоroute.handler(request). -
__eq__(other)— два маршрута считаются равными, если у них совпадаютmethodиpath(handlerв сравнении не участвует). Еслиotherне является объектомRoute— верниNotImplemented. -
__str__и__repr__— как обычно, читаемое и техническое представление ("GET /api/users"и"Route(method='GET', path='/api/users')"соответственно).
Обрати внимание: если класс переопределяет
__eq__, но не переопределяет__hash__, объект становится нехешируемым (разобрано в Уроке 17). В этой задачеRouteникогда не кладётся вsetи не используется как ключ словаря, поэтому дополнительно реализовывать__hash__не требуется — но важно понимать, почему в других ситуациях это могло бы стать проблемой.
Хранит все зарегистрированные маршруты и находит подходящий для входящего запроса.
Атрибуты:
| Атрибут | Тип | Описание |
|---|---|---|
_routes |
list[Route] |
protected-список зарегистрированных маршрутов |
Методы:
-
__init__()— создаёт пустой список_routes. -
register(method, path, handler)— создаёт объектRoute, предварительно проверяя через__eq__(оператором==, без ручного сравнения полей), что маршрут с такими жеmethodиpathещё не зарегистрирован. Если уже зарегистрирован — выбрасываетValueError. Иначе добавляет новый маршрут в_routes. -
dispatch(request)— ищет в_routesмаршрут, у которогоmethodиpathсовпадают сrequest.methodиrequest.path. Если нашёлся — вызывает его как функцию (route(request), используя__call__) и возвращает результат. Если не нашёлся — возвращаетResponse(404, {"error": "Маршрут не найден"}).
Имитация соединения с базой данных, реализующая протокол контекстного менеджера.
Атрибуты:
| Атрибут | Тип | Описание |
|---|---|---|
name |
str |
условное имя базы данных |
is_open |
bool |
признак того, что соединение сейчас открыто |
Методы:
-
__init__(name)— сохраняет имя, устанавливаетis_open = False. -
__enter__()— устанавливаетis_open = True, печатает сообщение об открытии соединения, возвращаетself. -
__exit__(exc_type, exc_val, exc_tb)— устанавливаетis_open = False. Если внутри блокаwithпроизошло исключение (exc_type is not None) — печатает сообщение о том, что транзакция отменена из-за ошибки; иначе — сообщение об успешном закрытии соединения. В любом случае должен возвращатьFalse, чтобы не подавлять исключение, если оно произошло (это разбиралось на Уроке 15). -
fetch(query)— печатает сообщение о выполняемом запросе и возвращает словарь-заглушку вида{"query": query, "result": "OK"}— имитацию данных, «прочитанных» из базы.
Хранит историю всех обработанных пар «запрос — ответ» и позволяет обходить её в цикле for.
Атрибуты:
| Атрибут | Тип | Описание |
|---|---|---|
_entries |
list[tuple[Request, Response]] |
protected-список пар «запрос, ответ» |
_position |
int |
protected-текущая позиция при обходе |
Методы:
-
__init__()— создаёт пустой список_entriesи_position = 0. -
add(request, response)— добавляет пару(request, response)в конец_entries. -
__iter__()— сбрасывает_positionв0и возвращаетself(класс сам выступает и итерируемым объектом, и итератором — тот же подход, что разбирался на Уроке 14 на примереCountdown). -
__next__()— если_positionдостиг длины_entries, выбрасываетStopIteration; иначе возвращает текущую пару и увеличивает_positionна 1.
Обрати внимание: поскольку
__iter__каждый раз сбрасывает_position, повторные последовательные проходы (for ... in log, выполненные один за другим) будут работать корректно. А вот вложенные одновременные проходы по одному и тому же объектуRequestLog(один циклforвнутри другого, оба — по одному и тому же логу) будут работать некорректно, так как оба цикла делят одну и ту же позицию_position. Для этой задачи такой сценарий не нужен, но важно понимать эту границу применимости — она прямо разбиралась на Уроке 14.
Связывает маршрутизацию и историю запросов в единую точку входа.
Атрибуты:
| Атрибут | Тип | Описание |
|---|---|---|
router |
Router |
маршрутизатор приложения |
log |
RequestLog |
история обработанных запросов |
Методы:
-
__init__()— создаётRouter()иRequestLog(). -
register_route(method, path, handler)— делегирует вself.router.register(...). -
handle_request(request)— получает ответ черезself.router.dispatch(request), добавляет пару вself.log, возвращает ответ.
main.py
└── создаёт App
├── App.router (Router) хранит список Route
│ └── Route.__call__ вызывает функцию-обработчик,
│ которая внутри может открыть DatabaseConnection (with)
└── App.log (RequestLog) — история пар (Request, Response),
по которой можно пройтись через for
Здесь регистрируются функции-обработчики (обычные функции, не методы класса — они принимают Request и возвращают Response), создаётся несколько тестовых объектов Request, каждый прогоняется через app.handle_request(...), и в конце распечатывается вся история через RequestLog.
# main.py — структура (не финальный код, только скелет)
from models.request import Request
from app.app import App
def get_users(request):
# проверить request.authorization через __getattr__
# открыть DatabaseConnection через with, получить данные
# вернуть Response
pass
def main():
app = App()
app.register_route("GET", "/api/users", get_users)
# зарегистрировать другие маршруты
requests = [
# несколько тестовых объектов Request
]
for req in requests:
response = app.handle_request(req)
# напечатать req, response, и результат `if response:`
# пройтись по app.log и напечатать историю
if __name__ == "__main__":
main()Конкретная реализация всегда зависит от разработчика.
from models.request import Request
from models.response import Response
from db.database_connection import DatabaseConnection
from app.app import App
def get_users(request):
if not request.authorization:
return Response(401, {"error": "Требуется авторизация"})
with DatabaseConnection("users_db") as db:
data = db.fetch("SELECT * FROM users")
return Response(200, data)
def create_order(request):
if not request.body:
return Response(400, {"error": "Тело запроса не может быть пустым"})
with DatabaseConnection("orders_db") as db:
data = db.fetch(f"INSERT INTO orders VALUES ({request.body})")
return Response(201, data)
def main():
app = App()
app.register_route("GET", "/api/users", get_users)
app.register_route("POST", "/api/orders", create_order)
requests = [
Request("GET", "/api/users", headers={"Authorization": "Bearer abc123"}),
Request("GET", "/api/users"),
Request("POST", "/api/orders", body={"item": "Книга", "qty": 2}),
Request("DELETE", "/api/users/1"),
]
for req in requests:
print(f"\n>>> {req}")
response = app.handle_request(req)
print(response)
print("Успешно" if response else "Ошибка")
print("\n=== История запросов ===")
for req, resp in app.log:
print(f"{req} -> {resp.status_code}")
if __name__ == "__main__":
main()Ожидаемый вывод программы:
>>> GET /api/users
[DB] Соединение с 'users_db' открыто
[DB] Выполняется запрос: SELECT * FROM users
[DB] Соединение с 'users_db' закрыто, транзакция подтверждена
[200 OK] {'query': 'SELECT * FROM users', 'result': 'OK'}
Успешно
>>> GET /api/users
[401 ERROR] {'error': 'Требуется авторизация'}
Ошибка
>>> POST /api/orders
[DB] Соединение с 'orders_db' открыто
[DB] Выполняется запрос: INSERT INTO orders VALUES ({'item': 'Книга', 'qty': 2})
[DB] Соединение с 'orders_db' закрыто, транзакция подтверждена
[201 OK] {'query': "INSERT INTO orders VALUES ({'item': 'Книга', 'qty': 2})", 'result': 'OK'}
Успешно
>>> DELETE /api/users/1
[404 ERROR] {'error': 'Маршрут не найден'}
Ошибка
=== История запросов ===
GET /api/users -> 200
GET /api/users -> 401
POST /api/orders -> 201
DELETE /api/users/1 -> 404
Ориентировочное время выполнения: 3–4 академических часа.