4.5. Проект «Создание ASGI/WSGI-сервера»

Если возникли сложности, обратитесь к подсказкам по каждой из трёх задач.

ПодсказкиПодсказки

# WSGI-сервер (генератор) — эту функцию нужно реализовать! 
def wsgi_server(requests):
    """Обрабатывает запросы и возвращает каждый из ответов через yield"""
    request = yield None #запуск генератора, приём первого запроса app.send(None) 
    while True:
    # Здесь необходимо написать обработку request и вызов нужной ручки для получения response  
        request = yield response

# Функция маршрутизации
def router(path):
    if path == "/":
        return home_handler
    elif path.startswith("/user"):
        return user_handler
    elif path.startswith("/attempt"):
        return attempt_handler
    elif path.startswith("/admin"):
        return admin_handler

    return not_found_handler

request = “POST /attempt?name=Dima&task=1”


path = "/attempt?name=Dima&task=1"
params = {"name": "Dima", "task": "1"}

handler = router(path)    #Идентично handler = attempt_handler
response = handler(params)    #Применяете как обычную функцию и получаете ответ

{"handler": "home_handler", "params": {}, "response": [200, "Home Page"]}
{"handler": "not_found_handler", "params": {}, "response": [404, "Not Found"]}
{"handler": "attempt_handler", "params": {"name": "Max", "task": "Snake"}, "response": [200, "Good try, Max! Your attempt on task 'Snake' is accepted!"]}

from json import dumps

log_entry_en = {
    "handler": "greet",
    "params": {"name": "Max", "age": 18},
    "response": "Hello, Max! You are 18 years old."
}

print(dumps(log_entry_en))

{"handler": "greet", "params": {"name": "Max", "age": 18}, "response": "Hello, Max! You are 18 years old."}

log_entry_ru = {
    "handler": "greet",
    "params": {"name": "Алиса", "age": 25},
    "response": "Привет, Алиса! Тебе 25 лет."
}

print(dumps(log_entry_ru, ensure_ascii=False))

{"handler": "greet", "params": {"name": "Алиса", "age": 25}, "response": "Привет, Алиса! Тебе 25 лет."}

{"handler": "greet", "params": {"name": "\u0410\u043b\u0438\u0441\u0430", "age": 25}, "response": "\u041f\u0440\u0438\u0432\u0435\u0442, \u0410\u043b\u0438\u0441\u0430! \u0422\u0435\u0431\u0435 25 \u043b\u0435\u0442."}

{"handler": "attempt_handler", "params": {"name": "Max", "task": "Snake"}, "response": [200, "Good try, Max! Your attempt on task 'Snake' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Ivan", "task": "Queens"}, "response": [200, "Good try, Ivan! Your attempt on task 'Queens' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Max", "task": "TwoSum"}, "response": [200, "Good try, Max! Your attempt on task 'TwoSum' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Alex", "task": "Queens"}, "response": [200, "Good try, Alex! Your attempt on task 'Queens' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Mari", "task": "Parenthesis"}, "response": [200, "Good try, Mari! Your attempt on task 'Parenthesis' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Max", "task": "Parenthesis"}, "response": [200, "Good try, Max! Your attempt on task 'Parenthesis' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Alex", "task": "Snake"}, "response": [200, "Good try, Alex! Your attempt on task 'Snake' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Ivan", "task": "Snake"}, "response": [200, "Good try, Ivan! Your attempt on task 'Snake' is accepted!"]}

import json

json_string = '{"handler": "attempt_handler", "params": {"name": "Max", "task": "Snake"}, "response": [200, "Good try, Max! Your attempt on task \'Snake\' is accepted!"]}'
data = json.loads(json_string)

{
  'handler': 'attempt_handler',
  'params': {
    'name': 'Max',
    'task': 'Snake'
  },
  'response': [
    200,
    "Good try, Max! Your attempt on task 'Snake' is accepted!"
  ]
}

Если читаете этот текст — значит, у вас получилось справиться с задачами или возникли трудности. В любом случае вы взялись за эту задачу — а это похвально!

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

