Framework для python Flask - Подписки на сигналы на основе декоратора

Старт:21.10.2024

Срок обучения:9 мес.

Python-разработчик

Практический онлайн-курс, на котором вы с нуля изучите самый универсальный и востребованный язык программирования — Python. Создадите свое портфолио разработчика, которое выгодно покажет вас на рынке труда, и сможете зарабатывать в IT через полгода.

136 000 ₽340 000 ₽

11 333₽/мес рассрочка

Подробнее

Для эффективной работы с событиями и сигналами в приложениях Flask рекомендуется использовать декоратор @app.signal. Это позволяет быстро и просто подписывать функции на определённые события, не усложняя код и не требуя создания сложной структуры.

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

from flask import Flask app = Flask(__name__) @app.signal('order_status_changed') def order_status_handler(order_id, new_status): # Логика обработки изменения статуса заказа print(f"Статус заказа {order_id} изменён на {new_status}") # ... (остальной код вашего приложения)

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

Framework для Python Flask - Подписки на сигналы на основе декоратора

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

Пример структуры:

import functools
from flask import Flask
app = Flask(__name__)
signals = {}
def signal(event_name):
def decorator(func):
if event_name not in signals:
signals[event_name] = []
signals[event_name].append(func)
return func
return decorator
@signal("user_created")
def handle_user_creation(user):
print(f"Пользователь {user} создан.")
@signal("user_updated")
def handle_user_update(user):
print(f"Пользователь {user} обновлен.")
@app.route('/create_user', methods=['POST'])
def create_user():
# Логика создания пользователя
user = 'Новый пользователь'
# Важно! Отправка сигнала!
for handler in signals.get("user_created", []):
handler(user)
return 'OK'
if __name__ == '__main__':
app.run(debug=True)

Ключевые моменты:

• Разделение логики обработки событий (функции `handle_user_creation`, `handle_user_update`) от основной логики (маршрут `/create_user`).
• Использование словаря `signals` для хранения подписчиков на события.
• Декоратор `signal` для регистрации функций-обработчиков. Он добавляет функцию в список обработчиков конкретного события.
• В `create_user` вызываются все обработчики события `user_created`.
• Обратите внимание на важность `signals.get("user_created", [])`, это предотвращает ошибку `KeyError` если нет зарегистрированных обработчиков для данного события.

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

Установка и импорт необходимого пакета

Для работы с фреймворком Flask и декораторами подписок на сигналы используйте пакет Flask-Signal.

Инструкция по установке:

Используйте команду pip install flask-signal в терминале.

Пример импорта:

from flask_signal import Signal

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

Создание базовой структуры приложения Flask

Для начала создайте папку проекта и внутри нее файл app.py.

• Создайте файл app.py с кодом ниже

from flask import Flask
app = Flask(__name__)
@app.route("/")
def index():
return "Главная страница"
if __name__ == "__main__":
app.run(debug=True)

Этот код импортирует Flask, создаёт приложение Flask с именем app и определяет маршрут /. Функция index возвращает строку "Главная страница". Блок if __name__ == "__main__": запускает приложение в отладочном режиме.

• Далее, убедитесь, что у вас установлен Flask: pip install Flask
• Запустите приложение из командной строки в папке проекта: python app.py
• В браузере перейдите по адресу: http://127.0.0.1:5000/

Если всё работает, вы увидите "Главная страница".

• Добавьте в проект папку signals, с файлами, связанными с подписками

Определение сигналов и декораторов подписки

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

def handle_submission(user_id, data)
# Обработка данных
print(f"Данные пользователя {user_id} получены")
return True

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

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

Необходимо использовать корректное имя сигнала.

Подключение обработки событий к сигналам

Для обработки сигнала в Flask-фреймворке, используйте декоратор @app.signal. Внутри функции, помеченной этим декоратором, используйте ключевые слова аргументов для обозначения данных события. Например:

Пример:

