События в Python: от базовых концепций до продвинутых паттернов

class EventEmitter:
def __init__(self):
self.callbacks = {}

def on(self, event_name, callback):
"""Подписка на событие"""
if event_name not in self.callbacks:
self.callbacks[event_name] = []
self.callbacks[event_name].append(callback)

def emit(self, event_name, *args, **kwargs):
"""Генерация события"""
if event_name in self.callbacks:
for callback in self.callbacks[event_name]:
callback(*args, **kwargs)

# Использование
emitter = EventEmitter()

def message_handler(message):
print(f"Получено сообщение: {message}")

emitter.on('message', message_handler)
emitter.emit('message', "Привет, мир!") # Вывод: Получено сообщение: Привет, мир!

# Модуль мониторинга CPU
def monitor_cpu():
while True:
usage = get_cpu_usage()
if usage > 90:
event_emitter.emit('high_cpu', usage)
time.sleep(5)

# Обработчик события
def handle_high_cpu(usage):
notify_admin(f"Критическая загрузка CPU: {usage}%")
restart_some_services()

event_emitter.on('high_cpu', handle_high_cpu)

import sys
from PyQt5.QtWidgets import QApplication, QPushButton, QVBoxLayout, QWidget

class MyApp(QWidget):
def __init__(self):
super().__init__()
self.initUI()

def initUI(self):
layout = QVBoxLayout()

self.button = QPushButton('Нажми меня', self)
self.button.clicked.connect(self.on_button_click) # Подключаем сигнал к слоту

layout.addWidget(self.button)
self.setLayout(layout)
self.setWindowTitle('PyQt События')
self.setGeometry(300, 300, 300, 200)

def on_button_click(self):
print('Кнопка была нажата!')

if __name__ == '__main__':
app = QApplication(sys.argv)
ex = MyApp()
ex.show()
sys.exit(app.exec_())

from PyQt5.QtCore import pyqtSignal

class Counter(QObject):
# Определяем собственный сигнал
value_changed = pyqtSignal(int)

def __init__(self):
super().__init__()
self._value = 0

def increment(self):
self._value += 1
# Излучаем сигнал с новым значением
self.value_changed.emit(self._value)

import tkinter as tk

def on_button_click():
print("Кнопка была нажата!")

root = tk.Tk()
root.title("Tkinter События")
root.geometry("300x200")

button = tk.Button(root, text="Нажми меня")
button.pack(pady=50)

# Привязываем функцию к событию нажатия левой кнопки мыши
button.bind("<Button-1>", lambda event: on_button_click())

# Альтернативный способ
button.config(command=on_button_click)

root.mainloop()

import asyncio

async def handle_client(reader, writer):
"""Обработчик подключения клиента"""
addr = writer.get_extra_info('peername')
print(f"Подключен клиент с адреса {addr}")

while True:
# Ожидаем события получения данных
data = await reader.read(100)
if not data:
break

message = data.decode()
print(f"Получено сообщение от {addr}: {message}")

# Отправляем ответ
response = f"Эхо: {message}"
writer.write(response.encode())
await writer.drain()

print(f"Клиент {addr} отключился")
writer.close()

async def main():
# Создаём сервер, который будет принимать подключения
server = await asyncio.start_server(
handle_client, '127.0.0.1', 8888)

addr = server.sockets[0].getsockname()
print(f"Сервер запущен на {addr}")

# Сервер работает до вызова server.close()
async with server:
await server.serve_forever()

if __name__ == "__main__":
asyncio.run(main())

import asyncio
import time

async def delayed_message(delay, message):
"""Выводит сообщение после задержки"""
await asyncio.sleep(delay)
print(f"{time.strftime('%H:%M:%S')}: {message}")

async def periodic_task(interval):
"""Выполняет задачу периодически"""
count = 0
while True:
count += 1
print(f"{time.strftime('%H:%M:%S')}: Периодическое событие #{count}")
await asyncio.sleep(interval)

async def wait_for_external_event(future):
"""Ожидает внешнего события (заполнения future)"""
print(f"{time.strftime('%H:%M:%S')}: Ожидание события...")
result = await future
print(f"{time.strftime('%H:%M:%S')}: Событие произошло! Результат: {result}")

async def simulate_external_event(future, delay):
"""Имитирует внешнее событие, устанавливая результат future"""
await asyncio.sleep(delay)
future.set_result("Данные от внешнего источника")

async def main():
# Создаём future для имитации внешнего события
event_future = asyncio.Future()

# Планируем различные задачи
tasks = [
asyncio.create_task(delayed_message(2, "Сообщение с задержкой")),
asyncio.create_task(periodic_task(3)),
asyncio.create_task(wait_for_external_event(event_future)),
asyncio.create_task(simulate_external_event(event_future, 5))
]

# Запускаем все задачи и завершаемся через 10 секунд
await asyncio.sleep(10)

# Отменяем все задачи при выходе
for task in tasks:
task.cancel()

if __name__ == "__main__":
asyncio.run(main())

from abc import ABC, abstractmethod
from typing import List, Any

# Интерфейс Observer
class Observer(ABC):
@abstractmethod
def update(self, subject: 'Subject', *args, **kwargs) -> None:
"""Метод, вызываемый при обновлении субъекта"""
pass

# Класс Subject
class Subject:
def __init__(self):
self._observers: List[Observer] = []

def attach(self, observer: Observer) -> None:
"""Подключает наблюдателя к субъекту"""
if observer not in self._observers:
self._observers.append(observer)

def detach(self, observer: Observer) -> None:
"""Отключает наблюдателя от субъекта"""
try:
self._observers.remove(observer)
except ValueError:
pass

def notify(self, *args, **kwargs) -> None:
"""Уведомляет всех наблюдателей о событии"""
for observer in self._observers:
observer.update(self, *args, **kwargs)

# Конкретный субъект – текстовый редактор
class TextEditor(Subject):
def __init__(self):
super().__init__()
self._text = ""

@property
def text(self) -> str:
return self._text

@text.setter
def text(self, value: str) -> None:
self._text = value
# Уведомляем наблюдателей о событии изменения текста
self.notify(event_type="text_changed", new_text=value)

def save(self, filename: str) -> None:
with open(filename, 'w') as file:
file.write(self._text)
# Уведомляем наблюдателей о событии сохранения
self.notify(event_type="saved", filename=filename)

# Конкретные наблюдатели
class LoggingObserver(Observer):
def update(self, subject: Subject, *args, **kwargs) -> None:
event_type = kwargs.get("event_type", "unknown")
if event_type == "text_changed":
print(f"[LOG] Текст был изменен на: {kwargs.get('new_text')}")
elif event_type == "saved":
print(f"[LOG] Текст сохранен в файл: {kwargs.get('filename')}")

class SpellCheckObserver(Observer):
def update(self, subject: Subject, *args, **kwargs) -> None:
event_type = kwargs.get("event_type", "unknown")
if event_type == "text_changed":
text = kwargs.get('new_text', "")
# Простая проверка орфографии
if "нету" in text:
print("[SPELL] Ошибка: используйте 'нет' вместо 'нету'")

# Использование
editor = TextEditor()
logger = LoggingObserver()
spell_checker = SpellCheckObserver()

editor.attach(logger)
editor.attach(spell_checker)

editor.text = "Привет, мир!" # Вызовет событие text_changed
editor.text = "Тут нету ошибок" # Вызовет событие text_changed и сообщение от проверки орфографии
editor.save("document.txt") # Вызовет событие saved

import time
import threading
from abc import ABC, abstractmethod
from typing import Dict, List

# Базовый класс для событий
class Event:
def __init__(self, event_type: str, data: Dict):
self.timestamp = time.time()
self.type = event_type
self.data = data

# Интерфейс слушателя событий
class EventListener(ABC):
@abstractmethod
def handle_event(self, event: Event) -> None:
pass

# Диспетчер событий
class EventDispatcher:
def __init__(self):
self._listeners: Dict[str, List[EventListener]] = {}

def add_listener(self, event_type: str, listener: EventListener) -> None:
if event_type not in self._listeners:
self._listeners[event_type] = []
self._listeners[event_type].append(listener)

