01
import time
02

03
from fastapi import Request
04

05

06
async def add_process_time(request: Request, call_next):
07
    """记录请求处理耗时，并把结果写入响应头。"""
08

09
    started_at = time.perf_counter()
10
    response = await call_next(request)
11
    duration = time.perf_counter() - started_at
12
    response.headers["X-Process-Time"] = f"{duration:.6f}"
13
    return response

01
import re
02
from uuid import uuid4
03

04
from fastapi import Request
05

06

07
request_id_pattern = re.compile(r"^[A-Za-z0-9._:-]{1,128}$")
08

09

10
async def add_request_id(request: Request, call_next):
11
    """复用合法的客户端 Request ID，否则生成新的随机 ID。"""
12

13
    candidate = request.headers.get("X-Request-Id", "")
14
    request_id = (
15
        candidate
16
        if request_id_pattern.fullmatch(candidate)
17
        else uuid4().hex
18
    )
19
    request.state.request_id = request_id
20

21
    response = await call_next(request)
22
    response.headers["X-Request-Id"] = request_id
23
    return response

01
from fastapi import FastAPI
02
from fastapi.middleware.cors import CORSMiddleware
03

04
from app.core.config import Settings
05

06
from .request_id import add_request_id
07
from .security_headers import add_security_headers
08
from .timing import add_process_time
09

10

11
def register_middleware(app: FastAPI, settings: Settings) -> None:
12
    """按统一顺序注册应用级中间件。"""
13

14
    app.middleware("http")(add_security_headers)
15
    app.middleware("http")(add_request_id)
16
    app.middleware("http")(add_process_time)
17

18
    # 最后注册 CORS，使它位于最外层，错误响应也能获得跨域响应头。
19
    app.add_middleware(
20
        CORSMiddleware,
21
        allow_origins=settings.allowed_origins,
22
        allow_credentials=settings.allow_credentials,
23
        allow_methods=settings.cors_allow_methods,
24
        allow_headers=settings.cors_allow_headers,
25
        expose_headers=settings.cors_expose_headers,
26
    )

01
from fastapi import FastAPI
02
from fastapi.middleware.cors import CORSMiddleware
03

04
app = FastAPI()
05

06
app.add_middleware(
07
    CORSMiddleware,
08
    allow_origins=[
09
        "https://langchainpy.usehook.cn",
10
        "http://localhost:3000",
11
    ],
12
    allow_credentials=False,
13
    allow_methods=["GET", "POST", "OPTIONS"],
14
    allow_headers=[
15
        "Content-Type",
16
        "X-Local-Token",
17
        "X-Request-Id",
18
    ],
19
    expose_headers=["X-Request-Id"],
20
)

01
from fastapi import Request
02

03

04
async def add_security_headers(request: Request, call_next):
05
    """为 API 响应添加不会依赖部署环境的基础安全头。"""
06

07
    response = await call_next(request)
08
    response.headers.setdefault("X-Content-Type-Options", "nosniff")
09
    if request.url.path.startswith("/api/"):
10
        response.headers.setdefault("Cache-Control", "no-store")
11
    return response

01
import time
02
from uuid import uuid4
03

04
from fastapi import FastAPI, Request
05
from fastapi.middleware.cors import CORSMiddleware
06

07

08
def create_app() -> FastAPI:
09
    app = FastAPI()
10

11
    app.add_middleware(
12
        CORSMiddleware,
13
        allow_origins=["https://langchainpy.usehook.cn"],
14
        allow_credentials=False,
15
        allow_methods=["GET"],
16
        allow_headers=[],
17
        expose_headers=["X-Request-Id", "X-Process-Time"],
18
    )
19

20
    @app.middleware("http")
21
    async def add_request_context(request: Request, call_next):
22
        started_at = time.perf_counter()
23
        request_id = uuid4().hex
24
        request.state.request_id = request_id
25

26
        response = await call_next(request)
27
        response.headers["X-Request-Id"] = request_id
28
        response.headers["X-Process-Time"] = (
29
            f"{time.perf_counter() - started_at:.6f}"
30
        )
31
        return response
32

33
    @app.get("/health")
34
    async def health() -> dict[str, str]:
35
        return {"status": "ok"}
36

37
    return app
38

39

40
app = create_app()