ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Реализовать Session Management с TTL

1. Цель задачи

Научиться проектировать и реализовывать систему управления сессиями с автоматическим истечением срока действия (TTL) с использованием Redis. Основной фокус — правильно настроить время жизни ключа (1 час) и убедиться, что сессия гарантированно удаляется не позднее чем через 65 минут (учёт возможных задержек удаления). В результате вы получите готовый модуль для управления сессиями, пригодный для интеграции в микросервисную архитектуру.

Ключевой результат Работающая реализация SessionManager с Redis, которая создаёт сессии с TTL = 1 час, и автоматическое удаление ключа подтверждено экспериментально в течение 65 минут.

2. Исходные данные

Если нет реального инструмента — симулируем:

1. Если нет доступа к Docker — используем Redis в памяти (fake redis через fakeredis): pip install fakeredis
2. Для теста TTL на fakeredis нужно эмулировать время (заморозить time.time). Но для реальной проверки удаления через 65 минут обязательно использовать живой Redis (через Docker или локально).

3. Технологический стек

4. Этапы выполнения

Этап 1: Настройка окружения и прототип подключения (30 минут)

Действия

1. Запустить Redis (если локально):

   docker run --name session-redis -p 6379:6379 -d redis:7-alpine
   

   Проверить через redis-cli ping → PONG.

2. Создать Python-проект:

   • mkdir session_manager && cd session_manager
   • python -m venv venv && source venv/bin/activate
   • Установить зависимости: pip install redis pytest pytest-timeout

3. Создать файл session_manager.py с базовым классом:

   import redis
   import json
   import uuid
   from datetime import datetime
   
   class SessionManager:
       def __init__(self, redis_url="redis://localhost:6379/0"):
           self.redis = redis.from_url(redis_url, decode_responses=True)
       
       def create_session(self, user_id, ttl=3600):
           session_id = str(uuid.uuid4())
           now = datetime.utcnow().isoformat()
           session_data = json.dumps({
               "session_id": session_id,
               "user_id": user_id,
               "created_at": now,
           })
           self.redis.setex(f"session:{session_id}", ttl, session_data)
           return session_id
   
       def get_session(self, session_id):
           data = self.redis.get(f"session:{session_id}")
           if data:
               return json.loads(data)
           return None
   
       def delete_session(self, session_id):
           self.redis.delete(f"session:{session_id}")
   
       def close(self):
           self.redis.close()
   

4. Написать простой тест:

   import pytest
   from session_manager import SessionManager
   
   @pytest.fixture
   def sm():
       manager = SessionManager()
       yield manager
       manager.close()
   
   def test_create_and_get(sm):
       sid = sm.create_session("user_123")
       session = sm.get_session(sid)
       assert session["user_id"] == "user_123"
   

Ожидаемый результат этапа

• Redis запущен и отвечает
• Есть работающий класс SessionManager с методами create, get, delete
• Тест test_create_and_get проходит

Этап 2: Реализация корректного TTL (45 минут)

Действия

1. Проверить, что TTL устанавливается правильно:

   def test_ttl_set(sm):
       sid = sm.create_session("user_456", ttl=60)
       # Проверить, что TTL в Redis ≈ 60
       ttl = sm.redis.ttl(f"session:{sid}")
       assert ttl > 55 and ttl <= 60
   

2. Добавить поддержку обновления TTL (скользящее окно — опционально, но полезно):

   def refresh_session(self, session_id, new_ttl=3600):
       self.redis.expire(f"session:{session_id}", new_ttl)
   

3. Обработать крайние случаи:

   • Сессия не найдена: get_session возвращает None
   • Передача пустого session_id → вызывает ValueError
   • Истечение TTL во время запроса (не блокируем)

4. Проверить сериализацию: убедиться, что created_at имеет тип datetime после десериализации (добавить конвертацию).

   def get_session(self, session_id):
       data = self.redis.get(f"session:{session_id}")
       if data:
           obj = json.loads(data)
           obj["created_at"] = datetime.fromisoformat(obj["created_at"])
           return obj
       return None
   

Ожидаемый результат этапа

• TTL корректно устанавливается и проверяется
• Метод refresh_session работает
• Все граничные случаи покрыты юнит-тестами

Этап 3: Написание интеграционного теста на истечение TTL (1 час)

Действия

1. Создать тест, который проверяет удаление через заданное время (используется короткий TTL, например, 2 секунды):

   @pytest.mark.timeout(10)
   def test_ttl_expiration(sm):
       sid = sm.create_session("user_ttl", ttl=2)
       assert sm.get_session(sid) is not None
       import time
       time.sleep(2.5)  # ждём чуть больше TTL
       assert sm.get_session(sid) is None
   

2. Проверить механизм уведомления об истечении (опционально, через Keyspace notifications):

   • Включить в Redis: CONFIG SET notify-keyspace-events Ex
   • Подписаться на канал __keyevent@0__:expired
   • Проверить, что приходит событие