Ниже — код автора. Изучите его и сравните со своим решением. Или изучите основные идеи и попробуйте самостоятельно их реализовать. А дальше автор описал размышления о подходе к задачам 1–3 и разобрал каждый кусочек кода.

Загляните в блоки, к которым остались вопросы или в которых вы увидели новые элементы решения.

РазборРазбор

import sys


def home_handler(params):
    return 200, "Home Page"


def admin_handler(params):
    return 200, "Admin Panel"


def user_handler(params):
    # Должен использовать параметр name из URL
    return 200, f"Hello, {params['name']}!"


def attempt_handler(params):
    # Должен использовать параметр name из URL
    return (
        200,
        f"Good try, {params['name']}! Your attempt on task '{params['task']}' is accepted!",
    )


def not_found_handler(params):
    return 404, "Not Found"


# Функция маршрутизации
def router(path):
    if path == "/":
        return home_handler
    elif path.startswith("/user"):
        return user_handler
    elif path.startswith("/attempt"):
        return attempt_handler
    elif path.startswith("/admin"):
        return admin_handler

    return not_found_handler


def wsgi_server():
    """Обрабатывает запрос и возвращает ответ через yield"""
    request = yield None
    while True:
        # Парсинг запроса
        first_line = request.split("\r\n")[0]
        method, path = first_line.split()

        # Разделяем путь и параметры
        params = {}
        if "?" in path:
            query = path.split("?")[1]
            for pair in query.split("&"):
                if "=" in pair:
                    key, value = pair.split("=")
                    params[key] = value

        # Выбор handler'а
        handler = router(path)

        # Получение ответа (все сразу, без разделения на header/body)
        response = handler(params)
        request = yield response


# Имитация работы сервера
app = wsgi_server()
app.send(None)

for request in sys.stdin:
    print(app.send(request))

import sys
from functools import wraps
from json import dumps


def require_auth(func):
    """Декоратор проверки авторизации"""

    @wraps(func)
    def wrapper(params):
        if "name" not in params:
            return 403, "Forbidden"
        return func(params)

    return wrapper


def log_call(func):
    """Декоратор логирования"""

    @wraps(func)
    def wrapper(params):
        result = func(params)

        log_entry = {
            "handler": func.__name__,
            "params": params,
            "response": result,
        }
        print(dumps(log_entry))
        return result

    return wrapper


# Ручки для разных маршрутов
@log_call
def home_handler(params):
    return 200, "Home Page"


@log_call
def admin_handler(params):
    return 200, "Admin Panel"


@log_call
@require_auth
def user_handler(params):
    # Должен использовать параметр name из URL
    return 200, f"Hello, {params['name']}!"


@log_call
@require_auth
def attempt_handler(params):
    # Должен использовать параметр name из URL
    return (
        200,
        f"Good try, {params['name']}! Your attempt on task '{params['task']}' is accepted!",
    )


@log_call
def not_found_handler(params):
    return 404, "Not Found"


# Функция маршрутизации
def router(path):
    if path == "/":
        return home_handler
    elif path.startswith("/user"):
        return user_handler
    elif path.startswith("/attempt"):
        return attempt_handler
    elif path.startswith("/admin"):
        return admin_handler
    else:
        return not_found_handler


def wsgi_server():
    """Обрабатывает запрос и возвращает ответ через yield"""
    request = yield None
    while True:
        # Парсинг запроса
        first_line = request.split("\r\n")[0]
        method, path = first_line.split()

        # Разделяем путь и параметры
        params = {}
        if "?" in path:
            query = path.split("?")[1]
            for pair in query.split("&"):
                if "=" in pair:
                    key, value = pair.split("=")
                    params[key] = value

        # Выбор ручки
        handler = router(path)

        # Получение ответа (все сразу, без разделения на header/body)
        response = handler(params)
        request = yield response


# Имитация работы сервера
app = wsgi_server()
app.send(None)

for request in sys.stdin:
    app.send(request)

import sys
from json import loads
from collections import defaultdict

# читаем строки ДО выхода из with
with open("app.log", "r", encoding="utf-8") as f:
    lines = f.readlines()

# теперь файл можно закрыть, данные у нас
lines_to_dicts = map(lambda line: loads(line), lines)

