search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
add exercise-api script for generating test traffic
+117
+117
scripts/exercise-api
+117
scripts/exercise-api
···
1
+
#!/usr/bin/env -S uv run --script --quiet
2
+
# /// script
3
+
# requires-python = ">=3.12"
4
+
# dependencies = ["httpx"]
5
+
# ///
6
+
"""
7
+
Exercise all leaflet-search API endpoints to generate traffic for stats.
8
+
9
+
Usage:
10
+
./scripts/exercise-api # run with defaults
11
+
./scripts/exercise-api --count 20 # more requests per endpoint
12
+
"""
13
+
14
+
import asyncio
15
+
import random
16
+
import sys
17
+
18
+
import httpx
19
+
20
+
BASE_URL = "https://leaflet-search-backend.fly.dev"
21
+
22
+
SEARCH_QUERIES = [
23
+
"python", "rust", "zig", "javascript", "typescript",
24
+
"prefect", "workflow", "automation", "data", "api",
25
+
"database", "sqlite", "machine learning", "llm", "claude",
26
+
"bluesky", "atproto", "leaflet", "publishing", "markdown",
27
+
"web", "server", "deploy", "docker", "async",
28
+
]
29
+
30
+
31
+
async def exercise_search(client: httpx.AsyncClient, count: int):
32
+
"""Hit search endpoint with various queries."""
33
+
print(f"search: {count} requests...")
34
+
for i in range(count):
35
+
q = random.choice(SEARCH_QUERIES)
36
+
resp = await client.get(f"/search", params={"q": q})
37
+
if resp.status_code != 200:
38
+
print(f" search failed: {resp.status_code}")
39
+
print(f" done")
40
+
41
+
42
+
async def exercise_similar(client: httpx.AsyncClient, count: int):
43
+
"""Hit similar endpoint with document URIs."""
44
+
print(f"similar: {count} requests...")
45
+
# first get some URIs from search
46
+
resp = await client.get("/search", params={"q": "python"})
47
+
if resp.status_code != 200:
48
+
print(" failed to get URIs")
49
+
return
50
+
docs = resp.json()
51
+
if not docs:
52
+
print(" no docs found")
53
+
return
54
+
55
+
uris = [d["uri"] for d in docs[:5]]
56
+
for i in range(count):
57
+
uri = random.choice(uris)
58
+
resp = await client.get("/similar", params={"uri": uri})
59
+
if resp.status_code != 200:
60
+
print(f" similar failed: {resp.status_code}")
61
+
print(f" done")
62
+
63
+
64
+
async def exercise_tags(client: httpx.AsyncClient, count: int):
65
+
"""Hit tags endpoint."""
66
+
print(f"tags: {count} requests...")
67
+
for i in range(count):
68
+
resp = await client.get("/tags")
69
+
if resp.status_code != 200:
70
+
print(f" tags failed: {resp.status_code}")
71
+
print(f" done")
72
+
73
+
74
+
async def exercise_popular(client: httpx.AsyncClient, count: int):
75
+
"""Hit popular endpoint."""
76
+
print(f"popular: {count} requests...")
77
+
for i in range(count):
78
+
resp = await client.get("/popular")
79
+
if resp.status_code != 200:
80
+
print(f" popular failed: {resp.status_code}")
81
+
print(f" done")
82
+
83
+
84
+
async def main():
85
+
count = 12
86
+
if "--count" in sys.argv:
87
+
idx = sys.argv.index("--count")
88
+
if idx + 1 < len(sys.argv):
89
+
count = int(sys.argv[idx + 1])
90
+
91
+
print(f"exercising {BASE_URL} ({count} requests per endpoint)\n")
92
+
93
+
async with httpx.AsyncClient(base_url=BASE_URL, timeout=30) as client:
94
+
await asyncio.gather(
95
+
exercise_search(client, count),
96
+
exercise_similar(client, count),
97
+
exercise_tags(client, count),
98
+
exercise_popular(client, count),
99
+
)
100
+
101
+
print("\nfetching stats...")
102
+
async with httpx.AsyncClient(base_url=BASE_URL, timeout=30) as client:
103
+
resp = await client.get("/api/dashboard")
104
+
data = resp.json()
105
+
timing = data.get("timing", {})
106
+
print("\nendpoint count p50 p95")
107
+
print("-" * 40)
108
+
for name in ["search", "similar", "tags", "popular"]:
109
+
t = timing.get(name, {})
110
+
c = t.get("count", 0)
111
+
p50 = t.get("p50_ms", 0)
112
+
p95 = t.get("p95_ms", 0)
113
+
print(f"{name:<12} {c:<6} {p50:>6.0f}ms {p95:>6.0f}ms")
114
+
115
+
116
+
if __name__ == "__main__":
117
+
asyncio.run(main())