Coverage for main.py: 72%

1import asyncio 

2import hashlib 

3import hmac 

4import os 

5import time 

6import traceback 

7from typing import Any, List 

8from fastapi import Depends, FastAPI, HTTPException, Header, Request 

9from dotenv import load_dotenv 

10from fastapi.responses import JSONResponse 

11from langchain.chat_models import init_chat_model 

12from langchain_core.messages import HumanMessage, SystemMessage 

13from opentelemetry import trace 

14from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter 

15from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor 

16from opentelemetry.sdk.resources import Resource, SERVICE_NAME 

17from opentelemetry.sdk.trace import TracerProvider 

18from opentelemetry.sdk.trace.export import BatchSpanProcessor 

19from pydantic import BaseModel 

20from langchain_core.language_models.chat_models import BaseChatModel 

21 

22from prometheus_fastapi_instrumentator import Instrumentator 

23from client.cooking_assistant_gen_ai_services_api_internal_client.models.recipe_request_forwarded import ( 

24 RecipeRequestForwarded, 

25) 

26 

27from client.cooking_assistant_gen_ai_services_api_internal_client.models.recipe_input import ( 

28 RecipeInput, 

29) 

30 

31from client.cooking_assistant_gen_ai_services_api_internal_client.models.nutrient_request_forwarded import ( 

32 NutrientRequestForwarded, 

33) 

34 

35# Load variables from .env for local testing 

36load_dotenv() 

37 

38_otlp_base = os.getenv( 

39 "OTEL_EXPORTER_OTLP_ENDPOINT", "http://alloy.monitoring.svc.cluster.local:4318" 

40) 

41_provider = TracerProvider( 

42 resource=Resource.create({SERVICE_NAME: "py-recipe-service"}) 

43) 

44_provider.add_span_processor( 

45 BatchSpanProcessor(OTLPSpanExporter(endpoint=f"{_otlp_base}/v1/traces")) 

46) 

47trace.set_tracer_provider(_provider) 

48_tracer = trace.get_tracer("py-recipe-service") 

49 

50app = FastAPI(title="Cooking Assistant GenAI Service") 

51Instrumentator().instrument(app).expose(app) 

52FastAPIInstrumentor.instrument_app(app, excluded_urls="/health,/metrics") 

53 

54 

55# autogenerated classes must be mapped to local ones that are compatible with pydantic 

56class LocalRecipeIngredient(BaseModel): 

57 # Optional unlike nutrients below: omitting them is sometimes the *correct* answer 

58 # ("2 eggs" has no unit, "salt" has neither), so the llm is guided by prompt instead. 

59 quantity: float | None = None 

60 unit: str | None = None 

61 name: str 

62 

63 

64class LocalRecipeNutrients(BaseModel): 

65 calories: int 

66 protein: int 

67 fat: int 

68 carbs: int 

69 

70 

71class LocalRecipeInput(BaseModel): 

72 title: str 

73 ingredients: List[LocalRecipeIngredient] 

74 instructions: List[str] 

75 portions: float 

76 nutrients: LocalRecipeNutrients # required field here as llm can't handle optional fields reliably 

77 

78 

79# local wrapper for recipes - llms are optimized for producing json whereas api spec expects array of json 

80class RecipeListWrapper(BaseModel): 

81 recipes: List[LocalRecipeInput] 

82 

83 

84@app.exception_handler(HTTPException) 

85async def http_exception_handler(request: Request, exc: HTTPException): 

86 body = exc.detail if isinstance(exc.detail, dict) else {"message": exc.detail} 

87 return JSONResponse(status_code=exc.status_code, content=body, headers=exc.headers) 

88 

89 

90LANGUAGE_NAMES = {"EN": "English", "DE": "German", "HU": "Hungarian"} 

91 

92SECRET_KEY_STR = os.getenv("INTERNAL_AUTH_SECRET") 

93if not SECRET_KEY_STR: 

94 raise RuntimeError( 

95 "CRITICAL: INTERNAL_AUTH_SECRET environment variable is missing!" 

96 ) 

97 

98SECRET_KEY_BYTES = SECRET_KEY_STR.encode("utf-8") 

99 

100 

101async def verify_internal_hmac( 

102 request: Request, 

103 x_internal_timestamp: str = Header(None), 

104 x_internal_signature: str = Header(None), 

105): 

106 """ 

107 Validates that the incoming request contains an authentic HMAC signature bound to the timestamp and the request payload. 

108 Caution: Redundant versions of this method in py-help-service and py-recipe service. 

109 """ 

110 if not x_internal_timestamp or not x_internal_signature: 

111 raise HTTPException( 

112 status_code=401, 

113 detail={ 

114 "message": "Unauthorized: Missing security authentication headers." 

115 }, 

116 ) 

117 