from flask import Flask
app = Flask(__name__)
@app.signal('user_registered')
def register_user_handler(user_data):
print(f"Пользователь {user_data['username']} успешно зарегистрирован.")
# Дополнительные действия, например, отправка email
return True  # Возвращаем True, если обработка успешна.

В этом примере, при срабатывании события user_registered, вызывается функция register_user_handler. Параметр user_data содержит информацию о зарегистрированном пользователе. Возвращаемое значение (True в данном случае) указывает успешную обработку события.

Важно! Сигналы и связанные с ними данные передаются явно. Не предполагайте наличия данных, если они не определены. Определите тип данных, передаваемых в сигнале, и получите доступ к нем через соответствующий параметр в вашем обработчике.

Пример с типом данных:

from flask import Flask
import json
app = Flask(__name__)
@app.signal('order_placed', dict)
def order_placed_handler(order_details):
order_details['status'] = 'processed'
print(f"Заказ обработан: {json.dumps(order_details)}")
return True

Здесь в сигнале order_placed передаются данные типа dict. Обработчик получает этот словарь в параметре order_details, изменяет его и логирует результат. Это показывает как работать с конкретным типом данных.

Обработка сигналов с передачей данных

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

Пример:

import flask
from flask_signals import signal
app = flask.Flask(__name__)
@signal('user_created')
def user_created_handler(user_id):
print(f"Пользователь с ID {user_id} создан.")
# Дополнительная логика, связанная с созданием пользователя
return {'success': True}
@app.route('/create_user')
def create_user():
user_id = 123
signal('user_created').send(user_id)
return "Пользователь создан."
if __name__ == '__main__':
app.run(debug=True)

В этом примере декоратор @signal('user_created') получает параметр user_id, который передаётся в обработчик user_created_handler. Обработчик может использовать эту информацию для выполнения нужной задачи. Обратите внимание на возвращаемое значение функции user_created_handler – словарь, в котором хранится информация о результате обработки сигнала. Это позволяет организовать более сложный callback.

Можно передавать объекты класса, словари и любые другие типы данных.

При большом количестве данных для передачи используйте сериализацию JSON для упрощения обработки.

Примеры использования подписки на разные сигналы

Подписка на сигналы в Flask позволяет реагировать на события в приложении. Ниже примеры подписки на разные типы сигналов.

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

from flask import Flask
from flask_signals import signal
app = Flask(__name__)
@signal('order_status_changed')
def order_status_changed_handler(order_id, new_status):
print(f'Статус заказа {order_id} изменен на {new_status}')
#код, который генерирует сигнал
@app.route('/order//status/')
def change_status(order_id, new_status):
order_status_changed.send(app, order_id=order_id, new_status=new_status)
return 'Статус изменен'
if __name__ == '__main__':
app.run(debug=True)

Функция order_status_changed_handler обрабатывает сигнал order_status_changed, получая идентификатор заказа и новый статус. Маршрут проверяет статус и генерирует сигнал.

Пример 2: Подписка на сигнал сброса данных:

@signal('data_reset')
def data_reset_handler():
try:
# Обработка сброса данных
print("Данные сброшены")
except Exception as e:
print(f"Ошибка при сбросе данных: {e}")
# код, который генерирует сигнал reset
@app.route('/reset_data')
def reset_data():
data_reset.send(app)
return  'Данные сброшены'

Обработка ошибки включена для практичности. Сигнал обработан без дополнительных параметров.

Пример 3: Подписка на сигнал при регистрации нового пользователя:

@signal('user_registered')
def user_registered_handler(user_id, username):
print(f'Пользователь {username} (ID: {user_id}) успешно зарегистрирован')
#код, который генерирует сигнал
def create_user(user_id, username):
user_registered.send(app, user_id=user_id, username=username)

Этот пример показывает обработку сигнала с данными о зарегистрированном пользователе (ID, имя пользователя).

Вопрос-ответ: