Обработчики

Аiohttp позволяет реализовать обработчики асинхронными функциями и классами. Классы более расширяемы: во-первых, код, относящийся к одному обработчику, можно разместить в одном месте, а во вторых, классы позволяют использовать наследование для избавления от дублирования кода (например, каждому обработчику требуется соединение с базой данных)

POST /imports

- Добавить выгрузку с информацией о жителях

На вход обработчик получает json с данными о жителях. Проверить и десериализовать данные. Если они некорректные, нужно вернуть HTTP-ответ 400: Bad Request. Потребуются две схемы Marhsmallow. Первая, CitizenSchema, проверяет данные каждого отдельного жителя, а также десериализует строку с днем рождения в объект datetime.date

Вторая схема, ImportSchema, проверяет выгрузку в целом

Если данные корректные, их необходимо добавить в БД с новым уникальным import_id. Для добавления данных потребуется выполнить несколько запросов в разные таблицы. Чтобы в БД не осталось частично добавленных данных в случае возникновения ошибки или исключения (например, при отключении клиента, который не получил ответ полностью, aiohttp бросит исколючение CancelledError), необходимо использовать транзакцию.

Рисунок 24– Код файла “Shema.py”

Рисунок 25– Код файла “import_citizens.py”

Рисунок 26– Код файла “Imports.py”

Рисунок 27– Код файла “Imports.py”

Рисунок 28– Код файла “Imports.py”

Рисунок 26– Код файла “Imports.py”

GET /imports/$import_id/citizens

Обработчик возвращает всех жителей для выгрузки с указанным import_id. Если указанная выгрузка не существует, необходимо вернуть HTTP-ответ 404: Not Found. Это поведение выглядит общим для обработчиков, которым требуется существующая выгрузка, поэтому я вынес код проверки в отдельный класс.

Рисунок 26– Код файла “Base.py”

Чтобы получить список родственников для каждого жителя, потребуется выполнить LEFT JOIN из таблицы citizens в таблицу relations, агрегируя поле relations.relative_id с группировкой по import_id и citizen_id.

Если у жителя нет родственников, то LEFT JOIN вернет для него в поле relations.relative_id значение NULL и в результате агрегации список родственников будет выглядеть как [NULL].

Чтобы исправить это некорректное значение, я воспользовался функцией array_remove.

Aiohttp обнаружит в реестре aiohttp.PAYLOAD_REGISTRY зарегистрированный сериализатор AsyncGenJSONListPayload для объектов типа AsyncIterable. Затем сериализатор будет итерироваться по объекту SelectQuery и отправлять данные клиенту. При первом обращении объект SelectQuery получает соединение к БД, открывает транзакцию и создает курсор, при дальнейшей итерации будет получать данные из БД курсором и возвращать их построчно.

Этот подход позволяет не выделять память на весь объем данных при каждом запросе, но у него есть особенность: приложение не сможет вернуть клиенту соответствующий HTTP-статус, если возникнет ошибка (ведь клиенту уже был отправлен HTTP-статус, заголовки, и пишутся данные).

При возникновении исключения не остается ничего, кроме как разорвать соединение. Исключение, конечно, можно залогировать, но клиент не сможет понять, какая именно ошибка произошла.

Рисунок 27– Код файла “Querry.py”

PATCH /imports/$import_id/citizens/$citizen_id

Обработчик получает на вход идентификатор выгрузки import_id, жителя citizen_id, а также json с новыми данными о жителе. В случае обращения к несуществующей выгрузке или жителю возвращается HTTP-ответ 404: Not Found. Переданные клиентом данные требуется проверяются и десериализовываются. Если они некорректные —возвращаю HTTP-ответ 400: Bad Request. Я реализовал Marshmallow-схему PatchCitizenSchema, которая проверяет:

• Тип и формат данных для указанных полей.

• Дату рождения. Она должна быть указана в формате DD.MM.YYYY и не может иметь значения из будущего.

• Список родственников каждого жителя. Он должен иметь уникальные идентификаторы жителей

Рисунок 28– Код файла “Patch_citizen.py”

Существование родственников, указанных в поле relatives, можно отдельно не проверять: при добавлении в таблицу relations несуществующего жителя PostgreSQL вернет ошибку ForeignKeyViolationError, которую можно обработать и вернуть HTTP-статус 400: Bad Request. Если клиент прислал некорректные данные для несуществующего жителя или выгрузки, я сначала проверяю данные, и только если они корректные, обращаться к базе.

Если данные корректные, обновляю информацию о жителе в БД. В обработчике делаю несколько запросов к разным таблицам. Если возникнет ошибка или исключение, изменения в базе данных отменяются, поэтому запросы выполняются в транзакции.

Метод PATCH позволяет передавать лишь некоторые поля для изменяемого жителя.

Обработчик необходимо написать таким образом, чтобы он не падал при обращении к данным, которые не указал клиент, а также не выполнял запросы к таблицам, данные в которых не изменились. Если клиент указал поле relatives, необходимо получить список существующих родственников. Если он изменился — определить, какие записи из таблицы relatives необходимо удалить, а какие добавить, чтобы привести базу данных в соответствие с запросом клиента. По умолчанию в PostgreSQL для изоляции транзакций используется уровень READ COMMITTED. Это означает, что в рамках текущей транзакции будут видны изменения существующих (а также добавления новых) записей других завершенных транзакций. Это может привести к состоянию гонки между конкурентными запросами. Предположим, существует выгрузка с жителями #1, #2, #3, без родственных связей. Сервис получает два одновременных запроса на изменение жителя #1: {"relatives": [2]} и {"relatives": [3]}. aiohttp создаст два обработчика, которые одновременно получат текущее состояние жителя из PostgreSQL. Каждый обработчик не обнаружит ни одной родственной связи и примет решение добавить новую связь с указанным родственником. В результате у жителя #1 поле relatives равно [2,3].

Рисунок 29– Таблица запросов

Такое поведение нельзя назвать очевидным. Есть два варианта ожидаемо решить исход гонки: выполнить только первый запрос, а для второго вернуть HTTP-ответ 409: Conflict (чтобы клиент повторил запрос), либо выполнить запросы по очереди (второй запрос будет обработан только после завершения первого).

Первый вариант можно реализовать, включив режим изоляции SERIALIZABLE. Если во время обработки запроса кто-то уже успел изменить и закоммитить данные, будет брошено исключение, которое можно обработать и вернуть соответствующий HTTP-статус.

Минус такого решения — большое число блокировок в PostgreSQL, SERIALIZABLE будет вызывать исключение, даже если конкурентные запросы меняют записи жителей из разных выгрузок.

Также можно воспользоваться механизмом рекомендательных блокировок. Если получить такую блокировку по import_id, конкурентные запросы для разных выгрузок смогут выполняться параллельно.

Для обработки конкурентных запросов в одной выгрузке можно реализовать поведение любого из вариантов: функция pg_try_advisory_xact_lock пытается получить блокировку и возвращает результат boolean немедленно (если блокировку получить не удалось — можно бросить исключение), а pg_advisory_xact_lock ожидает, пока ресурс не станет доступен для блокировки (в этом случае запросы выполнятся последовательно, я остановился на этом варианте).

В итоге обработчик должен вернуть актуальную информацию об обновленном жителе. Можно было ограничиться возвращением клиенту данных из его же запроса (раз мы возвращаем ответ клиенту, значит, исключений не было и все запросы успешно выполнены). Или — воспользоваться ключевым словом RETURNING в запросах, изменяющих БД, и сформировать ответ из полученных результатов. Но оба этих подхода не позволили бы увидеть и протестировать случай с гонкой состояний.

К сервису не предъявлялись требования по высокой нагрузке, поэтому можно запрашивать все данные о жителе заново и возвращать клиенту честный результат из БД.

GET /imports/$import_id/citizens/birthdays

GET /imports/$import_id/citizens/birthdays Обработчик вычисляет число подарков, которое приобретет каждый житель выгрузки своим родственникам (первого порядка). Число сгруппировано по месяцам для выгрузки с указанным import_id. В случае обращения к несуществующей выгрузке возвращаю HTTP-ответ 404: Not Found.

Есть два варианта реализации:

1. Получить данные для жителей с родственниками из базы, а на стороне Python агрегировать данные по месяцам и сгенерировать списки для тех месяцев, для которых нет данных в БД.

2. Cоставить json-запрос в базу и дописать для отсутствующих месяцев заглушки.

Я остановился на первом варианте — визуально он выглядит более понятным и поддерживаемым. Число дней рождений в определенном месяце можно получить, сделав JOIN из таблицы с родственными связями (relations.citizen_id — житель, для которого мы считаем дни рождения родственников) в таблицу citizens (содержит дату рождения, из которой требуется получить месяц)

Значения месяцев не должны содержать ведущих нулей. Месяц, получаемый из поля birth_date c помощью функции date_part, может содержать ведущий ноль. Чтобы убрать его, я выполнил cast к integer в SQL-запросе. Несмотря на то, что в обработчике требуется выполнить два запроса (проверить существование выгрузки и получить информации о днях рождения и подарках), транзакция не требуется

По умолчанию PostgreSQL использует режим READ COMMITTED, при котором в текущей транзакции видны все новые (добавляемые другими транзакциями) и существующие (изменяемые другими транзакциями) записи после их успешного завершения. Например, если в момент получения данных будет добавлена новая выгрузка — она никак не повлияет на существующие. Если в момент получения данных будет выполнен запрос на изменение жителя — то либо данные еще не будут видны (если транзакция, меняющая данные, не завершилась), либо транзакция полностью завершится и станут видны сразу все изменения. Целостность получаемых из базы данных не нарушится.

Рисунок 30– Код файла «citizen_bithzd.py»

GET /imports/$import_id/towns/stat/percentile/age

Обработчик вычисляет 50-й, 75-й и 99-й перцентили возрастов (полных лет) жителей по городам в выборке с указанным import_id. В случае обращения к несуществующей выгрузке возвращаю HTTP-ответ 404: Not Found.

Есть два варианта реализации:

1. Получить из БД возраста жителей, сгруппированные по городам, а затем на стороне Python вычислить перцентили с помощью numpy (который в задании указан как эталонный) и округлить до двух знаков после запятой.

2. Сделать всю работу на стороне PostgreSQL: функция percentile_cont вычисляет перцентиль с линейной интерполяцией, затем округляем полученные значения до двух знаков после запятой в рамках одного SQLзапроса, а numpy используем для тестирования.

Второй вариант требует передавать меньше данных между приложением и PostgreSQL, но у него есть не очень очевидный подводный камень: в PostgreSQL округление математическое, (SELECT ROUND(2.5) вернет 3), а в Python — бухгалтерское, к ближайшему целому (round(2.5) вернет 2).

Чтобы тестировать обработчик, реализация должна быть одинаковой и в PostgreSQL, и в Python (реализовать функцию с математическим округлением в Python выглядит проще). Стоит отметить, что при вычислении перцентилей numpy и PostgreSQL могут возвращать немного отличающиеся числа, но с учетом округления эта разница будет незаметна

Рисунок 31 – Код файла «town_stat_citizen.py»