zzstoatzz.io
prefect server in zig
add SDK-based test for events/out websocket
verifies compatibility with prefect.events.clients.PrefectEventSubscriber
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
+90
test_events_out_sdk.py
+90
test_events_out_sdk.py
···
···
1
+
"""
2
+
test /events/out using the actual Prefect SDK subscriber
3
+
4
+
usage:
5
+
# start the server first, then:
6
+
PREFECT_API_URL=http://localhost:4200/api uv run python test_events_out_sdk.py
7
+
"""
8
+
9
+
import asyncio
10
+
import os
11
+
12
+
os.environ.setdefault("PREFECT_API_URL", "http://localhost:4200/api")
13
+
14
+
from prefect.events import emit_event
15
+
from prefect.events.clients import PrefectEventSubscriber
16
+
from prefect.events.filters import EventFilter
17
+
18
+
19
+
async def test_sdk_subscriber():
20
+
received_events: list = []
21
+
22
+
print("\n=== test_events_out_sdk ===\n")
23
+
print(f"API URL: {os.environ.get('PREFECT_API_URL')}")
24
+
25
+
async def subscriber_task():
26
+
"""subscribe and collect events using Prefect SDK"""
27
+
print("[subscriber] connecting with PrefectEventSubscriber...")
28
+
29
+
# use empty filter to match all events
30
+
filter = EventFilter()
31
+
32
+
async with PrefectEventSubscriber(filter=filter) as subscriber:
33
+
print("[subscriber] connected and subscribed!")
34
+
35
+
try:
36
+
async for event in subscriber:
37
+
received_events.append(event)
38
+
print(f"[subscriber] received: {event.event}")
39
+
40
+
# stop after receiving 3 events
41
+
if len(received_events) >= 3:
42
+
break
43
+
except asyncio.CancelledError:
44
+
print("[subscriber] cancelled")
45
+
46
+
async def publisher_task():
47
+
"""emit test events using Prefect SDK"""
48
+
# wait for subscriber to connect
49
+
await asyncio.sleep(1.0)
50
+
51
+
print("[publisher] emitting events...")
52
+
53
+
for i in range(3):
54
+
emit_event(
55
+
event=f"prefect.test.sdk-event-{i}",
56
+
resource={"prefect.resource.id": f"test-resource-{i}"},
57
+
payload={"index": i, "sdk": True},
58
+
)
59
+
print(f"[publisher] emitted: prefect.test.sdk-event-{i}")
60
+
await asyncio.sleep(0.2)
61
+
62
+
print("[publisher] all events emitted")
63
+
64
+
# run both with timeout
65
+
try:
66
+
await asyncio.wait_for(
67
+
asyncio.gather(subscriber_task(), publisher_task()),
68
+
timeout=10.0
69
+
)
70
+
except asyncio.TimeoutError:
71
+
print("[timeout] stopping...")
72
+
73
+
# results
74
+
print(f"\n=== results ===")
75
+
print(f"received {len(received_events)} events")
76
+
77
+
for e in received_events:
78
+
print(f" - {e.event} (id={e.id})")
79
+
80
+
if len(received_events) >= 3:
81
+
print("\n✓ SDK subscriber test passed!")
82
+
return True
83
+
else:
84
+
print("\n✗ expected at least 3 events")
85
+
return False
86
+
87
+
88
+
if __name__ == "__main__":
89
+
success = asyncio.run(test_sdk_subscriber())
90
+
exit(0 if success else 1)