import httpx
import os
from typing import Optional, List, Dict, Any, Union
from pydantic import BaseModel, Field, validator
class NutrientDetail(BaseModel):
id: int
number: str
name: str
rank: Optional[int] = None
unitName: str
class FoodNutrientDerivation(BaseModel):
id: Optional[int] = None
code: Optional[str] = None
description: Optional[str] = None
foodNutrientSource: Optional[Dict[str, Any]] = None
# Flexible FoodNutrient model that can handle both nested and flat structures
class FoodNutrient(BaseModel):
type: Optional[str] = None
id: Optional[int] = None
# Nested structure fields (for detailed responses)
nutrient: Optional[NutrientDetail] = None
amount: Optional[float] = None
foodNutrientDerivation: Optional[FoodNutrientDerivation] = None
# Flat structure fields (for some responses)
number: Optional[str] = None
name: Optional[str] = None
unitName: Optional[str] = None
value: Optional[float] = None
# Additional fields that might be present
nutrientId: Optional[int] = None
nutrientName: Optional[str] = None
nutrientNumber: Optional[str] = None
percentDailyValue: Optional[float] = None
foodNutrientId: Optional[int] = None
@validator('amount', always=True)
def set_amount_from_value(cls, v, values):
"""If amount is None but value is present, use value as amount."""
if v is None and 'value' in values and values['value'] is not None:
return values['value']
return v
@validator('nutrient', always=True)
def create_nutrient_from_flat_fields(cls, v, values):
"""If nutrient is None but flat fields are present, create nutrient object."""
if v is None:
# Check if we have flat nutrient fields
number = values.get('number') or values.get('nutrientNumber')
name = values.get('name') or values.get('nutrientName')
unit_name = values.get('unitName')
nutrient_id = values.get('nutrientId')
if number and name and unit_name:
return NutrientDetail(
id=nutrient_id or 0, # Use 0 as default if not provided
number=number,
name=name,
unitName=unit_name
)
return v
# For search responses (flat structure)
class SearchFoodNutrient(BaseModel):
nutrientId: int
nutrientName: str
nutrientNumber: str
unitName: str
value: float
percentDailyValue: Optional[float] = None
foodNutrientId: Optional[int] = None
class FoodSearchResult(BaseModel):
fdcId: int
description: str
dataType: str
gtinUpc: Optional[str] = None
publishedDate: Optional[str] = None
brandOwner: Optional[str] = None
ingredients: Optional[str] = None
servingSize: Optional[float] = None
servingSizeUnit: Optional[str] = None
foodNutrients: List[SearchFoodNutrient] = []
class SearchResponse(BaseModel):
foods: List[FoodSearchResult]
totalHits: int
currentPage: int
totalPages: int
pageSize: Optional[int] = None
class FoodDetailResponse(BaseModel):
fdcId: int
description: str
dataType: str
foodClass: Optional[str] = None
gtinUpc: Optional[str] = None
publishedDate: Optional[str] = None
brandOwner: Optional[str] = None
ingredients: Optional[str] = None
servingSize: Optional[float] = None
servingSizeUnit: Optional[str] = None
foodNutrients: List[FoodNutrient] = []
# Make these fields flexible to handle both string and int types
ndbNumber: Optional[Union[str, int]] = None
foodCode: Optional[Union[str, int]] = None
# Additional fields from the debug output
publicationDate: Optional[str] = None
modifiedDate: Optional[str] = None
availableDate: Optional[str] = None
discontinuedDate: Optional[str] = None
brandedFoodCategory: Optional[str] = None
householdServingFullText: Optional[str] = None
marketCountry: Optional[str] = None
dataSource: Optional[str] = None
foodComponents: Optional[List[Dict[str, Any]]] = None
foodAttributes: Optional[List[Dict[str, Any]]] = None
foodPortions: Optional[List[Dict[str, Any]]] = None
foodUpdateLog: Optional[List[Dict[str, Any]]] = None
labelNutrients: Optional[Dict[str, Any]] = None
class FoodDataCentralAPI:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.nal.usda.gov/fdc/v1"
self.client = httpx.AsyncClient()
async def search_foods(
self,
query: str,
data_type: Optional[List[str]] = None,
page_size: int = 50,
page_number: int = 1,
sort_by: Optional[str] = None,
sort_order: Optional[str] = None,
brand_owner: Optional[str] = None,
start_date: Optional[str] = None,
end_date: Optional[str] = None
) -> SearchResponse:
"""Search for foods using the FDC API"""
params = {
"api_key": self.api_key,
"query": query,
"pageSize": page_size,
"pageNumber": page_number
}
if data_type:
params["dataType"] = data_type
if sort_by:
params["sortBy"] = sort_by
if sort_order:
params["sortOrder"] = sort_order
if brand_owner:
params["brandOwner"] = brand_owner
if start_date:
params["startDate"] = start_date
if end_date:
params["endDate"] = end_date
response = await self.client.get(
f"{self.base_url}/foods/search",
params=params
)
response.raise_for_status()
return SearchResponse(**response.json())
async def get_food_details(
self,
fdc_id: int,
format_type: str = "full",
nutrients: Optional[List[int]] = None
) -> FoodDetailResponse:
"""Get detailed information about a specific food item"""
params = {
"api_key": self.api_key,
"format": format_type
}
if nutrients:
params["nutrients"] = ",".join(map(str, nutrients))
response = await self.client.get(
f"{self.base_url}/food/{fdc_id}",
params=params
)
response.raise_for_status()
return FoodDetailResponse(**response.json())
async def get_multiple_foods(
self,
fdc_ids: List[int],
format_type: str = "full",
nutrients: Optional[List[int]] = None
) -> List[FoodDetailResponse]:
"""Get detailed information about multiple food items"""
params = {
"api_key": self.api_key,
"fdcIds": ",".join(map(str, fdc_ids)),
"format": format_type
}
if nutrients:
params["nutrients"] = ",".join(map(str, nutrients))
response = await self.client.get(
f"{self.base_url}/foods",
params=params
)
response.raise_for_status()
data = response.json()
return [FoodDetailResponse(**item) for item in data]
async def close(self):
"""Close the HTTP client"""
await self.client.aclose()
def get_fdc_api_client() -> FoodDataCentralAPI:
"""Create and return a Food Data Central API client"""
api_key = os.getenv("USDA_API_KEY")
if not api_key:
raise ValueError("USDA_API_KEY environment variable is required")
return FoodDataCentralAPI(api_key)