def remove_listener(self, event_type: str, listener: EventListener) -> None:
if event_type in self._listeners and listener in self._listeners[event_type]:
self._listeners[event_type].remove(listener)

def dispatch(self, event: Event) -> None:
if event.type in self._listeners:
for listener in self._listeners[event.type]:
listener.handle_event(event)

# Система мониторинга
class ServerMonitor:
def __init__(self, event_dispatcher: EventDispatcher):
self.dispatcher = event_dispatcher
self.servers = {}
self.running = False
self._monitor_thread = None

def add_server(self, server_id: str, server_info: Dict) -> None:
self.servers[server_id] = server_info

def start_monitoring(self) -> None:
if self.running:
return

self.running = True
self._monitor_thread = threading.Thread(target=self._monitoring_loop)
self._monitor_thread.daemon = True
self._monitor_thread.start()

def stop_monitoring(self) -> None:
self.running = False
if self._monitor_thread:
self._monitor_thread.join(timeout=1.0)

def _monitoring_loop(self) -> None:
while self.running:
for server_id, server_info in self.servers.items():
# Имитация проверки метрик
cpu_usage = self._get_cpu_usage(server_id)
memory_usage = self._get_memory_usage(server_id)
disk_usage = self._get_disk_usage(server_id)

# Проверка критических условий и генерация событий
if cpu_usage > 90:
self.dispatcher.dispatch(Event("high_cpu", {
"server_id": server_id,
"cpu_usage": cpu_usage
}))

if memory_usage > 85:
self.dispatcher.dispatch(Event("high_memory", {
"server_id": server_id,
"memory_usage": memory_usage
}))

if disk_usage > 95:
self.dispatcher.dispatch(Event("high_disk", {
"server_id": server_id,
"disk_usage": disk_usage
}))

time.sleep(30) # Проверка каждые 30 секунд

# Методы имитации получения метрик (в реальной системе здесь были бы API-вызовы)
def _get_cpu_usage(self, server_id: str) -> float:
# Имитация получения данных о CPU
return 95.0 if server_id == "server1" else 50.0

def _get_memory_usage(self, server_id: str) -> float:
# Имитация получения данных о памяти
return 70.0

def _get_disk_usage(self, server_id: str) -> float:
# Имитация получения данных о диске
return 98.0 if server_id == "server2" else 60.0

# Конкретные обработчики событий
class EmailNotifier(EventListener):
def handle_event(self, event: Event) -> None:
server_id = event.data.get("server_id", "unknown")
print(f"[EMAIL] Отправка уведомления о событии {event.type} для сервера {server_id}")
# В реальном коде здесь был бы вызов API для отправки email

class LoggingListener(EventListener):
def handle_event(self, event: Event) -> None:
print(f"[LOG] {time.ctime(event.timestamp)}: {event.type} – {event.data}")

class AutoScaler(EventListener):
def handle_event(self, event: Event) -> None:
if event.type == "high_cpu":
server_id = event.data.get("server_id")
cpu_usage = event.data.get("cpu_usage")
print(f"[AUTOSCALE] Увеличиваем ресурсы для сервера {server_id} (CPU: {cpu_usage}%)")
# В реальном коде здесь был бы вызов API облачного провайдера

# Использование системы
dispatcher = EventDispatcher()
monitor = ServerMonitor(dispatcher)

# Регистрация слушателей
logger = LoggingListener()
emailer = EmailNotifier()
scaler = AutoScaler()

dispatcher.add_listener("high_cpu", logger)
dispatcher.add_listener("high_cpu", emailer)
dispatcher.add_listener("high_cpu", scaler)
dispatcher.add_listener("high_memory", logger)
dispatcher.add_listener("high_memory", emailer)
dispatcher.add_listener("high_disk", logger)
dispatcher.add_listener("high_disk", emailer)

# Добавление серверов для мониторинга
monitor.add_server("server1", {"name": "Web Server", "ip": "192.168.1.10"})
monitor.add_server("server2", {"name": "Database Server", "ip": "192.168.1.11"})

# Запуск мониторинга
monitor.start_monitoring()