attempt_logs = [
    log for log in lines_to_dicts
    if log.get("handler") == "attempt_handler"
]

if not attempt_logs:
    print("No attempt logs found")
    sys.exit(1)

unique_users = sorted({log["params"]["name"] for log in attempt_logs})

task_counts = defaultdict(int)
for log in attempt_logs:
    task_counts[log["params"]["task"]] += 1

most_popular_task = max(task_counts.items(), key=lambda item: item[1])

print("Analytics Results:")
print(f"Unique users who solved tasks: {len(unique_users)}")
print(f"Unique users list: {', '.join(unique_users)}")
print(f"Task popularity: {dict(sorted(task_counts.items()))}")
print(
    f"Most popular task: '{most_popular_task[0]}' with {most_popular_task[1]} attempts"
)

Задача № 1. WSGI-сервер как функцияЗадача № 1. WSGI-сервер как функция

Ключевое открытие — оказывается, генераторная функция может не только возвращать, но и получать данные на вход и с ними работать. Всё благодаря методу .send(), который мы рассмотрели в описании задачи.

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

Из каких частей состоит шаблонИз каких частей состоит шаблон

Шаблон состоит из четырёх частей. Разберём их подробнее.

Часть № 1: генератор запросовЧасть № 1: генератор запросов

Реализован с помощью метода .send(). Каждый новый запрос улетает в генераторную функцию wsgi_server().

# Имитация работы сервера
app = wsgi_server()
app.send(None)

for request in sys.stdin:
    print(app.send(request))

Часть № 2: генераторная функцияЧасть № 2: генераторная функция

Её нужно написать. Основная роль этой функции — находить нужного адресата-ручку и запускать его с конкретными параметрами.

# WSGI-сервер (генератор) — эту функцию нужно реализовать! 
def wsgi_server(requests):
    """Обрабатывает запросы и возвращает каждый из ответов через yield"""
    request = yield None #запуск генератора, приём первого запроса app.send(None) 
    while True:
    # Здесь необходимо написать обработку request и вызов нужной ручки для получения response  
        request = yield response

Часть № 3: функция-маршрутизаторЧасть № 3: функция-маршрутизатор

Можно её использовать, чтобы по заданному пути вызвать нужную ручку. Это вспомогательный помощник для генераторной функции.

# Функция маршрутизации
def router(path):
    if path == "/":
        return home_handler
    elif path.startswith("/user"):
        return user_handler
    elif path.startswith("/attempt"):
        return attempt_handler
    elif path.startswith("/admin"):
        return admin_handler

    return not_found_handler

Часть № 4: сами ручки (handler)Часть № 4: сами ручки (handler)

Перечень ручек, к которым обращается генераторная функция. В зависимости от запроса запустится одна из них.

# Ручки для разных маршрутов
def home_handler(params):
    return 200, "Home Page"


def admin_handler(params):
    return 200, "Admin Panel"


def user_handler(params):
    # Должен использовать параметр name из URL
    return 200, f"Hello, {params['name']}!"


def attempt_handler(params):
    # Должен использовать параметр name из URL
    return (
        200,
        f"Good try, {params['name']}! Your attempt on task '{params['task']}' is accepted!",
    )


def not_found_handler(params):
    return 404, "Not Found"

Шаги решения задачиШаги решения задачи

Получается, вам нужно собрать генераторную функцию, которая на вход получает строку вида:

[метод] [путь] [параметры]

Например, GET /user?name=John

• метод — в основном в задаче используется GET или POST, бывают ещё PUT, PATCH. Не будем вдаваться в подробности, что делает каждый метод: об этом мы расскажем в одном из дальнейших статей.
• путь — путь до ручки, как директория в проводнике. Например, просто / или /user.
• параметры — параметры, передаваемые в ручку для обработки, в формате ?Параметр_1=Значение_1&Параметр_2=Значение2. Например, ?name=John.

А на выходе выдаёт ответ от вызванных ручек.

То есть, чтобы решить задачу, нужно:

1. Вычленить из строки метод, путь и параметры.
2. Найти ручки, которые будем запускать.
3. Получить и вернуть ответ от нужной ручки с переданными параметрами.

Разберём каждый шаг.

Шаг № 1: вычленить из строки метод, путь и параметрыШаг № 1: вычленить из строки метод, путь и параметры

Для этого последовательно вызовем метод .split().

• Метод отделяется от остальной части с помощью пробела.
• Путь и параметры отделяются друг от друга с помощью символа ?.
• Сами различные параметры и их значения отделяются друг от друга с помощью символов & и =.

 def wsgi_server():
    """Обрабатывает запрос и возвращает ответ через yield"""
    request = yield None
    while True:
        # Парсинг запроса
        first_line = request.split('\r\n')[0]
        method, path = first_line.split()

        # Разделяем путь и параметры
        params = {}
        if '?' in path:
            query = path.split('?')[1]
            for pair in query.split('&'):
                if '=' in pair:
                    key, value = pair.split('=')
                    params[key] = value
        …

Шаг № 2: найти ручки, которые будете запускатьШаг № 2: найти ручки, которые будете запускать

Это можно сделать с помощью уже прописанной функции router().

def wsgi_server():
    """Обрабатывает запрос и возвращает ответ через yield"""
    request = yield None
    while True:
        # Парсинг запроса
       …

        # Выбор ручки
        handler = router(path)

        …

Шаг № 3: получить и вернуть ответ от нужной ручки с переданными параметрамиШаг № 3: получить и вернуть ответ от нужной ручки с переданными параметрами

Для этого запустим функцию, найденную router.

def wsgi_server():
    """Обрабатывает запрос и возвращает ответ через yield"""
    request = yield None
    while True:
        # Парсинг запроса
        … 

        # Выбор ручки
        handler = router(path)

        # Получение ответа
        response = handler(params)
        request = yield response

…

Про тестированиеПро тестирование

В рамках задачи не просят обрабатывать крайние значения и уверяют, что обязательно будут нужные параметры в нужном виде, если ручкам они нужны.
Поэтому главное — организовать вывод всех ручек, с разными методами и избыточным количеством параметров.
Например, можно проверить корректность следующей цепочкой вызовов:

GET /
GET /admin
GET /user?name=John
GET /unknown
POST /attempt?name=Alex&task=queens
GET /attempt?name=Vasya&task=snake

Вывод будет:

(200, 'Home Page')
(200, 'Admin Panel')
(200, 'Hello, John!')
(404, 'Not Found')
(200, "Good try, Alex! Your attempt on task 'queens' is accepted!")
(200, "Good try, Vasya! Your attempt on task 'snake' is accepted!")

Что в итогеЧто в итоге

Давайте зафиксируем промежуточные итоги после первой задачи:

• Разобрались в чужом коде.
• Познакомились с работой генератора с помощью метода .send()
• Повторили использование .split()

Задача № 2. ДекораторыЗадача № 2. Декораторы

Ключевое открытие — можно использовать несколько декораторов вместе. Как устроено их взаимодействие, разобрали на странице с описанием задачи.

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

Из каких частей состоит шаблонИз каких частей состоит шаблон

Шаблон состоит из двух частей. Расскажем о каждой подробнее.

Часть № 1: блок из предыдущей задачиЧасть № 1: блок из предыдущей задачи

Генераторную функцию нужно вставить из первой задачи. Набор ручек, маршрутизатор и генератор запросов остались неизменны.

При этом декораторы логирования и авторизации (@log_call и @require_auth) уже расставлены для разных ручек. Расставлять их не нужно.

# Ручки для разных маршрутов
@log_call
def home_handler(params):
    return 200, "Home Page"


@log_call
def admin_handler(params):
    return 200, "Admin Panel"


@log_call
@require_auth
def user_handler(params):
    # Должен использовать параметр name из URL
    return 200, f"Hello, {params['name']}!"


@log_call
@require_auth
def attempt_handler(params):
    # Должен использовать параметр name из URL
    return (
        200,
        f"Good try, {params['name']}! Your attempt on task '{params['task']}' is accepted!",
    )


@log_call
def not_found_handler(params):
    return 404, "Not Found"

Часть № 2: функции декораторовЧасть № 2: функции декораторов

Их нам и нужно прописать.

def require_auth(func):
    """Декоратор проверки авторизации"""
    pass


def log_call(func):
    """Декоратор логирования"""
    pass

Декоратор проверки авторизации проверяет, есть ли имя пользователя (name) в параметрах, и возвращает код ошибки (403, “Forbidden”), если не находит параметр.

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

{"handler": "attempt_handler", "params": {"name": "Alex", "task": "Queens"}, "response": [200, "Good try, Alex! Your attempt on task 'Queens' is accepted!"]}

Шаги решения задачиШаги решения задачи

Задача сводится к определённой работе с параметрами вызываемых ручек.

• Для проверки авторизации — анализируем конкретный параметр и делаем проверку.
• Для логирования — собираем определённый набор данных из работы ручки и переданных параметров.

Разберём каждый шаг подробнее.

Шаг № 1: пишем декоратор проверки авторизацииШаг № 1: пишем декоратор проверки авторизации

На вход он берёт параметры функции и проходится по ним в поиске параметра name. Если не находит, то возвращает соответствующий ответ.

def require_auth(func):
    """Декоратор проверки авторизации"""

    @wraps(func)
    def wrapper(params):
        if "name" not in params:
            return 403, "Forbidden"
        return func(params)

    return wrapper

При этом обратите внимание, что в некоторых частях кода декоратор @require_auth будет входной функцией для декоратора логирования @log_call.

@log_call
@require_auth
def attempt_handler(params):
…

А для декоратора логирования @log_call важно, чтобы он получил метаданные первичной функции, так как они нужны будут ему для вывода. Поэтому не забывайте использовать декоратор для декораторов @wraps. Подробнее о нём мы рассказали в описании задачи.

Шаг № 2: пишем декоратор логированияШаг № 2: пишем декоратор логирования

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

def log_call(func):
    """Декоратор логирования"""

    @wraps(func)
    def wrapper(params):
        result = func(params)

        log_entry = {
            "handler": func.__name__,
            "params": params,
            "response": result,
        }
        print(dumps(log_entry))
        return result

    return wrapper

Чтобы случайно в func.__name__ не прилетело название другого декоратора, убеждаемся, что в них используется декоратор для декораторов @wraps. Рассмотрим, как будет работать декоратор в связке с декоратором проверки авторизации @require_auth на примере.

@log_call
@require_auth
def attempt_handler(params):
    # Должен использовать параметр name из URL
    return (
        200,
        f"Good try, {params['name']}! Your attempt on task '{params['task']}' is accepted!",
    )

Допустим, на вход подаётся следующий вызов:

POST /attempt?name=Alex&task=Queens

Генераторная функция wsgi_server() запустит вызов ручки attempt_handler.

• Декораторы срабатывают снизу вверх при оборачивании. Так что первым сработает декоратор @require_auth.
• Он найдёт параметр name и вернёт из ручки значение 200, "Good try, Alex! Your attempt on task 'Queens' is accepted!"
• Декоратор @log_call создаст лог, состоящий из следующих параметров:
  — handler — благодаря декоратору декораторов @wraps в @require_auth.
  — params — это переданные в ручку параметры.
  — response — это ответ от декоратора @require_auth.

На выходе получаем JSON в нужном формате:

{"handler": "attempt_handler", "params": {"name": "Alex", "task": "Queens"}, "response": [200, "Good try, Alex! Your attempt on task 'Queens' is accepted!"]}

Если бы входной запрос был таким:

POST /attempt?nam=Alex&task=Queens

Тогда бы декоратор @require_auth не нашёл параметр name в запросе и вернул значение 403, “Forbidden”.

А на выходе мы получили бы такой JSON:

{"handler": "attempt_handler", "params": {"nam": "Alex", "task": "Queens"}, "response": [403, "Forbidden"]}

Также обратите внимание на вывод JSON с помощью функции dumps(). Подробнее о ней рассказали в подсказках.

Про тестированиеПро тестирование

Надо проверить, что:

1. Для любой ручки получаем корректный response при наличии одного или нескольких параметров в запросе.
2. Параметры вызываемой ручки передаются корректно.
3. Финальный вывод в JSON организован верно.