118 # 1. Parse incoming timestamp string context securely 

119 try: 

120 request_time = int(x_internal_timestamp) 

121 except ValueError: 

122 raise HTTPException( 

123 status_code=400, 

124 detail={"message": "Invalid timestamp metadata formatting."}, 

125 ) 

126 

127 # 2. Reject requests with more than 5 minutes of clock drift 

128 current_time = int(time.time()) 

129 if abs(current_time - request_time) > 300: 

130 raise HTTPException( 

131 status_code=401, detail={"message": "Request token signature expired."} 

132 ) 

133 

134 # 3. Read the raw body bytes directly from the stream 

135 body_bytes = await request.body() 

136 

137 # 4. Recalculate signature locally by hashing both pieces together 

138 # Using a separator byte like b'.' prevents boundary shifting bugs 

139 hmac_context = hmac.new(SECRET_KEY_BYTES, digestmod=hashlib.sha256) 

140 hmac_context.update(x_internal_timestamp.encode("utf-8")) 

141 hmac_context.update(b".") 

142 hmac_context.update(body_bytes) 

143 

144 expected_signature = hmac_context.hexdigest() 

145 

146 # 5. Use constant-time comparison to completely prevent timing attacks 

147 if not hmac.compare_digest(expected_signature, x_internal_signature): 

148 raise HTTPException( 

149 status_code=403, 

150 detail={"message": "Forbidden: HMAC signature validation mismatch."}, 

151 ) 

152 

153 

154def get_llm(): 

155 """ 

156 Dynamically builds the correct LLM structure using environment variables. 

157 """ 

158 provider = os.getenv("PROVIDER", "google_genai") 

159 

160 kwargs = {"timeout": 60} 

161 

162 if provider == "local": 

163 kwargs["base_url"] = os.getenv( 

164 "LOCAL_BASE_URL", "http://host.docker.internal:1234/v1" 

165 ) 

166 kwargs["api_key"] = os.getenv("LOCAL_KEY", "not-needed") 

167 kwargs["response_format"] = {"type": "json_object"} 

168 model_name = os.getenv("LOCAL_MODEL", "local-model") 

169 

170 # We explicitly enforce the underlying OpenAI integration layout 

171 provider_target = "openai" 

172 

173 elif provider == "openai": 

174 logos_key = os.getenv("LOGOS_KEY") 

175 if not logos_key: 

176 raise RuntimeError("CRITICAL: LOGOS_KEY is missing!") 

177 

178 kwargs["base_url"] = os.getenv( 

179 "LOGOS_BASE_URL", "https://logos.aet.cit.tum.de/v1" 

180 ) 

181 kwargs["api_key"] = logos_key 

182 kwargs["reasoning_effort"] = ( 

183 "low" # times out for medium and high, "minimal not supported by Harmony" 

184 ) 

185 model_name = os.getenv("LOGOS_MODEL", "openai/gpt-oss-120b") 

186 

187 provider_target = "openai" 

188 

189 else: 

190 gemini_key = os.getenv("GEMINI_RECIPE_SERVICE_KEY") 

191 if not gemini_key: 

192 raise RuntimeError("CRITICAL: GEMINI_RECIPE_SERVICE_KEY is missing!") 

193 

194 kwargs["google_api_key"] = gemini_key 

195 kwargs["thinking_level"] = ( 

196 "low" # inference time fluctuate a lot 

197 # minimal doesn't seem a lot faster than low but seemingly produces more mistakes in json output formatting 

198 # medium and high seem noticeably slower than low 

199 ) 

200 model_name = os.getenv("GEMINI_MODEL", "gemini-3.1-flash-lite") 

201 

202 provider_target = "google_genai" 

203 

204 try: 

205 return init_chat_model( 

206 model=model_name, model_provider=provider_target, **kwargs 

207 ) 

208 except Exception as e: 

209 raise RuntimeError(f"Failed to boot LLM provider '{provider}': {e}") 

210 

211 

212@app.get("/health") 

213def health_check(): 

214 return {"status": "healthy"} 

215 

216 

217@app.post("/ai/recipes", dependencies=[Depends(verify_internal_hmac)]) 

218async def generate_recipes( 

219 request_data: dict[str, Any], llm: BaseChatModel = Depends(get_llm) 

220): 

221 

222 try: 

223 request = RecipeRequestForwarded.from_dict(request_data) 

224 except Exception as e: 

225 raise HTTPException( 

226 status_code=400, detail={"message": f"Invalid request format: {str(e)}"} 

227 ) 

228 

229 if not request.profile or not request.profile.preferences: 

230 raise HTTPException( 

231 status_code=400, detail={"message": "Missing required profile preferences."} 

232 ) 

233 

234 try: 

235 # 1. Extract Profile Context 

236 prefs = request.profile.preferences 