# В реальном приложении здесь был бы бесконечный цикл или другой механизм поддержания работы программы
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
monitor.stop_monitoring()
print("Мониторинг остановлен")

import asyncio
import aiohttp
from bs4 import BeautifulSoup
from urllib.parse import urlparse, urljoin

class WebScraper:
def __init__(self, base_url, max_pages=10):
self.base_url = base_url
self.max_pages = max_pages
self.visited_urls = set()
self.queue = asyncio.Queue()
self.event_handlers = {
'page_downloaded': [],
'link_found': [],
'scraping_completed': [],
'error_occurred': []
}

def on(self, event_type, callback):
"""Подписка на событие"""
if event_type in self.event_handlers:
self.event_handlers[event_type].append(callback)
return self

async def emit(self, event_type, data):
"""Генерация события"""
if event_type in self.event_handlers:
for callback in self.event_handlers[event_type]:
if asyncio.iscoroutinefunction(callback):
await callback(data)
else:
callback(data)

async def download_page(self, url):
"""Загрузка страницы по URL"""
if url in self.visited_urls:
return None

self.visited_urls.add(url)

try:
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=10) as response:
if response.status == 200:
html = await response.text()
await self.emit('page_downloaded', {'url': url, 'html': html})
return html
else:
await self.emit('error_occurred', {
'url': url, 
'status': response.status,
'message': f"Failed to download {url}: HTTP {response.status}"
})
return None
except Exception as e:
await self.emit('error_occurred', {
'url': url, 
'error': str(e),
'message': f"Error downloading {url}: {str(e)}"
})
return None

async def extract_links(self, url, html):
"""Извлечение ссылок из HTML"""
if not html:
return

soup = BeautifulSoup(html, 'html.parser')
base_domain = urlparse(self.base_url).netloc

for link in soup.find_all('a', href=True):
href = link['href']
full_url = urljoin(url, href)

# Проверяем, что ссылка ведёт на тот же домен
if urlparse(full_url).netloc == base_domain and full_url not in self.visited_urls:
await self.emit('link_found', {'source_url': url, 'found_url': full_url})
await self.queue.put(full_url)

async def process_page(self, url):
"""Обработка одной страницы"""
html = await self.download_page(url)
if html:
await self.extract_links(url, html)

async def run(self):
"""Запуск скрапера"""
# Добавляем начальный URL в очередь
await self.queue.put(self.base_url)

# Создаём воркеры для обработки страниц
workers = []
for _ in range(min(5, self.max_pages)): # Максимум 5 параллельных запросов
workers.append(asyncio.create_task(self.worker()))

# Ждём завершения всех воркеров
await asyncio.gather(*workers)
await self.emit('scraping_completed', {'visited_urls': list(self.visited_urls)})

async def worker(self):
"""Воркер для обработки страниц из очереди"""
while len(self.visited_urls) < self.max_pages:
try:
url = await asyncio.wait_for(self.queue.get(), timeout=5.0)
await self.process_page(url)
self.queue.task_done()

# Если очередь пуста и все страницы обработаны, выходим
if self.queue.empty() and len(self.visited_urls) > 0:
break

except asyncio.TimeoutError:
# Тайм-аут получения из очереди (очередь может быть пуста)
if self.queue.empty() and len(self.visited_urls) > 0:
break
except Exception as e:
await self.emit('error_occurred', {
'error': str(e),
'message': f"Worker error: {str(e)}"
})

# Пример использования
async def main():
scraper = WebScraper('https://python.org', max_pages=5)

# Регистрируем обработчики событий
scraper.on('page_downloaded', lambda data: print(f"Downloaded: {data['url']} ({len(data['html'])} bytes)"))
scraper.on('link_found', lambda data: print(f"Found link: {data['found_url']} on {data['source_url']}"))
scraper.on('error_occurred', lambda data: print(f"Error: {data['message']}"))

async def on_complete(data):
print(f"Scraping completed. Visited {len(data['visited_urls'])} URLs:")
for url in data['visited_urls']:
print(f" – {url}")

scraper.on('scraping_completed', on_complete)

# Запускаем скрапер
await scraper.run()

# Запуск основной функции
if __name__ == "__main__":
asyncio.run(main())