01
from functools import lru_cache
02

03
from pydantic import SecretStr
04
from pydantic_settings import BaseSettings, SettingsConfigDict
05

06

07
class Settings(BaseSettings):
08
    """集中管理应用配置，并负责从环境变量和 .env 读取值。"""
09

10
    model_config = SettingsConfigDict(
11
        env_file=".env",
12
        env_file_encoding="utf-8",
13
        extra="ignore",
14
    )
15

16
    app_name: str = "FastAPI LangChain"
17
    deepseek_model: str = "deepseek-v4-flash"
18
    deepseek_base_url: str = "https://api.deepseek.com"
19
    deepseek_api_key: SecretStr | None = None
20
    cors_allow_origins: str = "*"
21
    cors_allow_methods: list[str] = [
22
        "GET",
23
        "POST",
24
        "PUT",
25
        "DELETE",
26
        "OPTIONS",
27
    ]
28
    cors_allow_headers: list[str] = [
29
        "Content-Type",
30
        "Authorization",
31
        "X-Request-Id",
32
    ]
33
    cors_expose_headers: list[str] = [
34
        "X-Request-Id",
35
        "X-Process-Time",
36
    ]
37

38
    @property
39
    def allowed_origins(self) -> list[str]:
40
        """把教学环境中的逗号分隔配置转换成 CORS 需要的列表。"""
41

42
        origins = [
43
            origin.strip()
44
            for origin in self.cors_allow_origins.split(",")
45
            if origin.strip()
46
        ]
47
        return origins or ["*"]
48

49
    @property
50
    def allow_credentials(self) -> bool:
51
        """通配来源时关闭凭证，避免违反浏览器 CORS 规则。"""
52

53
        return "*" not in self.allowed_origins
54

55

56
@lru_cache
57
def get_settings() -> Settings:
58
    """在进程内复用配置对象，避免每个请求重复读取 .env。"""
59

60
    return Settings()

1
from pydantic import BaseModel, Field
2

3
class ChatRequest(BaseModel):
4
    question: str = Field(min_length=1, max_length=4000)
5

6
class ChatResponse(BaseModel):
7
    answer: str
8
    model: str

01
from langchain_core.language_models import BaseChatModel
02

03
from app.repositories.chat import ChatRepository
04

05
class ChatService:
06
    def __init__(
07
        self,
08
        model: BaseChatModel,
09
        model_name: str,
10
        repository: ChatRepository,
11
    ):
12
        self._model = model
13
        self.model_name = model_name
14
        self._repository = repository
15

16
    async def answer(self, question: str) -> str:
17
        message = await self._model.ainvoke(question)
18
        await self._repository.save_turn(question, message.text)
19
        await self._repository.commit()
20
        return message.text

01
from collections.abc import AsyncIterator
02
from contextlib import asynccontextmanager
03

04
from fastapi import FastAPI
05
from langchain_openai import ChatOpenAI
06
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
07

08
from app.core.config import Settings
09

10
@asynccontextmanager
11
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
12
    settings: Settings = app.state.settings
13

14
    app.state.model = ChatOpenAI(
15
        model=settings.deepseek_model,
16
        api_key=settings.deepseek_api_key.get_secret_value(),
17
        base_url=settings.deepseek_base_url,
18
    )
19

20
    app.state.db_engine = create_async_engine(
21
        settings.database_url,
22
        pool_pre_ping=True,
23
    )
24
    app.state.session_factory = async_sessionmaker(
25
        app.state.db_engine,
26
        expire_on_commit=False,
27
    )
28

29
    yield
30

31
    await app.state.db_engine.dispose()

01
from collections.abc import AsyncIterator
02
from typing import Annotated
03

04
from fastapi import Depends, Request
05
from langchain_core.language_models import BaseChatModel
06
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
07

08
from app.core.config import Settings
09
from app.repositories.chat import ChatRepository
10
from app.services.chat import ChatService
11

12
async def get_db_session(
13
    request: Request,
14
) -> AsyncIterator[AsyncSession]:
15
    session_factory: async_sessionmaker[AsyncSession] = (
16
        request.app.state.session_factory
17
    )
18

19
    async with session_factory() as session:
20
        yield session
21

22
SessionDep = Annotated[AsyncSession, Depends(get_db_session)]
23

24
def get_chat_service(
25
    request: Request,
26
    session: SessionDep,
27
) -> ChatService:
28
    settings: Settings = request.app.state.settings
29
    model: BaseChatModel = request.app.state.model
30

31
    return ChatService(
32
        model=model,
33
        model_name=settings.deepseek_model,
34
        repository=ChatRepository(session),
35
    )
36

37
Chat = Annotated[ChatService, Depends(get_chat_service)]

01
from fastapi import APIRouter
02

03
from app.api.dependencies import Chat
04
from app.schemas.chat import ChatRequest, ChatResponse
05

06
router = APIRouter(prefix="/chat", tags=["chat"])
07

08
@router.post("", response_model=ChatResponse)
09
async def chat(
10
    payload: ChatRequest,
11
    service: Chat,
12
) -> ChatResponse:
13
    answer = await service.answer(payload.question)
14

15
    return ChatResponse(
16
        answer=answer,
17
        model=service.model_name,
18
    )

01
from fastapi import APIRouter
02

03
from . import deps, req, res, validation
04
from .routes import health
05

06
# 需要版本管理的业务接口统一挂载在 /api/v1 下。
07
api_v1_router = APIRouter(prefix="/api/v1")
08
api_v1_router.include_router(health.router)
09
api_v1_router.include_router(deps.router)
10
api_v1_router.include_router(req.router)
11
api_v1_router.include_router(res.router)
12
api_v1_router.include_router(validation.router)
13

14
# 根路由负责聚合不同版本的 API，供应用入口一次性接入。
15
router = APIRouter()
16
router.include_router(api_v1_router)

01
from fastapi import FastAPI
02

03
from app.core.config import Settings, get_settings
04
from app.middleware import register_middleware
05
from src.fastapi.api.router import router as api_router
06

07

08
async def root() -> dict[str, str]:
09
    return {"message": "Hello FastAPI"}
10

11

12
def create_app(settings: Settings | None = None) -> FastAPI:
13
    """组装 FastAPI 应用，允许测试传入独立配置。"""
14

15
    current_settings = settings or get_settings()
16
    app = FastAPI(
17
        title=current_settings.app_name,
18
        version="0.1.0",
19
    )
20
    app.state.settings = current_settings
21

22
    register_middleware(app, current_settings)
23

24
    app.include_router(api_router)
25
    app.add_api_route("/", root, methods=["GET"])
26
    return app

1
from app.factory import create_app
2

3
app = create_app()