Skip to content

Commit 9e17e5a

Browse files
committed
Added CLI to generate FastAPI route manifest
1 parent 12b4116 commit 9e17e5a

File tree

2 files changed

+139
-0
lines changed

2 files changed

+139
-0
lines changed

pyproject.toml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,7 @@ GitHub = "https://github.com/DiamondLightSource/python-murfey"
8787
"murfey.decrypt_password" = "murfey.cli.decrypt_db_password:run"
8888
"murfey.generate_key" = "murfey.cli.generate_crypto_key:run"
8989
"murfey.generate_password" = "murfey.cli.generate_db_password:run"
90+
"murfey.generate_route_manifest" = "murfey.cli.generate_route_manifest:run"
9091
"murfey.instrument_server" = "murfey.instrument_server:run"
9192
"murfey.repost_failed_calls" = "murfey.cli.repost_failed_calls:run"
9293
"murfey.server" = "murfey.server.run:run"
Lines changed: 138 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,138 @@
1+
"""
2+
CLI to generate a manifest of the FastAPI router paths present in both the instrument
3+
server and backend server to enable lookup of the URLs based on function name.
4+
"""
5+
6+
import importlib
7+
import inspect
8+
import pkgutil
9+
from argparse import ArgumentParser
10+
from pathlib import Path
11+
from types import ModuleType
12+
from typing import Any
13+
14+
import yaml
15+
from fastapi import APIRouter
16+
17+
import murfey
18+
19+
20+
def find_routers(name: str) -> dict[str, APIRouter]:
21+
22+
def _extract_routers_from_module(module: ModuleType):
23+
routers = {}
24+
for name, obj in inspect.getmembers(module):
25+
if isinstance(obj, APIRouter):
26+
module_path = module.__name__
27+
key = f"{module_path}.{name}"
28+
routers[key] = obj
29+
return routers
30+
31+
routers = {}
32+
33+
# Import the module or package
34+
try:
35+
root = importlib.import_module(name)
36+
except ImportError:
37+
raise ImportError(
38+
f"Cannot import '{name}'. Please ensure that you've installed all the "
39+
"dependencies for the client, instrument server, and backend server "
40+
"before running this command."
41+
)
42+
43+
# If it's a package, walk through submodules and extract routers from each
44+
if hasattr(root, "__path__"):
45+
module_list = pkgutil.walk_packages(root.__path__, prefix=name + ".")
46+
for _, module_name, _ in module_list:
47+
try:
48+
module = importlib.import_module(module_name)
49+
except ImportError:
50+
raise ImportError(
51+
f"Cannot import '{module_name}'. Please ensure that you've "
52+
"installed all the dependencies for the client, instrument "
53+
"server, and backend server before running this command."
54+
)
55+
56+
routers.update(_extract_routers_from_module(module))
57+
58+
# Extract directly from single module
59+
else:
60+
routers.update(_extract_routers_from_module(root))
61+
62+
return routers
63+
64+
65+
def get_route_manifest(routers: dict[str, APIRouter]):
66+
67+
manifest = {}
68+
69+
for router_name, router in routers.items():
70+
routes = []
71+
for route in router.routes:
72+
path_params = []
73+
for param in route.dependant.path_params:
74+
param_type = param.type_ if param.type_ is not None else Any
75+
param_info = {
76+
"name": param.name if hasattr(param, "name") else "",
77+
"type": (
78+
param_type.__name__
79+
if hasattr(param_type, "__name__")
80+
else str(param_type)
81+
),
82+
}
83+
path_params.append(param_info)
84+
route_info = {
85+
"path": route.path if hasattr(route, "path") else "",
86+
"function": route.name if hasattr(route, "name") else "",
87+
"path_params": path_params,
88+
"methods": list(route.methods) if hasattr(route, "methods") else [],
89+
}
90+
routes.append(route_info)
91+
manifest[router_name] = routes
92+
return manifest
93+
94+
95+
def run():
96+
# Set up additional args
97+
parser = ArgumentParser()
98+
parser.add_argument(
99+
"--debug",
100+
action="store_true",
101+
default=False,
102+
help=("Outputs the modules being inspected when creating the route manifest"),
103+
)
104+
args = parser.parse_args()
105+
106+
# Find routers
107+
print("Finding routers...")
108+
routers = {
109+
**find_routers("murfey.instrument_server.api"),
110+
**find_routers("murfey.server.api"),
111+
}
112+
# Generate the manifest
113+
print("Extracting route information")
114+
manifest = get_route_manifest(routers)
115+
116+
# Verify
117+
if args.debug:
118+
for router_name, routes in manifest.items():
119+
print(f"Routes found in {router_name!r}")
120+
for route in routes:
121+
for key, value in route.items():
122+
print(f"\t{key}: {value}")
123+
print()
124+
125+
# Save the manifest
126+
murfey_dir = Path(murfey.__path__[0])
127+
manifest_file = murfey_dir / "util" / "route_manifest.yaml"
128+
with open(manifest_file, "w") as file:
129+
yaml.dump(manifest, file, default_flow_style=False, sort_keys=False)
130+
print(
131+
"Route manifest for instrument and backend servers saved to "
132+
f"{str(manifest_file)!r}"
133+
)
134+
exit()
135+
136+
137+
if __name__ == "__main__":
138+
run()

0 commit comments

Comments
 (0)