nutrition-api/
├── app/
│   ├── main.py              # FastAPI应用入口
│   ├── models/              # Pydantic模型
│   ├── services/            # 业务逻辑
│   ├── routers/             # API路由
│   └── utils/               # 工具函数
├── tests/
├── requirements.txt
└── docker-compose.yml

fastapi==0.104.1
uvicorn[standard]==0.24.0
redis==5.0.1
pydantic==2.5.0
pydantic-settings==2.1.0
python-dotenv==1.0.0
httpx==0.25.1
pytest==7.4.3
pytest-asyncio==0.21.1

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - REDIS_HOST=redis
      - REDIS_PORT=6379
    depends_on:
      - redis

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    command: redis-server --appendonly yes

volumes:
  redis_data:

from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime

class FoodItem(BaseModel):
    """食物基础信息"""
    food_id: str
    name: str
    name_en: Optional[str] = None
    category: str
    brand: Optional[str] = None
    serving_size: float = Field(gt=0, description: "份量大小(克)")
    serving_unit: str = "g"

class Nutrients(BaseModel):
    """营养成分(每100g)"""
    energy: float = Field(ge=0, description: "能量(kcal)")
    protein: float = Field(ge=0, description: "蛋白质(g)")
    fat: float = Field(ge=0, description: "脂肪(g)")
    carbohydrate: float = Field(ge=0, description: "碳水化合物(g)")
    fiber: float = Field(default=0, ge=0, description: "膳食纤维(g)")
    sugar: float = Field(default=0, ge=0, description: "糖(g)")
    sodium: float = Field(default=0, ge=0, description: "钠(mg)")
    potassium: Optional[float] = Field(default=None, ge=0, description: "钾(mg)")
    calcium: Optional[float] = Field(default=None, ge=0, description: "钙(mg)")
    iron: Optional[float] = Field(default=None, ge=0, description: "铁(mg)")
    vitamin_c: Optional[float] = Field(default=None, ge=0, description: "维生素C(mg)")
    vitamin_d: Optional[float] = Field(default=None, ge=0, description: "维生素μg")

class FoodDetail(FoodItem):
    """食物详细信息"""
    nutrients: Nutrients
    ingredients: Optional[List[str]] = None
    allergens: Optional[List[str]] = None
    created_at: datetime
    updated_at: datetime

class NutritionRequest(BaseModel):
    """营养计算请求"""
    items: List[tuple[str, float]] = Field(
        ...,
        description: "[(food_id, 份量g), ...]"
    )

class NutritionSummary(BaseModel):
    """营养汇总"""
    total_energy: float
    total_protein: float
    total_fat: float
    total_carbohydrate: float
    total_fiber: float
    micronutrients: dict[str, float]
    items: List[dict]

import json
import redis.asyncio as redis
from typing import Optional, Any
from datetime import timedelta

class RedisService:
    """异步Redis服务"""

    def __init__(self, host: str = "localhost", port: int = 6379, db: int = 0):
        self.host = host
        self.port = port
        self.db = db
        self._client: Optional[redis.Redis] = None

    async def connect(self):
        """建立连接"""
        self._client = await redis.Redis(
            host=self.host,
            port=self.port,
            db=self.db,
            encoding: "utf-8",
            decode_responses=True
        )
        return self

    async def close(self):
        """关闭连接"""
        if self._client:
            await self._client.close()

    async def get(self, key: str) -> Optional[Any]:
        """获取缓存"""
        if not self._client:
            return None
        value = await self._client.get(key)
        return json.loads(value) if value else None

    async def set(
        self,
        key: str,
        value: Any,
        expire: Optional[timedelta] = None
    ) -> bool:
        """设置缓存"""
        if not self._client:
            return False
        serialized = json.dumps(value, ensure_ascii=False)
        if expire:
            return await self._client.setex(
                key,
                int(expire.total_seconds()),
                serialized
            )
        return await self._client.set(key, serialized)

    async def delete(self, pattern: str) -> int:
        """删除匹配的键"""
        if not self._client:
            return 0
        keys = await self._client.keys(pattern)
        if keys:
            return await self._client.delete(*keys)
        return 0

    async def increment(self, key: str, amount: int = 1) -> int:
        """递增计数器"""
        if not self._client:
            return 0
        return await self._client.incrby(key, amount)

    async def get_ranking(self, key: str, start: int = 0, end: int = -1):
        """获取排行榜"""
        if not self._client:
            return []
        return await self._client.zrevrange(
            key,
            start,
            end,
            withscores=True
        )

from fastapi import APIRouter, HTTPException, Depends, Query
from typing import List, Optional
from datetime import timedelta

