📝🎨 More inline docs; small adjustments

This commit is contained in:
Brian Wiborg 2025-10-01 21:55:01 +02:00
parent 16f15a3d65
commit 00e18af8fd
No known key found for this signature in database

View file

@ -4,6 +4,7 @@ import importlib.util
import json import json
import pkgutil import pkgutil
import sys import sys
from http import HTTPStatus
from pathlib import Path from pathlib import Path
from types import ModuleType from types import ModuleType
from typing import Any, Dict, Generator, List, Optional, Type from typing import Any, Dict, Generator, List, Optional, Type
@ -21,8 +22,9 @@ class Project:
""" """
Project runtime loader + Tortoise/Aerich integration. Project runtime loader + Tortoise/Aerich integration.
- injects builtin apps as ohmyapi_<name> - aliases builtin apps as ohmyapi_<name>
- builds unified tortoise config for runtime - loads all INSTALLED_APPS into scope
- builds unified tortoise config for ORM runtime
- provides makemigrations/migrate methods using Aerich Command API - provides makemigrations/migrate methods using Aerich Command API
""" """
@ -34,8 +36,8 @@ class Project:
if str(self.project_path) not in sys.path: if str(self.project_path) not in sys.path:
sys.path.insert(0, str(self.project_path)) sys.path.insert(0, str(self.project_path))
# Pre-register builtin apps as ohmyapi_<name>. # Alias builtin apps as ohmyapi_<name>.
# This makes all builtin apps easily loadable via f"ohmyapi_{app_name}". # We need this, because Tortoise app-names may not include dots `.`.
spec = importlib.util.find_spec("ohmyapi.builtin") spec = importlib.util.find_spec("ohmyapi.builtin")
if spec and spec.submodule_search_locations: if spec and spec.submodule_search_locations:
for _, modname, _ in pkgutil.iter_modules(spec.submodule_search_locations): for _, modname, _ in pkgutil.iter_modules(spec.submodule_search_locations):
@ -200,7 +202,8 @@ class App:
self.project = project self.project = project
self.name = name self.name = name
# Reference to this app's models modules. # Reference to this app's models modules. Tortoise needs to know the
# modules where to lookup models for this app.
self._models: Dict[str, ModuleType] = {} self._models: Dict[str, ModuleType] = {}
# Reference to this app's routes modules. # Reference to this app's routes modules.
@ -219,14 +222,20 @@ class App:
return self.__repr__() return self.__repr__()
def __load_models(self, mod_name: str): def __load_models(self, mod_name: str):
"""
Recursively scan through a module and collect all models.
If the module is a package, iterate through its submodules.
"""
# An app may come without any models.
try: try:
importlib.import_module(mod_name) importlib.import_module(mod_name)
except ModuleNotFoundError: except ModuleNotFoundError:
print(f"no models detected: {mod_name}") print(f"no models detected: {mod_name}")
return return
# Acoid duplicates.
visited: set[str] = set() visited: set[str] = set()
out: Dict[str, ModuleType] = {}
def walk(mod_name: str): def walk(mod_name: str):
mod = importlib.import_module(mod_name) mod = importlib.import_module(mod_name)
@ -240,7 +249,7 @@ class App:
and issubclass(value, Model) and issubclass(value, Model)
and not name == Model.__name__ and not name == Model.__name__
): ):
out[mod_name] = out.get(mod_name, []) + [value] self._models[mod_name] = self._models.get(mod_name, []) + [value]
# if it's a package, recurse into submodules # if it's a package, recurse into submodules
if hasattr(mod, "__path__"): if hasattr(mod, "__path__"):
@ -249,18 +258,24 @@ class App:
): ):
walk(subname) walk(subname)
# Walk the walk.
walk(mod_name) walk(mod_name)
self._models = out
def __load_routes(self, mod_name: str): def __load_routes(self, mod_name: str):
"""
Recursively scan through a module and collect all APIRouters.
If the module is a package, iterate through all its submodules.
"""
# An app may come without any routes.
try: try:
importlib.import_module(mod_name) importlib.import_module(mod_name)
except ModuleNotFound: except ModuleNotFound:
print(f"no routes detected: {mod_name}") print(f"no routes detected: {mod_name}")
return return
# Avoid duplicates.
visited: set[str] = set() visited: set[str] = set()
out: Dict[str, ModuleType] = {}
def walk(mod_name: str): def walk(mod_name: str):
mod = importlib.import_module(mod_name) mod = importlib.import_module(mod_name)
@ -270,7 +285,7 @@ class App:
for name, value in vars(mod).copy().items(): for name, value in vars(mod).copy().items():
if isinstance(value, APIRouter) and not name == APIRouter.__name__: if isinstance(value, APIRouter) and not name == APIRouter.__name__:
out[mod_name] = out.get(mod_name, []) + [value] self._routers[mod_name] = self._routers.get(mod_name, []) + [value]
# if it's a package, recurse into submodules # if it's a package, recurse into submodules
if hasattr(mod, "__path__"): if hasattr(mod, "__path__"):
@ -280,22 +295,17 @@ class App:
submod = importlib.import_module(subname) submod = importlib.import_module(subname)
walk(submod) walk(submod)
# Walk the walk.
walk(mod_name) walk(mod_name)
self._routers = out
def __serialize_route(self, route): def __serialize_route(self, route):
"""Convert APIRoute to JSON-serializable dict.""" """
Convert APIRoute to JSON-serializable dict.
"""
return { return {
"path": route.path, "path": route.path,
"name": route.name, "method": list(route.methods)[0],
"methods": list(route.methods), "endpoint": f"{route.endpoint.__module__}.{route.endpoint.__name__}",
"endpoint": route.endpoint.__name__, # just the function name
"response_model": (
getattr(route, "response_model", None).__name__
if getattr(route, "response_model", None)
else None
),
"tags": getattr(route, "tags", None),
} }
def __serialize_router(self): def __serialize_router(self):
@ -303,6 +313,9 @@ class App:
@property @property
def models(self) -> List[ModuleType]: def models(self) -> List[ModuleType]:
"""
Return a list of all loaded models.
"""
out = [] out = []
for module in self._models: for module in self._models:
for model in self._models[module]: for model in self._models[module]:
@ -313,6 +326,9 @@ class App:
@property @property
def routes(self): def routes(self):
"""
Return an APIRouter with all loaded routes.
"""
router = APIRouter() router = APIRouter()
for routes_mod in self._routers: for routes_mod in self._routers:
for r in self._routers[routes_mod]: for r in self._routers[routes_mod]:
@ -320,6 +336,9 @@ class App:
return router.routes return router.routes
def dict(self) -> Dict[str, Any]: def dict(self) -> Dict[str, Any]:
"""
Convenience method for serializing the runtime data.
"""
return { return {
"models": [ "models": [
f"{self.name}.{m.__name__}" for m in self.models[f"{self.name}.models"] f"{self.name}.{m.__name__}" for m in self.models[f"{self.name}.models"]