Перейти к основному содержимому

5.02. Веб и API

В РАЗРАБОТКЕНЕ ДЛЯ НОВИЧКОВНЕ ОБЯЗАТЕЛЬНО
Разработчику Архитектору

Веб и API​

Экосистема веба в Python​

Python занимает устойчивую позицию среди ведущих языков для реализации серверной стороны веб-приложений и интеграционных решений. Это обусловлено не столько производительностью интерпретатора (где Python уступает компилируемым языкам), сколько зрелостью экосистемы, простотой чтения кода, богатством библиотек и широкой поддержкой сообщества. Веб-стек Python охватывает как микросервисную архитектуру, так и монолитные приложения, от внутренних инструментов до публичных API-платформ.

Важно сразу развести два смежных, но различных направления:

  1. Веб-разработка — создание приложений, взаимодействующих с пользователями посредством HTTP-протокола, чаще всего через браузер (HTML-интерфейс, формы, AJAX-запросы и т.п.).
  2. Работа с API — как на стороне провайдера (реализация собственного API), так и на стороне клиента (интеграция с внешними сервисами через их интерфейсы).

Оба направления используют один и тот же набор базовых технологий: HTTP, JSON, URI, методы запросов — но с разной семантикой и набором требований. Например, приложение с пользовательским интерфейсом требует управления состоянием (сессиями, куками), обработки файлов, валидации форм, работы с шаблонами. API же, по современным канонам, стремится к statelessness (отсутствию состояния), предсказуемости структуры ответов и строгому соблюдению HTTP-семантики.

В данной главе рассматриваются три ключевых аспекта:

  • создание веб-приложений на базе микрофреймворка Flask — как иллюстрация принципов построения HTTP-серверов на Python;
  • организация клиентской работы с внешними API через библиотеку requests;
  • отправка уведомлений по различным каналам — электронная почта и Telegram — как неотъемлемая часть практически любого серверного приложения.

Flask​

Flask — это микрофреймворк для создания веб-приложений на Python. Слово «микро» здесь не означает «неполноценный» — оно указывает на отсутствие жёстких требований к структуре проекта, отсутствие встроенного ORM, системы аутентификации или административного интерфейса «из коробки». Вместо этого Flask предоставляет минимальное, но расширяемое ядро, позволяющее разработчику гибко комбинировать компоненты под конкретную задачу.

Такой подход делает Flask идеальным инструментом для обучения, прототипирования, создания микросервисов и API, а также небольших внутренних приложений — особенно в тех случаях, когда не требуется «тяжёлая артиллерия» вроде Django.

Основы WSGI​

В основе Flask лежит WSGI — Web Server Gateway Interface. Это спецификация, описывающая стандарт взаимодействия между веб-сервером (например, Apache, Nginx, Gunicorn) и Python-приложением.

WSGI определяет, что приложение — это вызываемый объект (callable), принимающий два аргумента:

  • environ — словарь, содержащий переменные окружения, заголовки запроса, метод, URI и другие метаданные;
  • start_response — callable, который приложение вызывает для передачи HTTP-статуса и заголовков ответа.

Пример минимального WSGI-приложения (без использования Flask):

def application(environ, start_response):
    status = '200 OK'
    headers = [('Content-Type', 'text/plain; charset=utf-8')]
    start_response(status, headers)
    return [b'Hello from WSGI!']

Запустить такое приложение можно через встроенный сервер wsgiref.simple_server — но только для целей отладки. В продакшене используются серверы вроде Gunicorn или uWSGI, которые поддерживают многопроцессность, таймауты, логирование и другие production-требования.

Flask берёт на себя реализацию WSGI-интерфейса: объект Flask сам является callable, и его можно передать в WSGI-сервер напрямую. При этом разработчик не работает с environ напрямую — вместо этого Flask предоставляет высокоуровневые абстракции: request, session, g, current_app, которые уже содержат разобранные и удобные для использования данные.

Создание простого сервера​

Начнём с канонического примера — «Hello, World!» на Flask.

from flask import Flask

app = Flask(__name__)

@app.route('/')
def index():
    return 'Привет, Вселенная IT!'

if __name__ == '__main__':
    app.run(debug=True)