3. Написать долгий тест на 65 минут (только для production-валидации, отдельно):

   • Использовать pytest.mark.slow и не запускать по умолчанию
   • Создать сессию с TTL = 3600
   • Заснуть на 3900 секунд (65 минут)
   • Проверить, что сессии нет
   • Замерить время и зафиксировать, что удаление произошло между 3600 и 3900 секундами

Ожидаемый результат этапа

• Короткий тест на TTL работает (2-3 секунды)
• Долгий тест (необязательный) написан в коде и может быть запущен отдельно

Этап 4: Проверка признака успеха — сессия удаляется через 65 минут (30 минут выделено на подготовку, 65 минут на ожидание)

Действия

1. Написать скрипт validate_ttl_65min.py:

   import asyncio
   import time
   import logging
   from session_manager import SessionManager
   
   logging.basicConfig(level=logging.INFO)
   logger = logging.getLogger(__name__)
   
   def main():
       sm = SessionManager()
       sid = sm.create_session("user_65min", ttl=3600)
       logger.info(f"Created session {sid}, TTL=3600")
       start = time.time()
       while True:
           elapsed = time.time() - start
           session = sm.get_session(sid)
           if session is None:
               logger.info(f"Session deleted after {elapsed:.2f} seconds ({elapsed/60:.2f} minutes)")
               break
           if elapsed > 3900:
               logger.warning(f"Session still exists after {elapsed/60:.1f} minutes. TTL not honored?")
               break
           time.sleep(60)  # проверяем раз в минуту
       sm.close()
   
   if __name__ == "__main__":
       main()
   

2. Запустить скрипт и дождаться результата (можно в фоне или в screen).

3. Зафиксировать время удаления — при корректной реализации TTL должно произойти между 3600 и 3900 секундами.

Ожидаемый результат этапа

• Лог/отчёт с точным временем удаления сессии (например, "Session deleted after 3612.34 seconds")
• Подтверждение, что сессия удалена не позднее 65 минут

Этап 5: Рефакторинг и документирование (30 минут)

Действия

1. Добавить docstring ко всем публичным методам в формате Google/NumPy.
2. Добавить обработку ошибок (Redis connection error, timeout).
3. Покрыть код линтером (flake8, mypy) и исправить замечания.
4. Написать README.md, содержащее:
   • Описание модуля
   • quick start (запуск Redis, установка зависимостей)
   • пример использования
   • результаты валидации TTL
5. Создать Makefile для стандартных задач:

   test:
       pytest -v
   validate-ttl:
       python validate_ttl_65min.py
   

Ожидаемый результат этапа

• Код чистый, документированный
• README с инструкцией и результатом теста TTL
• Makefile для удобного запуска

5. Критерии приемки (Definition of Done)

• Реализован класс SessionManager с методами create_session, get_session, delete_session, refresh_session (опционально).
• Все методы используют Redis с префиксом session: и устанавливают TTL.
• TTL по умолчанию = 3600 секунд (1 час).
• Написаны юнит-тесты для каждого метода (минимум 5 тестов).
• Интеграционный тест с TTL = 2 секунды доказывает, что ключ исчезает после истечения.
• Скрипт validate_ttl_65min.py подтверждает, что сессия удаляется в промежутке от 3600 до 3900 секунд.
• Код документирован: docstring у публичных методов, README, Makefile.
• Код проходит статический анализ (flake8 с max-line-length=100, mypy строгий режим).
• Ни одна сессия не остаётся в Redis после удаления (проверка через SCAN в конце теста).
• Реализована обработка ошибок: падение Redis не приводит к зависанию, сессии не теряются (если Redis перезапустится, данные пропадут — это ожидаемо).

6. Ожидаемый результат

Основной артефакт Файл session_manager.py, реализующий управление сессиями с TTL через Redis.

Содержание файла

• Импорты
• Класс SessionManager с конструктором, методами CRUD, приватными методами валидации
• Возможность передать кастомный TTL (по умолчанию 3600)
• Код должен быть асинхронно-безопасным (можно версию async_session_manager.py — опционально)

Дополнительные результаты

• tests/test_session_manager.py — набор тестов (pytest)
• validate_ttl_65min.py — скрипт для долгой проверки
• README.md — документация
• Makefile
• Лог/файл validation_result.txt с результатом прогона скрипта (например, "Session deleted at 3615 seconds")

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

Примечание: Если выполняете впервые, заложите дополнительно 1 час на отладку Redis и изучение документации. Ожидание 65 минут можно совмещать с другими задачами.

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я создал Redis-клиент с корректным URL и параметрами (decode_responses=True).
• Я убедился, что setex устанавливает TTL ровно 3600 секунд для новой сессии.
• Я написал тест test_expiration_after_ttl, который ждёт 2.5 секунды и проверяет None.
• Я запустил validate_ttl_65min.py и дождался завершения — результат записан в файл.
• Я проверил, что после удаления сессии по TTL в Redis не остаётся ключей с этим session_id.
• Я задокументировал публичные методы и добавил типы (type hints).
• Я прочитал README от начала до конца и убедился, что другой разработчик сможет повторить.