1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
#!/usr/bin/env python3
import argparse
import gc
import json
import random
import statistics
import time
from dataclasses import dataclass
from typing import Any
from typing import Dict
from typing import List
from jsonschema_path import SchemaPath
from openapi_core.templating.paths.finders import APICallPathFinder
@dataclass(frozen=True)
class Result:
paths: int
templates_ratio: float
lookups: int
repeats: int
warmup: int
seconds: List[float]
def as_dict(self) -> Dict[str, Any]:
return {
"paths": self.paths,
"templates_ratio": self.templates_ratio,
"lookups": self.lookups,
"repeats": self.repeats,
"warmup": self.warmup,
"seconds": self.seconds,
"median_s": statistics.median(self.seconds),
"mean_s": statistics.mean(self.seconds),
"stdev_s": statistics.pstdev(self.seconds),
"ops_per_sec_median": self.lookups
/ statistics.median(self.seconds),
}
def build_spec(paths: int, templates_ratio: float) -> SchemaPath:
# Mix of exact and templated paths.
# Keep it minimal so we measure finder cost, not schema complexity.
tmpl = int(paths * templates_ratio)
exact = paths - tmpl
paths_obj: Dict[str, Any] = {}
# Exact paths (fast case)
for i in range(exact):
p = f"/resource/{i}/sub"
paths_obj[p] = {"get": {"responses": {"200": {"description": "ok"}}}}
# Template paths (slow case)
for i in range(tmpl):
p = f"/resource/{i}" + "/{item_id}/sub/{sub_id}"
paths_obj[p] = {"get": {"responses": {"200": {"description": "ok"}}}}
spec_dict = {
"openapi": "3.0.0",
"info": {"title": "bench", "version": "0"},
"servers": [{"url": "http://example.com"}],
"paths": paths_obj,
}
return SchemaPath.from_dict(spec_dict)
def build_urls(
paths: int, templates_ratio: float, lookups: int, seed: int
) -> List[str]:
rnd = random.Random(seed)
tmpl = int(paths * templates_ratio)
exact = paths - tmpl
urls: List[str] = []
for _ in range(lookups):
# 50/50 choose from each population, weighted by how many exist
if tmpl > 0 and (exact == 0 or rnd.random() < (tmpl / paths)):
i = rnd.randrange(tmpl) # matches template bucket
item_id = rnd.randrange(1_000_000)
sub_id = rnd.randrange(1_000_000)
urls.append(
f"http://example.com/resource/{i}/{item_id}/sub/{sub_id}"
)
else:
i = rnd.randrange(exact) if exact > 0 else 0
urls.append(f"http://example.com/resource/{i}/sub")
return urls
def run_once(finder: APICallPathFinder, urls: List[str]) -> float:
t0 = time.perf_counter()
for u in urls:
finder.find("get", u)
return time.perf_counter() - t0
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--paths", type=int, default=2000)
ap.add_argument("--templates-ratio", type=float, default=0.6)
ap.add_argument("--lookups", type=int, default=100_000)
ap.add_argument("--repeats", type=int, default=7)
ap.add_argument("--warmup", type=int, default=2)
ap.add_argument("--seed", type=int, default=1)
ap.add_argument("--output", type=str, default="")
ap.add_argument("--no-gc", action="store_true")
args = ap.parse_args()
spec = build_spec(args.paths, args.templates_ratio)
finder = APICallPathFinder(spec)
urls = build_urls(
args.paths, args.templates_ratio, args.lookups, args.seed
)
if args.no_gc:
gc.disable()
# Warmup (JIT-less, but warms caches, alloc patterns, etc.)
for _ in range(args.warmup):
run_once(finder, urls)
seconds: List[float] = []
for _ in range(args.repeats):
seconds.append(run_once(finder, urls))
if args.no_gc:
gc.enable()
result = Result(
paths=args.paths,
templates_ratio=args.templates_ratio,
lookups=args.lookups,
repeats=args.repeats,
warmup=args.warmup,
seconds=seconds,
)
payload = result.as_dict()
print(json.dumps(payload, indent=2, sort_keys=True))
if args.output:
with open(args.output, "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2, sort_keys=True)
if __name__ == "__main__":
main()
|
