Ачааллж байна...
Түвшин 3-ын төгсгөлд хүрлээ. Арван нэгэн хичээлийн туршид бид class-уудын гүн рүү орсон. classmethod-оор туслах хаалга нээж, method-ын гурван төрлийг сонгох дүрэм гаргаж, class attribute-ын хуваалцсан өгөгдлийн урхийг таньж, dataclass-ийн frozen, default_factory, __post_init__-ийг эзэмшсэн. Дараа нь class-уудаа Python-ий бэлэн төрлүүд шиг ажилладаг болгосон: operator-ууд, container протокол, __call__, __iter__. Эцэст нь pattern-ийн санаатай танилцаж, Strategy ба Factory-г Python маягаар бүтээсэн. Энэ хичээлд тэр бүх мэдлэг нэг бодит системд нэгдэнэ.
Бид мэдэгдлийн систем бүтээнэ: Observer pattern. Асуудал нь: нэг объектод ямар нэг үйл явдал болоход, жишээлбэл онлайн дэлгүүрт захиалга өгөгдөхөд, олон өөр хэсэг түүнд хариу үйлдэл үзүүлэх ёстой. Хэрэглэгчид имэйл илгээх, агуулахын нөөцийг багасгах, борлуулалтын тайланд бүртгэх. Хамгийн шууд арга нь захиалгын кодонд эдгээр гурван үйлдлийг шууд дуудах. Гэвч тэгвэл захиалгын код имэйл, агуулах, тайлангийн бүх нарийн ширийн зүйлийг мэдэх ёстой болно, шинэ хариу үйлдэл нэмэгдэх бүрт захиалгын кодыг засах хэрэгтэй. Observer энэ хамаарлыг тасалдаг: захиалгын код зөвхөн "захиалга өгөгдлөө" гэж зарлана, хэн сонсож байгааг огт мэдэхгүй.
Энэ бүтээлд өмнөх бүх түвшний хэрэгслүүд байраа олно. Сонсогчид нь Хичээл 8.3-ын callable-ууд: энгийн функц, __call__-тэй объект. Бүртгэл нь Хичээл 9.2-9.3-ын dict, бүртгэлийн decorator-той. Бүртгэлээс хасах нь Бүлэг 4-ийн closure. Үйл явдал нь Хичээл 7.4-ийн frozen dataclass. Захиалгын дугаар Хичээл 2.4-ийн count. Мэдэгдэх явцад жагсаалт өөрчлөгдөх аюулаас Хичээл 8.4-ийн сургамж хамгаална. Ажлыг Git branch дээр хийж, pytest тестүүдтэй commit хийнэ.
Ангийн эцэг эхчүүдийн мессежийн бүлгийг төсөөлөөд үзээрэй. Багш бүлэгт "маргааш хичээл цуцлагдлаа" гэж бичнэ. Тэр эцэг эх бүрийг тусад нь дуудахгүй, хэн бүлэгт байгааг ч сайн мэдэхгүй байж болно. Мессеж бүлгийн бүх гишүүнд хүрнэ. Нэг эцэг эх хүүхдээ гэртээ үлдээх зохицуулалт хийнэ, нөгөө нь ажлаасаа чөлөө авна, гурав дахь нь асран хамгаалагч руу утасдана. Хариу үйлдэл бүр эцэг эх бүрийн өөрийн хэрэг, багш тэдгээрийг мэдэх шаардлагагүй. Шинэ эцэг эх бүлэгт нэгдэхэд багшийн ажил өөрчлөгдөхгүй.
Observer pattern-д багш бол subject буюу "үйл явдлын эх сурвалж", эцэг эхчүүд бол observer буюу "сонсогч", мессежийн бүлэг бол тэднийг холбох хэрэгсэл. Python хөгжүүлэгчид энэ хэрэгслийг ихэвчлэн event bus буюу "үйл явдлын шугам" гэж нэрлэдэг. Бидний систем гурван хэсэгтэй:
event-hub/
events.py Event ба EventBus: ерөнхий мэдэгдлийн хэрэгсэл
shop.py OrderService (subject), сонсогчид, демо
test_events.py pytest тестүүдevents.py нь дэлгүүрийн тухай юу ч мэдэхгүй байна: ямар ч төсөлд дахин хэрэглэгдэх ерөнхий хэрэгсэл, Хичээл 6.4-ийн measurekit шиг. shop.py түүнийг хэрэглэдэг тодорхой програм.
git switch -c level3-event-hubRepository дотор event-hub хавтас үүсгэж, terminal-аар түүн рүү орно. venv идэвхтэй эсэхийг шалгаарай.
events.py файл үүсгэж, эхлээд хамгийн бага ажилладаг хувилбарыг бичье:
python
"""Жижиг үйл явдлын систем: Observer pattern."""
from collections import defaultdict
from dataclasses import dataclass, field
@dataclass(frozen=True)
class Event:
name: str
payload: dict = field(default_factory=dict)
class EventBus:
def __init__(self):
self._handlers = defaultdict(list)
def subscribe(self, event_name, handler):
self._handlers[event_name].append(handler)
def publish(self, event_name, **payload):
event = Event(event_name, payload)
for handler in self._handlers.get(event_name, ()):
handler(event)Event бол Хичээл 7.4-ийн frozen dataclass: үйл явдлын нэр ба түүнтэй холбоотой өгөгдөл. payload-ийн анхдагч утга default_factory-ээр, учир нь dict өөрчлөгдөж болох төрөл. frozen бол үйл явдал болсон баримт, сонсогч түүнийг өөрчлөх ёсгүй. Шударга тэмдэглэл: Хичээл 7.4-ийн "гүехэн frozen"-ийн дагуу payload dict-ийн дотрыг сонсогч өөрчилж чадна. Бид үүнийг дүрмээр хориглоно: сонсогчид payload-ийг зөвхөн уншина.
EventBus дотор _handlers нь collections модулийн defaultdict: түлхүүр байхгүй үед автоматаар анхдагч утга үүсгэдэг dict. defaultdict(list) нь шинэ түлхүүр анх хандагдах үед list() дуудаж хоосон жагсаалт үүсгэнэ. Энэ нь Хичээл 7.4-ийн default_factory-тэй ижил санаа: factory функцийг дамжуулсан, хаалтгүй. Тэгэхээр subscribe-д "энэ үйл явдлын жагсаалт байгаа эсэх"-ийг шалгах шаардлагагүй. Мөн _handlers __init__ дотор үүссэн, class attribute биш, Хичээл 7.3-ын урхинаас ангид: event bus бүр өөрийн бүртгэлтэй.
subscribe сонсогчийг тухайн үйл явдлын нэрийн жагсаалтад нэмнэ. publish Event объект үүсгэж, тэр нэрийн бүх сонсогчийг дарааллаар нь дуудна. **payload нь Хичээл 5.1-ийн **kwargs: дурын түлхүүр үгийн argument-уудыг dict болгон цуглуулна. publish-д .get(event_name, ()) ашигласныг анзаараарай: defaultdict-д шууд [event_name] гэж хандвал сонсогчгүй үйл явдал бүрт хоосон жагсаалт үүсэх байсан. get() шинэ түлхүүр үүсгэдэггүй.
Шалгах цэг 1. event-hub хавтаснаас Python-ий interactive горимыг нээнэ:
python
>>> from events import EventBus
>>> bus = EventBus()
>>> bus.subscribe("greet", lambda e: print("А сонслоо:", e.payload))
>>> bus.subscribe("greet", lambda e: print("Б сонслоо:", e.name))
>>> bus.publish("greet", name="Бат")
А сонслоо: {'name': 'Бат'}
Б сонслоо: greet
>>> bus.publish("nobody_listens", x=1)
>>>Хоёр сонсогч дарааллаар нь дуудагдсан. Сонсогчгүй үйл явдал юу ч хийгээгүй, алдаа ч өгөөгүй. Observer-ийн гол санаа аль хэдийн ажиллаж байна: publish-ийг дуудсан код хэн сонсож байгааг мэдээгүй.
Хамгийн энгийн хувилбарт бодит хэрэглээнд хэрэгтэй дөрвөн шинж чанар дутуу. events.py-г дараах бүрэн хувилбараар солино:
python
"""Жижиг үйл явдлын систем: Observer pattern."""
from collections import defaultdict
from dataclasses import dataclass, field
@dataclass(frozen=True)
class Event:
name: str
payload: dict = field(default_factory=dict)
def print_error(handler, event, error):
name = getattr(handler, "__name__", type(handler).__name__)
print(f" [алдаа] {name} '{event.name}'-г боловсруулж чадсангүй: {error}")
class EventBus:
def __init__(self, on_error=print_error):
self._handlers = defaultdict(list)
self.on_error = on_error
def subscribe(self, event_name, handler):
if not callable(handler):
raise TypeError(
f"Сонсогч дуудагдах боломжтой байх ёстой, "
f"{type(handler).__name__} ирлээ"
)
self._handlers[event_name].append(handler)
def unsubscribe():
handlers = self._handlers[event_name]
if handler in handlers:
handlers.remove(handler)
return unsubscribe
def on(self, event_name):
def register(handler):
self.subscribe(event_name, handler)
return handler
return register
def handlers_for(self, event_name):
return tuple(self._handlers.get(event_name, ()))
def publish(self, event_name, **payload):
event = Event(event_name, payload)
failures = []
for handler in self.handlers_for(event_name):
try:
handler(event)
except Exception as error:
failures.append((handler, error))
self.on_error(handler, event, error)
return failuresДөрвөн шинэ шинж чанарыг нэг нэгээр нь тайлбарлая.
Нэгдүгээрт: сонсогчийн шалгалт. subscribe эхлээд Хичээл 8.3-ын callable()-ээр сонсогчийг шалгана. Хэрэв хэн нэгэн функцийн оронд түүний нэрийг string болгон, эсвэл функцийн үр дүнг өгвөл алдаа бүртгэх үед гарна, үйл явдал болох хүртэл хүлээхгүй. Хичээл 6.4-ийн "буруу тохиргоог эрт илрүүлэх" зарчим.
Хоёрдугаарт: бүртгэлээс хасах closure. subscribe одоо unsubscribe гэсэн функц буцаадаг. Энэ бол Бүлэг 4-ийн closure: unsubscribe-ийн үүргэвчинд self, event_name, handler гурав хадгалагдсан. Сонсогч бүртгэлээс гарах шаардлагатай бол тэр функцийг дуудахад хангалттай, ямар үйл явдалд, ямар сонсогчийг хасахыг санах шаардлагагүй. Энэ загварыг олон орчин үеийн сангууд хэрэглэдэг.
Гуравдугаарт: decorator хэлбэрийн бүртгэл. on method Хичээл 9.2-9.3-ын бүртгэлийн decorator-ыг event bus-ийн method болгосон. Хичээл 5.4-т бид @width.setter гэж объектын method-ыг decorator болгон хэрэглэсэн. Энд ижил хэлбэр: @bus.on("order_placed").
Дөрөвдүгээрт: сонсогчдыг бие биенээс тусгаарлах. Энэ бол хамгийн чухал өөрчлөлт бөгөөд publish-ийн бүтцийг тодорхойлдог. Trace-ээр тайлбарлая.
Trace: publish-ийн хамгаалалтууд
Гурван сонсогчтой “order_placed” үйл явдлыг төсөөлье: email, inventory, log_sale. inventory алдаа шиддэг гэж үзье.
Алхам 1 (Event үүсэх). Event("order_placed", {"order": ...}). publish-ийн frame-д: event, failures = [].
Алхам 2 (жагсаалтын хуулбар). self.handlers_for(event_name) нь сонсогчдын жагсаалтын tuple хуулбарыг буцаана. for loop энэ хуулбар дээр давтана, жинхэнэ жагсаалт дээр биш. Энэ нь Хичээл 8.4-ийн Түгээмэл алдааны шийдэл: хэрэв сонсогч мэдэгдэл хүлээн авах явцдаа өөрийгөө бүртгэлээс хасвал жинхэнэ жагсаалт богиносч, давталт дараагийн сонсогчийг алгасах байсан. Хуулбар өөрчлөгдөхгүй тул бүх сонсогч мэдэгдлээ авна. Бид энэ тохиолдлыг Алхам 4-т тестээр батална.
Алхам 3 (email). try дотор email(event) дуудагдана, амжилттай.
Алхам 4 (inventory). inventory(event) ValueError шиднэ. except түүнийг барьж, (inventory, алдаа) хосыг failures-д нэмж, on_error-ийг дуудна. Давталт үргэлжилнэ.
Алхам 5 (log_sale). log_sale(event) дуудагдана, амжилттай.
Алхам 6 (буцах). publish failures жагсаалтыг буцаана: нэг элементтэй.
Хэрэв try/except байгаагүй бол Алхам 4-ийн алдаа publish-ээс гадагш гарч, log_sale хэзээ ч дуудагдахгүй байх байсан. Тэр ч бүү хэл захиалга өгсөн код агуулахын алдаагаар унах байсан, хэдийгээр захиалгын код агуулахын тухай юу ч мэдэхгүй байсан ч. Observer-ийн гол амлалт, subject сонсогчдоо мэдэхгүй, эвдэрнэ: нэг сонсогчийн асуудал бүхэл системийг зогсооно. Тусгаарлалт тэр амлалтыг хамгаална.
on_error бол Хичээл 6.4-ийн report-той ижил загвар: алдааг хаашаа мэдээлэхийг гаднаас сольж болдог, анхдагчаар хэвлэнэ, тестэд чимээгүй функц өгнө. print_error сонсогчийн нэрийг getattr-ээр авдаг: энгийн функцэд __name__ бий, харин __call__-тэй объектод байхгүй, тэр үед class-ийн нэрийг хэрэглэнэ. failures-ийг буцааж байгаа нь дуудагч тал хүсвэл алдаануудыг шалгах боломж олгодог: тестүүд үүнийг хэрэглэнэ.
except Exception бүх энгийн алдааг барина. Ctrl+C-ээр үүсэх KeyboardInterrupt зэрэг "системийн" exception-ууд Exception-ээс удамшдаггүй тул барьдаггүй: програмыг зогсоох хүсэлт сонсогчдын тусгаарлалтад залгигдахгүй. Энэ зөрүүг Бүлэг 11-д exception-уудын шатлалыг судлахдаа дэлгэрэнгүй харна.
Шалгах цэг 2.
python
>>> from events import EventBus
>>> bus = EventBus()
>>> def broken(event):
... raise RuntimeError("эвдэрсэн")
...
>>> bus.subscribe("ping", broken)
<function EventBus.subscribe.<locals>.unsubscribe at 0x7f...>
>>> stop = bus.subscribe("ping", lambda e: print("хоёр дахь сонсогч ажиллалаа"))
>>> failures = bus.publish("ping")
[алдаа] broken 'ping'-г боловсруулж чадсангүй: эвдэрсэн
хоёр дахь сонсогч ажиллалаа
>>> stop()
>>> len(bus.handlers_for("ping"))
1
>>> bus.subscribe("ping", "broken")
Traceback (most recent call last):
...
TypeError: Сонсогч дуудагдах боломжтой байх ёстой, str ирлээЭвдэрсэн сонсогч хоёр дахийг зогсоосонгүй. stop() дуудагдсаны дараа нэг сонсогч үлдсэн. String-ийг сонсогч болгох гэсэн оролдлого шууд татгалзагдсан. Эхний subscribe-ийн буцаасан unsubscribe функцийг interactive горим хэвлэсэн: түүний бүтэн нэр EventBus.subscribe.<locals>.unsubscribe нь Хичээл 4.1-ийн closure-ийн гарал үүслийг харуулж байна.
shop.py файл үүсгэнэ:
python
from dataclasses import dataclass
from itertools import count
from events import EventBus
ORDER_PLACED = "order_placed"
ORDER_CANCELLED = "order_cancelled"
@dataclass(frozen=True)
class Order:
order_id: int
customer: str
items: tuple
total: int
class OrderService:
def __init__(self, bus, prices):
self.bus = bus
self.prices = dict(prices)
self._ids = count(1001)
self.orders = {}
def place_order(self, customer, items):
total = sum(self.prices[product] * qty for product, qty in items)
order = Order(next(self._ids), customer, tuple(items), total)
self.orders[order.order_id] = order
self.bus.publish(ORDER_PLACED, order=order)
return order
def cancel_order(self, order_id):
order = self.orders.pop(order_id)
self.bus.publish(ORDER_CANCELLED, order=order)
return order
class EmailNotifier:
def __init__(self):
self.sent = []
def __call__(self, event):
order = event.payload["order"]
if event.name == ORDER_PLACED:
subject = f"Захиалга #{order.order_id} хүлээн авлаа"
else:
subject = f"Захиалга #{order.order_id} цуцлагдлаа"
self.sent.append((order.customer, subject))
print(f" [имэйл] {order.customer}: {subject}")
class InventoryUpdater:
def __init__(self, stock):
self.stock = dict(stock)
def __call__(self, event):
order = event.payload["order"]
if event.name == ORDER_PLACED:
for product, qty in order.items:
available = self.stock.get(product, 0)
if available < qty:
raise ValueError(f"'{product}' нөөц хүрэлцэхгүй: {available} үлдсэн")
sign = -1
else:
sign = 1
for product, qty in order.items:
self.stock[product] = self.stock.get(product, 0) + sign * qty
print(f" [агуулах] {self.stock}")
def log_sale(event):
order = event.payload["order"]
print(f" [тайлан] #{order.order_id}: {order.total:,}₮")
def main():
bus = EventBus()
shop = OrderService(bus, {"Ном": 25000, "Дэвтэр": 1500, "Үзэг": 800})
email = EmailNotifier()
inventory = InventoryUpdater({"Ном": 3, "Дэвтэр": 10, "Үзэг": 5})
for event_name in (ORDER_PLACED, ORDER_CANCELLED):
bus.subscribe(event_name, email)
bus.subscribe(event_name, inventory)
bus.subscribe(ORDER_PLACED, log_sale)
print("1) Бат захиалга өгнө:")
first = shop.place_order("Бат", [("Ном", 2), ("Үзэг", 1)])
print("2) Сараа таван ном захиална:")
shop.place_order("Сараа", [("Ном", 5)])
print("3) Батын захиалга цуцлагдана:")
shop.cancel_order(first.order_id)
print("Илгээсэн имэйл:", len(email.sent))
if __name__ == "__main__":
main()Хэсэг тус бүрийн үүргийг тодорхойлъё.
Үйл явдлын нэрс модулийн түвшний тогтмол болсон: ORDER_PLACED, ORDER_CANCELLED. “order_placed” гэж string-ийг газар бүрт бичихийн оронд нэг газарт тодорхойлж, нэрээр нь хэрэглэнэ. Үсгийн алдаа гарвал NameError шууд гарна, string-ийн үсгийн алдаа чимээгүй өнгөрөх байсан. Түгээмэл алдаанууд хэсэгт үүнийг дахин харна.
OrderService бол subject. Түүнд event bus өгөгддөг, Хичээл 9.2-ын стратеги шиг гаднаас. place_order нийт дүнг тооцоолж, Order объект үүсгэж, хадгалаад, үйл явдал зарлана. cancel_order захиалгыг хасаж, үйл явдал зарлана. OrderService имэйл, агуулах, тайлангийн тухай нэг ч мөр агуулаагүйг анзаараарай. Захиалгын дугаар Хичээл 2.4-ийн count(1001)-ээс, Хичээл 2.4-ийн Жишээ 1-ийн "хуваалцсан тоолуур" загвараар. Order frozen dataclass, items нь tuple: Хичээл 7.4-ийн "гүехэн frozen" сургамжийн дагуу list биш tuple.
EmailNotifier ба InventoryUpdater бол __call__-тэй сонсогчид. Хичээл 8.3-ын сонголтын дүрэм энд шийдвэр гаргасан: хоёулаа дуудлагын хооронд төлөв хадгалах ёстой. Илгээсэн имэйлүүдийн жагсаалт, агуулахын нөөц. Тиймээс closure биш, callable объект. Хоёулаа хоёр үйл явдлыг сонсож, event.name-ээр ялгана. InventoryUpdater захиалгад нөөцийг шалгаж, хүрэлцэхгүй бол ValueError шиднэ: Алхам 2-ын тусгаарлалтыг шалгах бодит тохиолдол.
log_sale бол энгийн функц. Төлөвгүй, нэг үйлдэлтэй: Хичээл 8.3-ын дүрмээр функц хангалттай. EventBus ялгааг мэдэхгүй: хоёулаа callable.
Шалгах цэг 3.
python shop.py1) Бат захиалга өгнө:
[имэйл] Бат: Захиалга #1001 хүлээн авлаа
[агуулах] {'Ном': 1, 'Дэвтэр': 10, 'Үзэг': 4}
[тайлан] #1001: 50,800₮
2) Сараа таван ном захиална:
[имэйл] Сараа: Захиалга #1002 хүлээн авлаа
[алдаа] InventoryUpdater 'order_placed'-г боловсруулж чадсангүй: 'Ном' нөөц хүрэлцэхгүй: 1 үлдсэн
[тайлан] #1002: 125,000₮
3) Батын захиалга цуцлагдана:
[имэйл] Бат: Захиалга #1001 цуцлагдлаа
[агуулах] {'Ном': 3, 'Дэвтэр': 10, 'Үзэг': 5}
Илгээсэн имэйл: 3Эхний захиалга бүх сонсогчоор амжилттай дамжсан: 2 ном 50,000, нэг үзэг 800. Хоёр дахь захиалгад агуулах алдаа гаргасан ч тусгаарлалтын ачаар тайлан бүртгэгдсэн. __call__-тэй объектын нэр нь print_error-ийн getattr fallback-ээр class-ийн нэр болсон. Цуцлалт нөөцийг анхны байдалд буцаасан.
Шударга ажиглалт: Observer юунд зориулагдаагүй вэ
Хоёр дахь захиалгын гаралтыг дахин хараарай. Агуулахад хангалттай ном байхгүй байсан. Гэвч захиалга өгөгдсөн: Сараа "хүлээн авлаа" гэсэн имэйл авсан, тайланд 125,000 төгрөг бүртгэгдсэн. Энэ бол алдаа, гэхдээ event bus-ийн алдаа биш, бидний дизайны алдаа.
Observer бол хариу үйлдэлд зориулсан pattern: "захиалга өгөгдсөн, үүнийг мэдэж ав". Тэр шийдвэрт зориулагдаагүй: "захиалга өгөгдөж болох уу?" Сонсогчид бие биенээсээ тусгаарлагдсан, subject тэдний хариуг хүлээдэггүй, энэ нь Observer-ийн хүч бөгөөд яг энэ шалтгаанаар сонсогч үйл явдлыг "буцааж" чадахгүй. Нөөцийн шалгалт захиалга өгөхөөс өмнө, OrderService-ийн шийдвэрийн нэг хэсэг байх ёстой: агуулахаас шууд асуух, үйл явдлаар биш. Энэ нь pattern-ийн тухай Хичээл 9.1-ийн "үр дагавар" хэсгийн бодит жишээ: pattern бүр зарим зүйлд тохирдог, бусдад тохирдоггүй. Энэ засварыг сонголтот өргөтгөлд санал болгоно.
Decorator хэлбэрийн бүртгэл
Event bus-ийн on method-ын хэрэглээг харуулахын тулд main-д нэг сонсогч нэмж үзье. bus үүссэний дараа, ямар нэг захиалгаас өмнө:
python
@bus.on(ORDER_PLACED)
def alert_large_order(event):
order = event.payload["order"]
if order.total >= 100_000:
print(f" [менежер] Том захиалга #{order.order_id}: {order.total:,}₮")Одоо python shop.py ажиллуулахад хоёр дахь захиалгын үед “[менежер] Том захиалга #1002: 125,000₮” мөр нэмэгдэнэ. Сонсогч тодорхойлогдох мөчид бүртгэгдсэн, Хичээл 5.1-ийн тодорхойлох цаг. OrderService-ийн кодыг огт хөндөөгүй: Хичээл 9.2-ын "өргөтгөлд нээлттэй, өөрчлөлтөд хаалттай" зарчим бодит програмд.
test_events.py файл үүсгэнэ:
python
from events import EventBus
from shop import ORDER_PLACED, OrderService
def silent(handler, event, error):
pass
def test_publish_calls_subscribers_in_order():
bus = EventBus(on_error=silent)
calls = []
bus.subscribe("ping", lambda e: calls.append("А"))
bus.subscribe("ping", lambda e: calls.append("Б"))
bus.publish("ping")
assert calls == ["А", "Б"]
def test_payload_reaches_handler():
bus = EventBus(on_error=silent)
received = []
bus.subscribe("ping", received.append)
bus.publish("ping", value=42)
assert received[0].name == "ping"
assert received[0].payload == {"value": 42}
def test_unsubscribe_stops_delivery():
bus = EventBus(on_error=silent)
received = []
stop = bus.subscribe("ping", received.append)
stop()
bus.publish("ping")
assert received == []
def test_failing_handler_does_not_block_others():
bus = EventBus(on_error=silent)
received = []
def broken(event):
raise RuntimeError("эвдэрсэн")
bus.subscribe("ping", broken)
bus.subscribe("ping", received.append)
failures = bus.publish("ping")
assert len(received) == 1
assert len(failures) == 1
assert isinstance(failures[0][1], RuntimeError)
def test_unsubscribe_during_publish_does_not_skip_next():
bus = EventBus(on_error=silent)
received = []
def once(event):
stop_once()
stop_once = bus.subscribe("ping", once)
bus.subscribe("ping", received.append)
bus.publish("ping")
assert len(received) == 1
assert bus.handlers_for("ping") == (received.append,)
def test_decorator_registration_keeps_function():
bus = EventBus(on_error=silent)
@bus.on("ping")
def handler(event):
pass
assert bus.handlers_for("ping") == (handler,)
assert handler.__name__ == "handler"
def test_order_service_publishes_order():
bus = EventBus(on_error=silent)
events = []
bus.subscribe(ORDER_PLACED, events.append)
shop = OrderService(bus, {"Ном": 25000})
order = shop.place_order("Бат", [("Ном", 2)])
assert order.total == 50000
assert events[0].payload["order"] == orderТестүүдийн гол санаануудыг товч тэмдэглэе. Олон тест сонсогч болгон received.append-ийг өгсөн: list-ийн append method бол callable, Хичээл 7.2-ын bound method, түүнд ирсэн үйл явдлыг list-д хадгална. Тусгай хуурамч class бичих шаардлагагүй. silent нь Хичээл 6.4-ийн адил: тестийн гаралтыг цэвэр байлгана.
Тав дахь тест Алхам 2-ын жагсаалтын хуулбарыг шалгадаг бөгөөд энэ бүтээлийн хамгийн нарийн тест. once сонсогч дуудагдах үедээ өөрийгөө бүртгэлээс хасдаг. Хэрэв publish жинхэнэ жагсаалт дээр давтдаг байсан бол trace: давталт 0-р байрлалд once-ийг дуудна, once жагсаалтаас хасагдаж, received.append 0-р байрлалд шилжинэ, давталт 1-р байрлал руу шилжиж, юу ч олохгүй. received.append хэзээ ч дуудагдахгүй, тест унана. Хуулбартай хувилбарт хоёр сонсогч хоёулаа дуудагдана. Хэрэв сонирхвол handlers_for дотор tuple-ийг түр хасаж, энэ тест унахыг өөрийн нүдээр харж болно. stop_once нэрийг once дотор хэрэглэхэд тэр хараахан тодорхойлогдоогүй мэт харагдана, гэвч Хичээл 2.1-ийн дүрмээр функцийн бие зөвхөн дуудагдах үед ажилладаг, тэр үед stop_once аль хэдийн оноогдсон.
Сүүлийн тест OrderService-ийг жинхэнэ сонсогчгүйгээр шалгадаг: имэйл илгээгдэхгүй, агуулах өөрчлөгдөхгүй, зөвхөн "үйл явдал зөв өгөгдөлтэй зарлагдсан уу" гэдгийг. Энэ бол Observer-ийн тестлэгдэх байдлын давуу тал: subject ба observer-уудыг тусад нь шалгаж болно.
pytest -q....... [100%]
7 passed in 0.03sgit status
git add events.py shop.py test_events.py
git commit -m "Add event bus with order service observers and tests"Branch-ийг main руу merge хийх эсэх нь таны сонголт.
events.py-ийн эцсийн хувилбар Алхам 2-т, shop.py Алхам 3-т, test_events.py Алхам 4-т бүрэн хэлбэрээрээ байгаа. Алхам 3-ын “Decorator хэлбэрийн бүртгэл” хэсгийн alert_large_order сонсогчийг main-д нэмэх эсэх нь сонголттой.
Хүсвэл дараах өргөтгөлүүдээс сонгож туршиж болно. Тэдгээр нь заавал биш бөгөөд дараагийн хичээлүүд тэдгээрийг хийсэн гэж огт үзэхгүй.
Нөөцийн шалгалтыг зөв газарт. InventoryUpdater-т can_fulfil(items) method нэмж, OrderService-д inventory-г шууд өгч, place_order дотор захиалга үүсгэхээс өмнө шалгах. Нөөц хүрэлцэхгүй бол захиалга үүсэхгүй, үйл явдал зарлагдахгүй, ValueError шууд дуудагч руу очно. Агуулахын нөөцийг багасгах хариу үйлдэл сонсогч хэвээр үлдэнэ. Энэ нь "шийдвэр шууд, хариу үйлдэл үйл явдлаар" гэсэн хуваалтыг бодитоор хэрэгжүүлэх.
Бүх үйл явдлыг сонсох. Тусгай “” нэр нэмж, түүнд бүртгэгдсэн сонсогчид бүх үйл явдлыг хүлээн авдаг болгох. Бүх үйл явдлыг log файлд бичих, debug хийхэд маш хэрэгтэй. publish дотор тухайн нэрийн сонсогчид ба “” сонсогчдыг нийлүүлэх: Хичээл 2.4-ийн chain хэрэг болж магадгүй.
Нэг удаагийн сонсогч. subscribe-д once=True тохиргоо нэмж, сонсогч эхний мэдэгдлийн дараа автоматаар бүртгэлээс гардаг болгох. Тав дахь тестийн once функц энэ санааг гараар хийсэн. Бүлэг 4-ийн closure-оор сонсогчийг ороож, дуудагдсаны дараа unsubscribe-ийг дуудах wrapper бүтээж болно. Хичээл 5.2-ын wraps-ийг мартахгүй.
Сонсогч хэзээ ч дуудагдахгүй байх. Мессежгүй, Observer-ийн хамгийн түгээмэл алдаа. Үйл явдлын нэрийг string-ээр бичихэд үсгийн алдаа гарвал: bus.subscribe("order_place", email). Event bus-ийн хувьд “order_place” нь хүчинтэй нэр, ердөө хэн ч түүнийг зарладаггүй. Ямар ч алдаа гарахгүй, имэйл ердөө илгээгдэхгүй. Энэ бол бид тогтмол ORDER_PLACED-ийг тодорхойлсон шалтгаан: bus.subscribe(ORDER_PLAC, email) гэвэл NameError шууд гарна. Дараагийн түвшинд бид type hint-ийн Literal-оор үйл явдлын нэрсийг хязгаарлаж, mypy-ээр ийм алдааг ажиллуулахаас өмнө илрүүлэх аргыг сурна.
[алдаа] log_sale ‘...’-г боловсруулж чадсангүй: log_sale() takes 0 positional arguments but 1 was given. Сонсогч гэрээг зөрчсөн үед гарна: def log_sale(): гэж event argument-гүйгээр бичих. Event bus бүх сонсогчийг ижил хэлбэрээр, handler(event) гэж дууддаг, Хичээл 9.2-ын стратегийн гэрээтэй ижил. Энэ алдааны онцлог нь тусгаарлалтын ачаар програм унахгүй, зөвхөн хэвлэгдэнэ. Энэ бол тусгаарлалтын сөрөг тал: кодын алдаа ажиллах явцын асуудал мэт харагдаж, анзаарагдахгүй өнгөрч болно. Тиймээс сонсогч бүрийг тестээр шалгах нь чухал.
TypeError: Сонсогч дуудагдах боломжтой байх ёстой, str ирлээ. Энэ бол бидний мессеж. Сонсогчийн оронд түүний нэрийг string болгон, эсвэл bus.subscribe(ORDER_PLACED, log_sale(event)) гэж функцийн дуудлагын үр дүнг өгсөн үед бүртгэх мөчид гарна. Хичээл 9.2-ын "картыг өлгө, тооцоолсон үнийг биш" дүрэм дахин.
KeyError: 'Цүнх'. “'Цүнх' түлхүүр олдсонгүй” гэсэн утгатай. Үнийн жагсаалтад байхгүй бараагаар захиалга өгөхөд place_order-ийн sum мөрөнд гарна. Анзаараарай: энэ алдаа OrderService-ийн дотор, үйл явдал зарлагдахаас өмнө гарна, тусгаарлалт түүнд хамаарахгүй. Энэ нь зөв: захиалгын шийдвэрийн алдаа дуудагч руу очих ёстой. Илүү тодорхой мессежийг Хичээл 7.1-ийн загвараар нэмж болно.
Сонсогчийн объект устахгүй байх. Мессежгүй, урт хугацааны програмд гардаг нарийн асуудал. Event bus сонсогч бүрийг жагсаалтдаа хадгалдаг тул тэр объект програмын бусад хэсэгт хэрэггүй болсон ч устахгүй: event bus түүнийг амьд байлгаж, үйл явдал бүрт дуудсаар байна. Шийдэл нь хэрэггүй болсон сонсогчийг unsubscribe-ээр заавал хасах. Стандарт сангийн weakref модуль энэ асуудлыг автоматаар шийдэх арга санал болгодог бөгөөд энэ нь энэ сургалтын хүрээнээс гадуур.
Бид Observer pattern-ыг бодит мэдэгдлийн систем болгон бүтээлээ. events.py-ийн EventBus ерөнхий хэрэгсэл: subscribe сонсогчийг callable эсэхийг шалгаж бүртгэдэг бөгөөд бүртгэлээс хасах closure буцаадаг. on method бүртгэлийг decorator хэлбэрээр хийдэг. publish frozen Event объект үүсгэж, сонсогчдын жагсаалтын хуулбар дээр давтаж, сонсогч бүрийг try/except-ээр тусгаарлаж, алдааг on_error-оор мэдээлж, бүтэлгүйтлүүдийн жагсаалтыг буцаадаг. Хуулбар нь сонсогч мэдэгдлийн явцад бүртгэлээс гарахад дараагийнх алгасагдахаас хамгаалдаг. shop.py-ийн OrderService subject: үйл явдлыг зарладаг ч сонсогчдыг мэддэггүй. Төлөвтэй сонсогчид __call__-тэй объект, төлөвгүй нь энгийн функц, event bus ялгааг мэддэггүй. Үйл явдлын нэрсийг тогтмол болгосноор үсгийн алдаа чимээгүй өнгөрөхөөс хамгаалсан. Observer хариу үйлдэлд зориулагдсан, шийдвэрт биш: захиалга өгч болох эсэхийг subject өөрөө шийдэх ёстой. Долоон тест бүх механизмыг, тэр дундаа мэдэгдлийн явц дахь бүртгэлээс гарах нарийн тохиолдлыг шалгасан. Ажил branch дээр commit хийгдсэн.
Энэ хичээлээр Түвшин 3 дуусч байна. Class-ууд таны хувьд нууцгүй боллоо: тэдгээрийг бэлэн төрлүүд шиг ажилладаг болгож, хоорондоо шалгарсан загвараар холбож чадна. Дараагийн түвшин Төрөл, алдаа, хурд. Эхний бүлэг type hint-ийг гүнзгийрүүлнэ. Энэ бүтээлийн хоёр сул талыг санаарай: үйл явдлын нэрсийн үсгийн алдаа чимээгүй өнгөрдөг, payload dict-ийн бүтцийг юу ч баталгаажуулдаггүй. Хичээл 10.1-д бид Union, Literal, TypedDict-тэй танилцаж, "энэ параметр зөвхөн эдгээр гурван string-ийн аль нэг байна", "энэ dict яг ийм түлхүүрүүдтэй" гэх мэт нарийн хэлбэрүүдийг кодонд бичиж сурна. Бүлгийн төгсгөлд mypy тэдгээрийг шалгаж, бидний алдааг програм ажиллахаас өмнө олох болно.
Бүртгэлтэй болсноор энэ сургалтын бүх хичээлд хандах эрх авна.