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