Так как первый пункт включает автоматически проверку второго и третьего пунктов, то достаточно прогнать через ввод тесты для вызовов всех возможных response с одним или несколькими параметрами на входе (где параметры ожидаются).

Например, можно проверить корректность следующей цепочкой вызовов:

GET /
GET /admin
GET /unknown
POST /attempt?name=Max&task=Frogs
POST /attempt?name=Alex&task=Queens
POST /user?name=Ivan
POST /user?name=Mari&group=158
POST /attempt?nam=Pasha
POST /attempt?nam=Grisha&task=Snakes
POST /user?nam=Petr
POST /user?nam=Sasha&group=104

Вывод будет:

{"handler": "home_handler", "params": {}, "response": [200, "Home Page"]}
{"handler": "admin_handler", "params": {}, "response": [200, "Admin Panel"]}
{"handler": "not_found_handler", "params": {}, "response": [404, "Not Found"]}
{"handler": "attempt_handler", "params": {"name": "Max", "task": "Frogs"}, "response": [200, "Good try, Max! Your attempt on task 'Frogs' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Alex", "task": "Queens"}, "response": [200, "Good try, Alex! Your attempt on task 'Queens' is accepted!"]}
{"handler": "user_handler", "params": {"name": "Ivan"}, "response": [200, "Hello, Ivan!"]}
{"handler": "user_handler", "params": {"name": "Mari", "group": "158"}, "response": [200, "Hello, Mari!"]}
{"handler": "attempt_handler", "params": {"nam": "Pasha"}, "response": [403, "Forbidden"]}
{"handler": "attempt_handler", "params": {"nam": "Grisha", "task": "Snakes"}, "response": [403, "Forbidden"]}
{"handler": "user_handler", "params": {"nam": "Petr"}, "response": [403, "Forbidden"]}
{"handler": "user_handler", "params": {"nam": "Sasha", "group": "104"}, "response": [403, "Forbidden"]}

Что в итогеЧто в итоге

Зафиксируем промежуточные итоги после второй задачи:

• Реализовали рабочую цепочку декораторов.
• Поняли, зачем нужен декоратор для декораторов @wraps()
• Изучили, как обращаться к метапараметрам функции.
• Ещё раз познакомились с выводом JSON через dumps().

Задача № 3. Лямбда-функцияЗадача № 3. Лямбда-функция

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

Шаги решения задачиШаги решения задачи

Это самая прямолинейная задача из всех трёх!

Нам нужно:

1. Считать данные.

2. Найти всё, что нас просят:

   2.1. логи только для attempt_handler;

   2.2. множество уникальных пользователей;

   2.3. количество решений для каждой задачи;

   2.4. самые популярные задачи.

3. Вывести всё собранное в нужном формате.

Разберём каждый шаг подробнее.

Шаг № 1: считать данныеШаг № 1: считать данные

Используем функцию loads(), которая преобразует строку в формате JSON в словарь. Подробнее про неё мы рассказали в подсказке.

import sys
from json import loads
from collections import defaultdict

with open("app.log", "r", encoding="utf-8") as f:
    lines = f.readlines()

lines_to_dicts = map(lambda line: loads(line), lines)

Давайте разберёмся, что тут происходит.

Первым делом считываем все строчки из файла app.log.

with open("app.log", "r", encoding="utf-8") as f:
    lines = f.readlines()

Затем сохраняем их в ленивый итератор lines_to dicts. Реализуем с помощью связки лямбда-функции и map().

lambda line: loads(line) — это лямбда-функция, которая преобразовывает строку в словарь:

# Было
{"handler":"attempt_handler","params":{"name":"Max","task":"Snake"},"response":[200,"Good try, Max! Your attempt on task 'Snake' is accepted!"]}

# Стало
{
  'handler': 'attempt_handler',
  'params': {
    'name': 'Max',
    'task': 'Snake'
  },
  'response': [
    200,
    "Good try, Max! Your attempt on task 'Snake' is accepted!"
  ]
}

А map() — итератор, который по одной строке применяет лямбда-функцию к строкам в lines.