Этот код делает следующее:

  • Создаёт экземпляр приложения Flask. Аргумент __name__ используется Flask для определения корневого пути приложения — это необходимо для поиска шаблонов, статических файлов и относительных импортов.
  • Декоратор @app.route('/') связывает URI-путь / с функцией index. Такой механизм называется маршрутизацией (routing).
  • При вызове app.run(debug=True) запускается встроенный сервер Werkzeug — надёжный и функциональный сервер для разработки, но не предназначенный для использования в production.

debug=True включает автоматическую перезагрузку при изменении кода и интерактивный отладчик при возникновении исключения. Это чрезвычайно удобно при разработке, но представляет серьёзную уязвимость в production (возможность выполнения произвольного кода через отладчик). Поэтому в продакшене debug всегда выключен, а приложение запускается через Gunicorn/uvicorn и проксируется через Nginx.

Маршрутизация и обработка GET/POST​

Маршрутизация в Flask основана на сопоставлении URI-путей с функциями-обработчиками (view functions). Помимо статических путей вроде /about, Flask поддерживает динамические сегменты:

@app.route('/user/<username>')
def show_user_profile(username):
    return f'Профиль пользователя: {username}'

Здесь <username> — это плейсхолдер, значение которого передаётся в функцию как аргумент. Тип сегмента можно ограничить: <int:user_id>, <uuid:token>, <path:subpath> — что обеспечивает валидацию на уровне маршрутизатора.

Более того, один и тот же путь может обрабатывать разные HTTP-методы:

from flask import request

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        # ... проверка учётных данных
        return f'Вход выполнен: {username}'
    else:
        return '''
            <form method="post">
                <input name="username" placeholder="Имя"><br>
                <input name="password" type="password" placeholder="Пароль"><br>
                <button type="submit">Войти</button>
            </form>
        '''

Объект request — глобальный, но потокобезопасный контекстный объект, содержащий данные текущего запроса: метод, заголовки, тело (form, json, files), куки, параметры строки запроса (args). Например:

  • request.args.get('page') — извлечение ?page=2;
  • request.json — разобранный JSON из тела POST-запроса (при Content-Type: application/json);
  • request.files['avatar'] — загруженный файл.

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

Шаблоны, статика, сессии​

Хотя Flask часто используется для API, он изначально предполагает поддержку рендеринга HTML через механизм шаблонов. По умолчанию используется шаблонизатор Jinja2, который позволяет встраивать логику (условия, циклы) и выражения в HTML, при этом сохраняя чёткое разделение между кодом и представлением.

Пример шаблона templates/profile.html:

<!DOCTYPE html>
<html>
<head><title>Профиль {{ user.name }}</title></head>
<body>
  <h1>Добро пожаловать, {{ user.name }}!</h1>
  {% if user.is_admin %}
    <p>Вы — администратор.</p>
  {% endif %}
  <ul>
  {% for role in user.roles %}
    <li>{{ role }}</li>
  {% endfor %}
  </ul>
</body>
</html>

Рендеринг в приложении:

from flask import render_template

@app.route('/profile/<int:user_id>')
def profile(user_id):
    user = get_user_by_id(user_id)  # условная функция
    return render_template('profile.html', user=user)

Ключевые особенности Jinja2:

  • экранирование HTML по умолчанию (защита от XSS);
  • наследование шаблонов ({% extends 'base.html' %}) — позволяет вынести общую структуру (шапку, подвал) в базовый шаблон;
  • фильтры ({{ name|upper }}, {{ price|round(2) }});
  • макросы — переиспользуемые блоки шаблонного кода.

Статические файлы — CSS, JS, изображения — размещаются в папке static/. Flask автоматически обслуживает их по пути /static/..., например:

<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">

Функция url_for() генерирует URL по имени маршрута или статического ресурса. Это предпочтительнее жёстко прописанных путей: при изменении структуры URL достаточно изменить только декоратор @route, а не все ссылки в шаблонах.

Сессии позволяют сохранять данные между запросами. Flask хранит сессию в подписанной (signed) cookie — то есть данные сериализуются, шифруются HMAC-ключом и отправляются клиенту. Сервер не хранит состояние — только секретный ключ для проверки подписи. Это обеспечивает безопасность от подделки (при условии, что SECRET_KEY неизвестен атакующему), но накладывает ограничения на объём данных (ограничение размера cookie — ~4 КБ).