237 diet = ", ".join(prefs.diet) if prefs.diet else "None" 

238 allergies = ", ".join(prefs.allergies) if prefs.allergies else "None" 

239 about = ", ".join(prefs.about_me) if prefs.about_me else "None" 

240 language = LANGUAGE_NAMES.get( 

241 getattr(prefs.language, "value", None) or "EN", "English" 

242 ) 

243 

244 # 2. Build the System Prompt 

245 system_prompt = ( 

246 "You are a professional chef. Create a collection of distinct high-quality recipes.\n" 

247 f"Constraint - Diet: {diet}\n" 

248 f"Constraint - Allergies: {allergies} (DO NOT USE THESE)\n" 

249 f"User Context: {about}\n\n" 

250 f"Write all recipe content (title, ingredients, units and instructions) in {language}. " 

251 f"Abbreviate units the way {language} conventionally abbreviates them, not the way English " 

252 "does (a tablespoon is 'tbsp' in English but 'EL' in German).\n" 

253 "Omit the unit for ingredients counted as whole items (2 eggs, 1 onion) rather than " 

254 "inventing one (2 pieces egg). Always give a quantity when you give a unit.\n" 

255 "Output the nutrients for the whole recipe in total, not per portion.\n" 

256 f"Keep the JSON keys in English as specified." 

257 ) 

258 

259 structured_llm = llm.with_structured_output(RecipeListWrapper) 

260 

261 # 3. Invoke LLM 

262 messages = [ 

263 SystemMessage(content=system_prompt), 

264 HumanMessage(content=f"Generate 3 distinct recipes for: {request.prompt}"), 

265 ] 

266 

267 with _tracer.start_as_current_span("llm.generate_recipes"): 

268 response = await asyncio.wait_for( 

269 structured_llm.ainvoke(messages), timeout=60 

270 ) 

271 

272 # return array of recipes as expeted from the api spec 

273 final_recipes = [] 

274 for r in response.recipes: 

275 # exclude_none: the spec marks an absent quantity/unit by omitting the key, 

276 # and forbids additional properties — a null would be rejected as neither. 

277 recipe_dict = r.model_dump(exclude_none=True) 

278 final_recipes.append(RecipeInput.from_dict(recipe_dict)) 

279 

280 return [r.to_dict() for r in final_recipes] 

281 

282 except asyncio.TimeoutError: 

283 raise HTTPException( 

284 status_code=504, detail={"message": "LLM connection timed out."} 

285 ) 

286 except Exception as e: 

287 traceback.print_exc() 

288 raise HTTPException(status_code=500, detail={"message": str(e)}) 

289 

290 

291@app.post("/ai/nutrients", dependencies=[Depends(verify_internal_hmac)]) 

292async def generate_nutrients( 

293 request_data: dict[str, Any], llm: BaseChatModel = Depends(get_llm) 

294): 

295 try: 

296 request = NutrientRequestForwarded.from_dict(request_data) 

297 except Exception as e: 

298 raise HTTPException( 

299 status_code=400, detail={"message": f"Invalid request format: {str(e)}"} 

300 ) 

301 

302 if not request.recipe: 

303 raise HTTPException( 

304 status_code=400, detail={"message": "Missing required recipe data."} 

305 ) 

306 

307 try: 

308 recipe_dict = request.recipe.to_dict() 

309 

310 system_prompt = ( 

311 "You are an expert nutritional scientist. Calculate the macronutrients and total calories " 

312 "for the provided recipe. Output the nutrients for the whole recipe in total, not per portion. " 

313 "Evaluate ingredient quantities, units, and base portion sizes carefully. " 

314 "An ingredient with a quantity but no unit is counted as whole items ('2 eggs') - assume a " 

315 "typical size for one item. An ingredient with neither is added to taste ('salt') - assume a " 

316 "negligible amount. " 

317 "Ensure the output strictly mirrors the exact target JSON object fields." 

318 ) 

319 

320 structured_llm = llm.with_structured_output(LocalRecipeNutrients) 

321 

322 messages = [ 

323 SystemMessage(content=system_prompt), 

324 HumanMessage( 

325 content=f"Calculate exact macronutrients for this recipe payload: {recipe_dict}" 

326 ), 

327 ] 

328 

329 with _tracer.start_as_current_span("llm.generate_nutrients"): 

330 response = await asyncio.wait_for( 

331 structured_llm.ainvoke(messages), timeout=60 

332 ) 

333 

334 return response.model_dump() 

335 

336 except asyncio.TimeoutError: 

337 raise HTTPException( 

338 status_code=504, detail={"message": "LLM connection timed out."} 

339 ) 

340 except Exception as e: 

341 traceback.print_exc() 

342 raise HTTPException(status_code=500, detail={"message": str(e)})