Таким образом, в переменной lines_to_dicts лежит ленивый итератор, который можно развернуть по запросу. Самый простой способ такой: list(lines_to_dicts), получите список словарей.

Для оптимизации кода в этом кусочке используем лямбда-функцию:

lambda line: loads(line)

Альтернативно можно написать отдельную функцию, и тогда код выглядел бы так:

def parse(line):
    return loads(line)
lines_to_dicts = map(parse, sys.stdin)

Но здесь три строчки вместо одной и дополнительная сущность в виде именованной функции parse(). Налицо усложнение!

Шаг № 2: найти всё, что нас просятШаг № 2: найти всё, что нас просят

Пойдём по порядку.

Чтобы отфильтровать словари, которые соответствуют ручке attempt_handler, воспользуемся функцией filter(). Она принимает на вход функцию и коллекцию, а возвращает ленивый итератор с теми элементами, для которых функция дала True.

А далее с помощью функции list прогоняем ленивый итератор и получаем список словарей с нужной ручкой.

# Фильтруем логи только для attempt_handler
attempt_logs = list(
    filter(lambda log: log["handler"] == "attempt_handler", lines_to_dicts)
)

Чтобы создать множество уникальных пользователей, воспользуемся функцией set().

# Уникальные пользователи
unique_users = set(map(lambda log: log["params"]["name"], attempt_logs))

Что тут происходит:

• Проходимся по каждому словарю в отфильтрованном списке attempt_logs.
• С помощью лямбда-функции lambda log: log["params"]["name"] достаём имя пользователя.
• С помощью функции map() проходимся по каждому элементу из attempt_logs.
• Проходя по каждому элементу, добавляем его во множество с помощью функции set(). Это позволяет убрать повторы.

Здесь функция map() тоже играет роль оптимизации, заменяя работу цикла for. Альтернативно код можно написать так, но в таком виде получаем три строчки вместо одной и лишний цикл:

users = set()
for log in attempt_logs:
    users.add(log["params"]["name"])

Для подсчёта решений каждой задачи пройдёмся циклом по отфильтрованным логам и на каждой итерации будем учитывать название задачи.

from collections import defaultdict

# Подсчет задач
task_counts = defaultdict(int)
for log in attempt_logs:
    task_counts[log["params"]["task"]] += 1

Хранить всё будем в defaultdict(int) — это словарь, в котором при первом обращении к новому ключу он автоматически получает значение 0. Этот «улучшенный словарь» идеально подходит для роли счётчика

Таким образом, в task_counts лежит словарь с количеством решений по каждой задаче.

{
  "Snake": 3,
  "Queens": 2,
  "Parenthesis": 2,
  "TwoSum": 1
}

Подробнее про коллекцию defaultdict() и ещё несколько заготовленных коллекций из модуля collections можете прочитать в документации.

Для поиска самой популярной задачи воспользуемся функцией max().

# Самая популярная задача
most_popular_task = max(task_counts.items(), key=lambda item: item[1])

Функция max(iterable, key=...) работает следующим образом: на вход получает итерируемый объект и функцию key, по которой элементы будут сравниваться друг с другом.

То есть — передаём словарь task_counts и просим провести сравнение по второму параметру (по количеству).

Шаг № 3: вывести всё собранное в нужном форматеШаг № 3: вывести всё собранное в нужном формате

Обратите внимание: чтобы не плодить дополнительные переменные, базовые операции в виде подсчёта количества пользователей или сортировки списка можно сразу обозначать в выводе.

print("Analytics Results:")
print(f"Unique users who solved tasks: {len(unique_users)}")
print(f"Unique users list: {', '.join(sorted(unique_users))}")
print(f"Task popularity: {dict(sorted(task_counts.items()))}")
print(
    f"Most popular task: '{most_popular_task[0]}' with {most_popular_task[1]} attempts"
)

Про тестированиеПро тестирование

В условии указано, что граничные случаи (нет строк с ручкой attempt_handler и есть несколько задач с максимальным количеством решений) пропускаем.

В таком случае нам нужно проверить, что всё правильно считается и сортируется.

Для этого собираем тесты, в которых:

1. Имена пользователей в логах идут не в лексикографическом порядке (проверить работу сортировки).
2. Названия задач идут не в лексикографическом порядке (проверить работу сортировки).
3. Есть пользователи, которые решили одну задачу, и есть пользователи, которые решили несколько задач (проверить работу отбора уникальных элементов).

Например, можно проверить корректность следующими логами:

{"handler": "attempt_handler", "params": {"name": "Max", "task": "Snake"}, "response": [200, "Good try, Max! Your attempt on task 'Snake' is accepted!"]}
{"handler": "user_handler", "params": {"name": "Ivan"}, "response": [200, "Hello, Ivan!"]}
{"handler": "attempt_handler", "params": {"name": "Ivan", "task": "Queens"}, "response": [200, "Good try, Ivan! Your attempt on task 'Queens' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Max", "task": "TwoSum"}, "response": [200, "Good try, Max! Your attempt on task 'TwoSum' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Alex", "task": "Queens"}, "response": [200, "Good try, Alex! Your attempt on task 'Queens' is accepted!"]}
{"handler": "user_handler", "params": {"name": "Mari"}, "response": [403, "Forbidden"]}
{"handler": "attempt_handler", "params": {"name": "Mari", "task": "Parenthesis"}, "response": [200, "Good try, Mari! Your attempt on task 'Parenthesis' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Max", "task": "Parenthesis"}, "response": [200, "Good try, Max! Your attempt on task 'Parenthesis' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Alex", "task": "Snake"}, "response": [200, "Good try, Alex! Your attempt on task 'Snake' is accepted!"]}
{"handler": "attempt_handler", "params": {"name": "Ivan", "task": "Snake"}, "response": [200, "Good try, Ivan! Your attempt on task 'Snake' is accepted!"]}
{"handler": "not_found_handler", "params": {"name": "Mari"}, "response": [404, "Not Found"]}

На выходе получаем:

Analytics Results:
Unique users who solved tasks: 4
Unique users list: Alex, Ivan, Mari, Max
Task popularity: {'Parenthesis': 2, 'Queens': 2, 'Snake': 3, 'TwoSum': 1}
Most popular task: 'Snake' with 3 attempts

Что в итогеЧто в итоге

И вот итоги после третьей задачи:

• Познакомились со способами оптимизации с помощью лямбда-функции, defaultdict() и промежуточных расчётов прямо в выводе.
• Научились считывать строчный JSON с помощью loads().

ИтогИтог

Если собрать воедино решения всех трёх задач, вы получите полноценный небольшой сервис, который:

1. Принимает запросы и запускает их обработку. При этом может принимать запросы вечно, не прекращая свою работу.
2. Логирует действия пользователей при работе с сервисом и проверяет наличие авторизации пользователя.
3. Проводит небольшую аналитику логов, чтобы понять, как сервисом пользуются.

Это конструкция, которая туда-сюда переносит по системе запросы, следит за правильностью выполнения и изучает поведение пользователей.

Надеемся, что у вас получилось на кончиках пальцев почувствовать это прямо в разработке. Ещё раз хотим вас похвалить — это большой шаг!

Также давайте подведём итоги: что изучили и что можно взять с собой дальше.

В рамках этих задач вы освоили:

• Создание генераторной функции. Которая не только отдаёт, но и принимает данные за счёт метода .send().
• Функциональную работу декораторов. Разработали цепочку декораторов, научились доставать метаданные функции, а также передавать их через цепочку с помощью декоратора для декораторов @wraps.
• Применение лямбда-функций для базовых операций работы над коллекциями. Использовали их при построчном считывании информации, фильтрации, подсчёте определённых параметров.
• Использование функции defaultdict(int) для упрощения кода в задачах с подсчётами.
• Считывание строчного формата JSON с помощью функции dumps().

Напоследок хотим посоветовать одну вещь: чаще задумывайтесь, как сделать свой код более лаконичным и оптимальным.

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

На этом освоение раздела завершено. Не устаём повторять: вы очень большой молодец! Продолжайте в том же духе, впереди нас ждёт важная и интересная тема: научимся работать с классами и методами.

Увидимся в следующем разделе!