from app.models.nutrition import (
    FoodItem,
    FoodDetail,
    NutritionRequest,
    NutritionSummary
)
from app.services.redis_service import RedisService
from app.services.nutrition_service import NutritionService

router = APIRouter(prefix: "/api/v1/nutrition", tags=["营养"])

# 依赖注入
async def get_redis():
    """获取Redis客户端"""
    redis_service = RedisService(host: "redis", port=6379)
    await redis_service.connect()
    try:
        yield redis_service
    finally:
        await redis_service.close()

async def get_nutrition_service(
    redis: RedisService = Depends(get_redis)
) -> NutritionService:
    """获取营养服务"""
    return NutritionService(redis)

@router.get("/foods/search", response_model=List[FoodItem])
async def search_foods(
    q: str = Query(..., min_length=1, description: "搜索关键词"),
    category: Optional[str] = Query(None, description: "食物分类"),
    limit: int = Query(20, ge=1, le=100),
    service: NutritionService = Depends(get_nutrition_service)
):
    """
    搜索食物

    - **q**: 搜索关键词(中英文名称)
    - **category**: 可选，按分类筛选
    - **limit**: 返回数量限制(1-100)
    """
    cache_key = f"search:{q}:{category}:{limit}"
    cached = await service.redis.get(cache_key)
    if cached:
        return cached

    results = await service.search_foods(q, category, limit)
    await service.redis.set(
        cache_key,
        results,
        expire=timedelta(minutes=15)
    )
    return results

@router.get("/foods/{food_id}", response_model=FoodDetail)
async def get_food_detail(
    food_id: str,
    service: NutritionService = Depends(get_nutrition_service)
):
    """
    获取食物详细信息

    - **food_id**: 食物唯一标识
    """
    cache_key = f"food:{food_id}"
    cached = await service.redis.get(cache_key)
    if cached:
        return cached

    food = await service.get_food_detail(food_id)
    if not food:
        raise HTTPException(status_code=404, detail: "食物不存在")

    await service.redis.set(
        cache_key,
        food.dict(),
        expire=timedelta(hours=24)
    )
    return food

@router.post("/calculate", response_model=NutritionSummary)
async def calculate_nutrition(
    request: NutritionRequest,
    service: NutritionService = Depends(get_nutrition_service)
):
    """
    计算食物组合的总营养成分

    - **items**: 食物列表，格式[(food_id, 份量g), ...]
    """
    summary = await service.calculate_nutrition(request.items)
    return summary

@router.get("/categories")
async def get_categories(
    service: NutritionService = Depends(get_nutrition_service)
):
    """获取所有食物分类"""
    cache_key = "categories:all"
    cached = await service.redis.get(cache_key)
    if cached:
        return cached

    categories = await service.get_categories()
    await service.redis.set(
        cache_key,
        categories,
        expire=timedelta(hours=6)
    )
    return categories

@router.get("/popular")
async def get_popular_foods(
    limit: int = Query(10, ge=1, le=50),
    service: NutritionService = Depends(get_nutrition_service)
):
    """
    获取热门食物

    基于查询频率统计
    """
    return await service.get_popular_foods(limit)

@router.post("/foods/{food_id}/view")
async def record_food_view(
    food_id: str,
    service: NutritionService = Depends(get_nutrition_service)
):
    """
    记录食物查看(用于热门统计)

    - **food_id**: 食物唯一标识
    """
    await service.record_view(food_id)
    return {"status": "recorded"}

from typing import List, Optional, Dict, Any
from datetime import datetime

from app.models.nutrition import FoodItem, FoodDetail, NutritionSummary, Nutrients
from app.services.redis_service import RedisService
from app.database.food_database import FOOD_DATABASE

class NutritionService:
    """营养服务核心逻辑"""

    def __init__(self, redis: RedisService):
        self.redis = redis

    async def search_foods(
        self,
        query: str,
        category: Optional[str],
        limit: int
    ) -> List[FoodItem]:
        """搜索食物"""
        results = []
        query_lower = query.lower()

        for food_id, food_data in FOOD_DATABASE.items():
            # 分类筛选
            if category and food_data.get("category") != category:
                continue

            # 关键词匹配
            name = food_data.get("name", "")
            name_en = food_data.get("name_en", "")
            if (query_lower in name.lower() or
                query_lower in name_en.lower()):
                results.append(FoodItem(
                    food_id=food_id,
                    **food_data
                ))
                if len(results) >= limit:
                    break

        # 记录搜索热度
        await self._record_search(query)

        return results

    async def get_food_detail(self, food_id: str) -> Optional[FoodDetail]:
        """获取食物详情"""
        food_data = FOOD_DATABASE.get(food_id)
        if not food_data:
            return None

        return FoodDetail(
            food_id=food_id,
            **food_data
        )

    async def calculate_nutrition(
        self,
        items: List[tuple[str, float]]
    ) -> NutritionSummary:
        """计算营养成分汇总"""
        total = {
            "energy": 0,
            "protein": 0,
            "fat": 0,
            "carbohydrate": 0,
            "fiber": 0,
            "micronutrients": {}
        }
        item_details = []

        for food_id, amount in items:
            food = await self.get_food_detail(food_id)
            if not food:
                continue

            # 按比例计算
            ratio = amount / food.serving_size
            nutrients = food.nutrients

            item_total = {
                "food_id": food_id,
                "name": food.name,
                "amount": amount,
                "energy": round(nutrients.energy * ratio, 1),
                "protein": round(nutrients.protein * ratio, 1),
                "fat": round(nutrients.fat * ratio, 1),
                "carbohydrate": round(nutrients.carbohydrate * ratio, 1)
            }
            item_details.append(item_total)

            # 累加总量
            total["energy"] += item_total["energy"]
            total["protein"] += item_total["protein"]
            total["fat"] += item_total["fat"]
            total["carbohydrate"] += item_total["carbohydrate"]
            total["fiber"] += round(nutrients.fiber * ratio, 1)

        return NutritionSummary(
            total_energy=round(total["energy"], 1),
            total_protein=round(total["protein"], 1),
            total_fat=round(total["fat"], 1),
            total_carbohydrate=round(total["carbohydrate"], 1),
            total_fiber=round(total["fiber"], 1),
            micronutrients=total["micronutrients"],
            items=item_details
        )

    async def get_categories(self) -> List[Dict[str, Any]]:
        """获取所有分类"""
        categories = set()
        for food in FOOD_DATABASE.values():
            categories.add(food.get("category", "其他"))

        return [
            {"name": cat, "count": sum(
                1 for f in FOOD_DATABASE.values()
                if f.get("category") == cat
            )}
            for cat in sorted(categories)
        ]

    async def get_popular_foods(self, limit: int) -> List[Dict[str, Any]]:
        """获取热门食物"""
        ranking = await self.redis.get_ranking(
            "food:views",
            0,
            limit - 1
        )

        results = []
        for food_id, score in ranking:
            food = FOOD_DATABASE.get(food_id)
            if food:
                results.append({
                    "food_id": food_id,
                    "name": food["name"],
                    "views": int(score)
                })

        return results

    async def record_view(self, food_id: str):
        """记录查看次数"""
        await self.redis._client.zincrby("food:views", 1, food_id)

    async def _record_search(self, query: str):
        """记录搜索热度"""
        await self.redis._client.zincrby("search:trending", 1, query)

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import nutrition
from app.services.redis_service import RedisService

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    # 启动时
    redis = RedisService(host: "redis", port=6379)
    await redis.connect()
    app.state.redis = redis
    yield
    # 关闭时
    await redis.close()

app = FastAPI(
    title: "营养信息API",
    description: "高性能营养数据查询和计算服务",
    version: "1.0.0",
    lifespan=lifespan
)

# CORS配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(nutrition.router)

@app.get("/health")
async def health_check():
    """健康检查"""
    return {"status": "healthy", "service": "nutrition-api"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "main:app",
        host: "0.0.0.0",
        port=8000,
        reload=True
    )

import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_search_foods():
    async with AsyncClient(app=app, base_url: "http://test") as client:
        response = await client.get(
            "/api/v1/nutrition/foods/search",
            params={"q": "苹果", "limit": 10}
        )
        assert response.status_code == 200
        data = response.json()
        assert isinstance(data, list)
        assert len(data) <= 10

@pytest.mark.asyncio
async def test_get_food_detail():
    async with AsyncClient(app=app, base_url: "http://test") as client:
        response = await client.get("/api/v1/nutrition/foods/apple_001")
        assert response.status_code == 200
        data = response.json()
        assert "food_id" in data
        assert "nutrients" in data

@pytest.mark.asyncio
async def test_calculate_nutrition():
    async with AsyncClient(app=app, base_url: "http://test") as client:
        response = await client.post(
            "/api/v1/nutrition/calculate",
            json={
                "items": [
                    ["apple_001", 150],
                    ["rice_001", 100]
                ]
            }
        )
        assert response.status_code == 200
        data = response.json()
        assert "total_energy" in data
        assert "total_protein" in data

GET /api/v1/nutrition/foods/search?q=苹果&category=水果&limit=20

GET /api/v1/nutrition/foods/{food_id}

POST /api/v1/nutrition/calculate
Content-Type: application/json

{
  "items": [
    ["apple_001", 150],
    ["rice_001", 100]
  ]
}

GET /api/v1/nutrition/categories

GET /api/v1/nutrition/popular?limit=10