feat(api): sanguo_api.main 容器入口(build_app + create_app factory,挂 SPA)

This commit is contained in:
2026-07-07 00:37:38 +08:00
parent 759acf2f6c
commit 4e86d9e00e
2 changed files with 129 additions and 0 deletions
+90
View File
@@ -0,0 +1,90 @@
"""Container uvicorn entrypoint.
Reads ``config/backtest.yaml`` to build the FastAPI app (db paths + auth +
workers) and exposes a module-level ``app`` for ``uvicorn sanguo_api.main:app``.
Optionally mounts the built Vue SPA at ``/`` (history fallback).
Task S0.1. Imported by uvicorn (sanguo_api.main:app) and tests/api/test_main.py.
"""
from __future__ import annotations
import os
from pathlib import Path
import yaml
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from starlette.responses import FileResponse
from .app import create_app as _create_app
def _load_config(config_path: str) -> dict:
p = Path(config_path)
if not p.exists():
return {}
with open(p, "r", encoding="utf-8") as f:
return yaml.safe_load(f) or {}
def build_app(config_path: str = "config/backtest.yaml", static_dir: str | None = None) -> FastAPI:
"""Build the FastAPI app from a backtest.yaml config file.
Args:
config_path: Path to backtest.yaml (db_path/file_dir/auth/pool).
static_dir: Optional directory of built Vue SPA files; mounted at ``/``.
Returns:
A configured FastAPI application with /api/v1 routes and optional SPA.
"""
cfg = _load_config(config_path)
bt = cfg.get("backtest", {})
auth = cfg.get("auth", {})
pool = cfg.get("pool", {})
db_path = bt.get("db_path", "/tmp/backtest_results.db")
file_dir = bt.get("file_dir", "/tmp/backtest_files")
auth_config = None
if auth:
auth_config = {
"username": auth.get("username", "admin"),
"password_hash": auth.get("password_hash", ""),
"jwt_secret": auth.get("jwt_secret", "change-me"),
"expire_minutes": auth.get("token_expire_minutes", 60),
}
max_workers = pool.get("max_workers", bt.get("max_workers", 2))
app = _create_app(
db_path=db_path,
file_dir=file_dir,
auth_config=auth_config,
max_workers=max_workers,
)
# SPA static mount with history fallback. Register the root route BEFORE
# mounting StaticFiles at "/" so the explicit handler wins for "/".
if static_dir and os.path.isdir(static_dir):
index_html = os.path.join(static_dir, "index.html")
@app.get("/", include_in_schema=False)
async def _spa_root() -> FileResponse: # noqa: D401
return FileResponse(index_html)
app.mount("/", StaticFiles(directory=static_dir, html=True), name="spa")
return app
def create_app() -> FastAPI:
"""No-arg factory for ``uvicorn sanguo_api.main:create_app --factory``.
Building the app here (rather than at module import) avoids constructing
the Orchestrator's ProcessPoolExecutor at import time, which would recurse
under the ``spawn`` start method. uvicorn calls this factory once at server
startup in the main process.
"""
repo = Path(__file__).resolve().parent.parent
return build_app(
str(repo / "config" / "backtest.yaml"),
static_dir=os.environ.get("SPA_STATIC_DIR", str(repo / "frontend" / "dist")),
)
+39
View File
@@ -0,0 +1,39 @@
"""Tests for sanguo_api.main container entrypoint (app loader + SPA mount).
Covers Task S0.1: build_app(config_path, static_dir) loads backtest.yaml and
mounts SPA static files when the directory exists.
"""
from sanguo_api.main import build_app
def _cfg(tmp_path) -> str:
cfg = tmp_path / "bt.yaml"
cfg.write_text(
"backtest:\n max_workers: 1\n db_path: %s\n file_dir: %s\n"
"api:\n host: 0.0.0.0\n port: 8000\n"
"auth:\n username: admin\n password_hash: x\n jwt_secret: s\n token_expire_minutes: 60\n"
"pool:\n max_workers: 1\n" % (tmp_path / "r.db", tmp_path / "f")
)
return str(cfg)
def test_build_app_has_api_routes(tmp_path):
app = build_app(_cfg(tmp_path))
paths = [getattr(r, "path", "") for r in app.routes]
assert "/api/v1/auth/login" in paths
def test_build_app_mounts_spa_when_static_exists(tmp_path):
spa = tmp_path / "spa"
spa.mkdir()
(spa / "index.html").write_text("<h1>SPA</h1>")
app = build_app(_cfg(tmp_path), static_dir=str(spa))
paths = [getattr(r, "path", "") for r in app.routes]
assert "/" in paths
def test_build_app_no_static_skips_mount(tmp_path):
"""When static dir absent, build must still succeed (no SPA mount)."""
app = build_app(_cfg(tmp_path), static_dir=str(tmp_path / "nope"))
paths = [getattr(r, "path", "") for r in app.routes]
assert "/api/v1/auth/login" in paths