from flask import session

app.secret_key = 'очень-секретный-ключ-никому-не-говорите'  # обязательно!

@app.route('/login', methods=['POST'])
def login():
    session['user_id'] = 123
    return 'Вы вошли'

@app.route('/profile')
def profile():
    user_id = session.get('user_id')
    if not user_id:
        return 'Неавторизован', 401
    return f'Профиль пользователя {user_id}'

Для хранения больших объёмов данных или при требовании server-side sessions (например, для logout всех сессий) используются расширения: Flask-Session, Flask-Login с поддержкой Redis, Memcached или баз данных.

REST API​

Flask одинаково хорошо подходит как для full-stack приложений, так и для реализации RESTful API. Для этого достаточно возвращать из обработчиков структурированные данные в формате JSON.

Минимальный API-эндпоинт:

from flask import jsonify

@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    user = {
        'id': user_id,
        'name': 'Алиса',
        'email': 'alice@example.com'
    }
    return jsonify(user)

Функция jsonify() делает три вещи:

  1. сериализует объект Python (dict, list, str, int и т.п.) в JSON;
  2. устанавливает заголовок Content-Type: application/json;
  3. гарантирует, что ответ возвращается с кодом 200 (если явно не указано иное).

Для более сложных сценариев (валидация входных данных, сериализация моделей, обработка ошибок) используются расширения, например Flask-RESTful или Flask-apispec (интеграция с OpenAPI/Swagger). Однако даже без них можно построить соответствующий REST-канонам API, если соблюдать следующие принципы:

  • идемпотентность: GET, PUT, DELETE должны давать предсказуемый результат при повторных вызовах;
  • статус-коды: 200 — успешное выполнение, 201 — создано (POST), 204 — нет содержимого (DELETE), 400 — ошибка клиента, 401/403 — аутентификация/авторизация, 404 — не найдено, 500 — ошибка сервера;
  • URI должны именовать ресурсы: /users, /users/123, /users/123/posts, а не /getUser, /deleteUser;
  • тело запроса для создания/изменения — JSON в application/json.

Пример полноценного CRUD-эндпоинта:

users = {}  # временное хранилище

@app.route('/api/users', methods=['POST'])
def create_user():
    data = request.get_json()
    if not data or 'name' not in data:
        return jsonify({'error': 'Поле name обязательно'}), 400

    user_id = max(users.keys(), default=0) + 1
    user = {'id': user_id, 'name': data['name'], 'email': data.get('email')}
    users[user_id] = user
    return jsonify(user), 201

@app.route('/api/users/<int:user_id>', methods=['GET'])
def read_user(user_id):
    user = users.get(user_id)
    if not user:
        return jsonify({'error': 'Пользователь не найден'}), 404
    return jsonify(user)

@app.route('/api/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    if user_id not in users:
        return jsonify({'error': 'Пользователь не найден'}), 404
    data = request.get_json()
    users[user_id].update({
        'name': data.get('name', users[user_id]['name']),
        'email': data.get('email', users[user_id]['email'])
    })
    return jsonify(users[user_id])

@app.route('/api/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    if user_id not in users:
        return jsonify({'error': 'Пользователь не найден'}), 404
    del users[user_id]
    return '', 204

Обратите внимание: даже при ошибке (400, 404) мы возвращаем JSON — это ожидаемое поведение для API-клиентов. Пустой ответ с 204 при удалении также соответствует рекомендациям RFC 7231.

Работа с API​

В современной инфраструктуре крайне редко встречаются полностью изолированные приложения. Даже внутренние сервисы взаимодействуют друг с другом через API. Python, благодаря своей выразительной синтаксической лёгкости и зрелой экосистеме, стал де-факто стандартом для написания интеграционных скриптов, ETL-процессов, автоматизации и микросервисов, обращающихся к сторонним системам.

Ключевой инструмент в этой области — библиотека requests. Её появление в 2011 году стало поворотным моментом: до этого стандартный urllib требовал многословного и непрозрачного кода для даже самых простых задач. requests ввела принцип «HTTP for Humans» — интуитивный, последовательный, отказоустойчивый интерфейс, ставший эталоном для библиотек во многих языках.

HTTP-запросы: requests​

Основной объект в requests — функции высшего уровня (get, post, put, delete и др.), каждая из которых инкапсулирует выполнение одного HTTP-запроса и возвращает объект Response. Рассмотрим базовый сценарий:

import requests

response = requests.get('https://api.github.com/users/octocat')
print(response.status_code)      # → 200
print(response.headers['Content-Type'])  # → application/json; charset=utf-8
print(response.text[:200])       # первые 200 символов тела ответа

Обратите внимание на несколько важных моментов:

  1. Автоматическое управление соединением. В отличие от urllib, requests не требует ручного открытия/закрытия соединений. Под капотом используется пул соединений через urllib3, что повышает производительность при множественных запросах к одному хосту.
  2. Кодировка определяется автоматически — на основе заголовка Content-Type и анализа содержимого (response.encoding можно переопределить, если обнаружена ошибка).
  3. Тело ответа доступно в нескольких форматах: text (строка), content (байты), json() (разобранный JSON), iter_content() (для потоковой обработки больших ответов).

Однако такой «быстрый старт» скрывает риски. В production-коде недопустимо оставлять запросы без явного указания таймаутов.

# Плохо: запрос может повиснуть на неопределённое время
requests.get('https://slow-service.example.com')

# Хорошо: явно задаём лимиты
response = requests.get(
    'https://api.example.com/data',
    timeout=(3.05, 27)  # (connect_timeout, read_timeout) в секундах
)

Таймауты — это обязательный элемент отказоустойчивости. Если внешний сервис не отвечает, ваше приложение не должно блокироваться. Рекомендуется использовать разные значения для подключения (обычно 1–5 сек) и чтения (зависит от ожидаемого объёма данных: 10–60 сек для больших выгрузок). Значение None (бесконечный таймаут) допустимо только в offline-скриптах, где отказ внешней системы не критичен.

Для повторяющихся запросов к одному API рекомендуется использовать объект Session:

session = requests.Session()
session.headers.update({'User-Agent': 'IT-Universe-Client/1.0'})
session.auth = ('user', 'pass')  # или другой механизм аутентификации

# Все запросы через session наследуют заголовки и учётные данные
response1 = session.get('https://api.example.com/v1/resource/1')
response2 = session.get('https://api.example.com/v1/resource/2')

Преимущества Session:

  • единый пул соединений для всех запросов (уменьшает накладные расходы на TLS-рукопожатия);
  • централизованное управление заголовками, аутентификацией, настройками (включая таймауты по умолчанию через session.request);
  • возможность подключения middleware через session.mount() (например, для автоматических повторов).

Авторизация: OAuth, токены, заголовки​

Большинство публичных API требуют идентификации клиента. Основные схемы:

  1. API-ключи — самый простой вариант. Ключ передаётся в заголовке или параметре запроса:

    headers = {'Authorization': 'Token abc123xyz'}
    # или
    params = {'api_key': 'abc123xyz'}
    requests.get(url, headers=headers)

  2. HTTP Basic Auth — учётные данные в заголовке Authorization: Basic base64(login:password). В requests:

    from requests.auth import HTTPBasicAuth
    requests.get(url, auth=HTTPBasicAuth('user', 'pass'))
    # или короче:
    requests.get(url, auth=('user', 'pass'))

  3. Bearer-токены (JWT, OAuth2 access tokens) — наиболее распространённый стандарт сегодня:

    headers = {'Authorization': 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...'}

  4. OAuth 2.0 — протокол делегированной авторизации. requests сам по себе не реализует OAuth, но интегрируется с requests-oauthlib, который поддерживает все grant types (authorization code, client credentials, refresh token и др.).

Частая ошибка — хранение токенов и ключей в коде. В production они должны извлекаться из переменных окружения, менеджеров секретов (HashiCorp Vault, AWS Secrets Manager) или конфигурационных файлов вне VCS. Для локальной разработки удобно использовать .env и python-dotenv:

# .env
GITHUB_TOKEN=ghp_abc123...

# app.py
import os
from dotenv import load_dotenv
load_dotenv()

headers = {'Authorization': f'Bearer {os.getenv("GITHUB_TOKEN")}'}

Важно: при использовании токенов, имеющих срок жизни (например, access token в OAuth2), необходимо предусматривать механизм обновления — либо через refresh token, либо через повторную аутентификацию. Это делается на уровне приложения, часто с кэшированием валидного токена в памяти.

Обработка ответов: JSON, статус-коды, исключения​

После получения Response необходимо выполнить несколько обязательных шагов:

  1. Проверить статус-код. Успешный HTTP-запрос — не всегда успешная бизнес-операция.

    if response.status_code == 200:
        data = response.json()
    elif response.status_code == 404:
        raise ResourceNotFoundError()
    elif response.status_code == 429:
        # Too Many Requests — нужно сделать паузу
        retry_after = int(response.headers.get('Retry-After', 60))
        time.sleep(retry_after)
    else:
        response.raise_for_status()  # бросит HTTPError для 4xx/5xx

    Метод response.raise_for_status() — удобный способ выбросить исключение requests.HTTPError для всех неуспешных кодов. Однако он не различает 4xx (ошибка клиента) и 5xx (ошибка сервера), поэтому в продакшене часто требуется более тонкая обработка.

  2. Безопасное чтение JSON. Не все ответы содержат валидный JSON — даже при коде 200. Например, при перегрузке сервер может вернуть HTML-страницу с сообщением об ошибке. Поэтому:

    try:
        data = response.json()
    except ValueError as e:  # в requests 2.x это requests.JSONDecodeError, наследник ValueError
        logging.error(f"Невозможно разобрать JSON: {response.text[:500]}")
        raise

  3. Валидация структуры ответа. Даже если JSON корректен, он может не соответствовать ожидаемой схеме (например, поле email отсутствует). Здесь помогают схемные валидаторы:

    from pydantic import BaseModel, HttpUrl
    from typing import List

    class User(BaseModel):
        id: int
        login: str
        avatar_url: HttpUrl
        email: str | None = None

    try:
        user = User.model_validate(data)
    except ValidationError as e:
        logging.error(f"Некорректная структура пользователя: {e}")

Пример: получение данных с GitHub API​

Рассмотрим комплексный пример — получение списка репозиториев пользователя GitHub с обработкой пагинации, ошибок и таймаутов.

import requests
import logging
import time
from typing import List, Dict, Any

logger = logging.getLogger(__name__)

def fetch_github_repos(username: str, token: str = None) -> List[Dict[str, Any]]:
    """
    Получает все публичные репозитории пользователя GitHub.
    Поддерживает пагинацию (до 1000 репозиториев — лимит GitHub).
    Возвращает список словарей с ключами: name, description, html_url, stargazers_count.
    """
    base_url = f'https://api.github.com/users/{username}/repos'
    headers = {'Accept': 'application/vnd.github.v3+json'}
    if token:
        headers['Authorization'] = f'Bearer {token}'

    session = requests.Session()
    session.headers.update(headers)

    repos = []
    page = 1
    per_page = 100  # максимум, разрешённый GitHub

    while True:
        try:
            response = session.get(
                base_url,
                params={'page': page, 'per_page': per_page},
                timeout=(5, 30)
            )

            # Явная обработка статусов
            if response.status_code == 403:
                # Превышен лимит запросов (Rate Limit)
                reset_time = int(response.headers.get('X-RateLimit-Reset', 0))
                sleep_duration = max(reset_time - time.time(), 60)
                logger.warning(f"Rate limit exceeded. Пауза на {sleep_duration:.0f} сек.")
                time.sleep(sleep_duration)
                continue  # повторить тот же запрос
            elif response.status_code == 404:
                raise ValueError(f"Пользователь '{username}' не найден на GitHub.")
            elif response.status_code != 200:
                response.raise_for_status()  # выбросит исключение для 5xx и других 4xx

            page_data = response.json()
            if not page_data:  # пустой ответ — конец пагинации
                break

            # Извлекаем только нужные поля
            for repo in page_data:
                repos.append({
                    'name': repo['name'],
                    'description': repo.get('description'),
                    'html_url': repo['html_url'],
                    'stargazers_count': repo.get('stargazers_count', 0)
                })

            # Проверяем, есть ли следующая страница (заголовок Link)
            link_header = response.headers.get('Link')
            if not link_header or 'rel="next"' not in link_header:
                break
            page += 1

        except requests.RequestException as e:
            logger.error(f"Ошибка при запросе к GitHub (страница {page}): {e}")
            raise

    return repos

Ключевые особенности реализации:

  • Использование Session для единообразия заголовков и эффективности соединений;
  • Явная обработка Rate Limit через заголовки X-RateLimit-* — критически важно для публичных API;
  • Пагинация через параметры и заголовок Link (стандарт RFC 5988);
  • Логирование ошибок с контекстом (номер страницы);
  • Выборка только необходимых полей — минимизация зависимости от структуры ответа;
  • Отказ от response.raise_for_status() в «сыром» виде — вместо этого — кейс-обработка для 403 и 404.

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

Отправка email и уведомлений​

Концептуально, отправка электронного письма — это клиент-серверное взаимодействие по протоколу SMTP (Simple Mail Transfer Protocol). Python предоставляет стандартную библиотеку smtplib, реализующую клиентскую часть протокола. Однако сам по себе SMTP — лишь транспорт. Для корректной доставки необходимо соблюсти множество дополнительных требований: форматирование заголовков, кодирование тела, поддержка вложений, аутентификация, шифрование соединения и совместимость с антиспам-фильтрами.

smtplib: отправка писем​

Минимальный пример отправки текстового письма через публичный SMTP-сервер (например, Gmail):

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

# Подготовка сообщения
msg = MIMEMultipart()
msg['From'] = 'sender@gmail.com'
msg['To'] = 'recipient@example.com'
msg['Subject'] = 'Тестовое письмо из Вселенной IT'

body = 'Это простое текстовое сообщение.'
msg.attach(MIMEText(body, 'plain', 'utf-8'))

# Соединение и отправка
try:
    with smtplib.SMTP('smtp.gmail.com', 587) as server:
        server.starttls()          # Включение шифрования
        server.login('sender@gmail.com', 'app_password_here')
        server.send_message(msg)
except smtplib.SMTPException as e:
    print(f'Ошибка SMTP: {e}')

Разберём ключевые компоненты:

  1. Формирование письма
    Простая строка недостаточна: письмо — это структура с заголовками (From, To, Subject, Date, Message-ID) и телом, возможно, в нескольких форматах (текст + HTML). Классы email.mime.* позволяют строить такие структуры в соответствии со стандартами RFC 5322 и RFC 2045–2049 (MIME).

    • MIMEMultipart: контейнер для нескольких частей (например, text/plain и text/html);
    • MIMEText: часть с текстом; параметры — содержимое, subtype (plain или html), кодировка;
    • MIMEImage, MIMEApplication — для вложений.

  2. Шифрование соединения
    Порт 587 используется для SMTP Submission с последующим переходом на защищённое соединение через команду STARTTLS. Это предпочтительнее устаревшего нешифрованного порта 25 или порта 465 (SMTPS, устаревший, но ещё встречающийся).
    Метод starttls() должен вызываться до login() — иначе учётные данные будут переданы в открытом виде.

  3. Аутентификация
    Современные почтовые сервисы запрещают вход по паролю от аккаунта. Вместо этого используются:

    • Пароли приложений (Gmail, Yandex) — одноразовые токены, привязанные к конкретному клиенту;
    • OAuth 2.0 — более безопасный, но сложный в реализации метод (требуется requests-oauthlib и обработка flow).

  4. Обработка ошибок
    smtplib выбрасывает иерархию исключений, наследующих от SMTPException:

    • SMTPRecipientsRefused — сервер отверг получателя (например, несуществующий email);
    • SMTPSenderRefused — проблема с отправителем (неподтверждённый домен, спам-репутация);
    • SMTPDataError — ошибка при передаче тела письма.

    В production важно логировать текст ошибки и код ответа сервера (e.smtp_code, e.smtp_error), поскольку он позволяет точно диагностировать причину (например, 550 — недоставляемый адрес, 421 — временная перегрузка).

Персонализированные рассылки​

Массовая отправка требует дополнительных мер:

  • Throttling — ограничение скорости отправки, чтобы не превысить лимиты SMTP-сервера (например, 100 писем/час у бесплатного Gmail);
  • Обработка ошибок на уровне получателя — отказ одного адресата не должен останавливать всю рассылку;
  • Шаблонизация — динамическое формирование тела с подстановкой имён, ссылок, данных.

Пример шаблонизированной рассылки с отложенной повторной отправкой при временной ошибке:

import time
from email.utils import formataddr
from typing import List, Dict, Tuple

def send_bulk_emails(
    recipients: List[Dict[str, str]],  # [{'name': '...', 'email': '...'}, ...]
    subject: str,
    text_template: str,
    html_template: str | None = None,
    smtp_config: Dict[str, str] = None
) -> Tuple[int, List[str]]:
    """
    Отправляет персонализированные письма списку получателей.
    Возвращает (успешно_отправлено, список_ошибок).
    """
    successes = 0
    errors = []

    with smtplib.SMTP(smtp_config['host'], smtp_config['port']) as server:
        server.starttls()
        server.login(smtp_config['user'], smtp_config['password'])

        for rec in recipients:
            try:
                # Формируем письмо для конкретного получателя
                msg = MIMEMultipart('alternative')
                msg['From'] = formataddr(('Вселенная IT', smtp_config['from_email']))
                msg['To'] = formataddr((rec['name'], rec['email']))
                msg['Subject'] = subject

                text = text_template.format(name=rec['name'])
                msg.attach(MIMEText(text, 'plain', 'utf-8'))

                if html_template:
                    html = html_template.format(name=rec['name'])
                    msg.attach(MIMEText(html, 'html', 'utf-8'))

                server.send_message(msg)
                successes += 1

                # Пауза между письмами (например, 1 сек — 3600 писем/час)
                time.sleep(smtp_config.get('delay', 1.0))

            except smtplib.SMTPRecipientsRefused as e:
                # Постоянная ошибка для этого получателя — фиксируем и идём дальше
                errors.append(f"{rec['email']}: {e.smtp_code} {e.recipients}")
            except smtplib.SMTPServerDisconnected:
                # Сервер разорвал соединение — переподключаемся
                server.connect(smtp_config['host'], smtp_config['port'])
                server.starttls()
                server.login(smtp_config['user'], smtp_config['password'])
                # Повторяем попытку для этого получателя (можно добавить счётчик)
            except Exception as e:
                errors.append(f"{rec['email']}: {type(e).__name__}: {e}")

    return successes, errors

Важные замечания:

  • Использование formataddr() гарантирует корректное экранирование имён (например, если в name есть запятые или кавычки);
  • MIMEMultipart('alternative') указывает почтовому клиенту выбрать наиболее подходящий формат (обычно HTML, если поддерживается);
  • Повторное подключение при SMTPServerDisconnected — стандартная практика для long-running рассылок.

Для высоконагруженных сценариев рекомендуется выносить отправку в фоновую очередь (Celery, RQ) и использовать dedicated email-сервисы (SendGrid, Mailgun, Amazon SES), которые предоставляют API, аналитику и встроенную обработку отписок/bounces.

Telegram-боты: python-telegram-bot​

Telegram API предлагает два режима взаимодействия:

  • Polling — клиент регулярно опрашивает сервер (getUpdates);
  • Webhook — сервер Telegram отправляет обновления на ваш endpoint по HTTPS.

Для уведомлений чаще всего используется прямая отправка сообщений через sendMessage, без обработки входящих команд — то есть в режиме «бот как клиент уведомлений». Для этого достаточно знать chat_id получателя и токен бота.

Библиотека python-telegram-bot (v20+) предоставляет высокоуровневый асинхронный API поверх Telegram Bot API.

Установка:

pip install python-telegram-bot --pre  # для v20+ (асинхронная версия)

Пример отправки текстового уведомления:

from telegram import Bot
import asyncio

async def send_telegram_alert(token: str, chat_id: int, message: str):
    bot = Bot(token=token)
    try:
        await bot.send_message(
            chat_id=chat_id,
            text=message,
            parse_mode='HTML',  # поддержка базовой HTML-разметки
            disable_web_page_preview=True
        )
    except Exception as e:
        logger.error(f"Ошибка отправки в Telegram: {e}")
        raise

Ключевые возможности:

  • HTML- и MarkdownV2-разметка — выделение, ссылки, код;
  • Клавиатуры — inline-кнопки (InlineKeyboardMarkup) для интерактивных уведомлений (например, «Подтвердить», «Отменить»);
  • Вложения — отправка изображений (send_photo), документов (send_document), голосовых сообщений;
  • Ограничения Telegram — 30 сообщений/сек на chat, 1 сообщение/сек на пользователя в личных чатах. При превышении — ошибка RetryAfter.

Для получения chat_id можно:

  1. Написать боту /start;
  2. Отправить POST-запрос к https://api.telegram.org/bot<TOKEN>/getUpdates и извлечь message.chat.id из ответа.

Пример: уведомление об ошибках​

Объединим рассмотренные подходы в реалистичный сценарий: мониторинг критических исключений в приложении и оповещение администратора.

import logging
import traceback
from typing import Optional

# Конфигурация — из переменных окружения
SMTP_CONFIG = {
    'host': os.getenv('SMTP_HOST'),
    'port': int(os.getenv('SMTP_PORT', 587)),
    'user': os.getenv('SMTP_USER'),
    'password': os.getenv('SMTP_PASSWORD'),
    'from_email': os.getenv('ALERT_FROM_EMAIL')
}

TELEGRAM_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
TELEGRAM_CHAT_ID = int(os.getenv('ADMIN_TELEGRAM_CHAT_ID'))

class AlertHandler(logging.Handler):
    """Кастомный лог-хендлер для отправки критических ошибок."""
    
    def emit(self, record: logging.LogRecord):
        if record.levelno < logging.ERROR:
            return

        # Формируем текст уведомления
        error_summary = f"КРИТИЧЕСКАЯ ОШИБКА в {record.module}:{record.funcName}\n"
        error_summary += f"Уровень: {record.levelname}\n"
        error_summary += f"Сообщение: {record.getMessage()}\n"

        if record.exc_info:
            error_summary += "\nТрассировка:\n"
            error_summary += ''.join(traceback.format_exception(*record.exc_info))[-1000:]  # последние 1000 символов

        # Отправка в Telegram (асинхронно, но в sync-контексте — через asyncio.run)
        try:
            asyncio.run(
                send_telegram_alert(TELEGRAM_TOKEN, TELEGRAM_CHAT_ID, 
                                   f"<b>🚨 Авария</b>\n<pre>{error_summary[:4000]}</pre>")
            )
        except Exception as e:
            # Если Telegram недоступен — fallback на email
            self._send_email_fallback(error_summary)

    def _send_email_fallback(self, body: str):
        try:
            msg = MIMEMultipart()
            msg['From'] = SMTP_CONFIG['from_email']
            msg['To'] = os.getenv('ADMIN_EMAIL')
            msg['Subject'] = '[CRITICAL] Ошибка в Вселенной IT'
            msg.attach(MIMEText(body, 'plain', 'utf-8'))

            with smtplib.SMTP(SMTP_CONFIG['host'], SMTP_CONFIG['port']) as server:
                server.starttls()
                server.login(SMTP_CONFIG['user'], SMTP_CONFIG['password'])
                server.send_message(msg)
        except Exception as e_mail:
            # Крайний fallback — запись в stderr
            print(f"ПОЛНЫЙ ОТКАЗ СИСТЕМЫ ОПОВЕЩЕНИЙ: {e_mail}", file=sys.stderr)

Подключение хендлера:

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
logger.addHandler(AlertHandler())

Такая архитектура обеспечивает:

  • Градуированную доставку: мгновенное оповещение в Telegram → резервное письмо → локальный лог;
  • Ограничение объёма: обрезка трассировки до 1000/4000 символов (ограничения Telegram/email-клиентов);
  • Изоляцию сбоев: сбой в одном канале не блокирует другой;
  • Интеграцию в стандартный logging-стек — ошибки можно логировать в файл, Sentry, Loki и одновременно отправлять администратору.