Coverage for main.py: 76%

1import asyncio 

2import hashlib 

3import hmac 

4import os 

5import time 

6from typing import Any 

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

8from dotenv import load_dotenv 

9from fastapi.responses import JSONResponse 

10from langchain.chat_models import BaseChatModel, init_chat_model 

11from langchain_core.messages import HumanMessage, SystemMessage 

12from opentelemetry import trace 

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

14from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor 

15from opentelemetry.sdk.resources import Resource, SERVICE_NAME 

16from opentelemetry.sdk.trace import TracerProvider 

17from opentelemetry.sdk.trace.export import BatchSpanProcessor 

18from pydantic import BaseModel 

19 

20from client.cooking_assistant_gen_ai_services_api_internal_client.types import Unset 

21 

22from prometheus_fastapi_instrumentator import Instrumentator 

23from client.cooking_assistant_gen_ai_services_api_internal_client.models.help_request_forwarded import ( 

24 HelpRequestForwarded, 

25) 

26 

27 

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

29class LocalHelpResponse(BaseModel): 

30 response: str 

31 

32 

33# Load variables from .env for local testing 

34load_dotenv() 

35 

36_otlp_base = os.getenv( 

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

38) 

39_provider = TracerProvider(resource=Resource.create({SERVICE_NAME: "py-help-service"})) 

40_provider.add_span_processor( 

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

42) 

43trace.set_tracer_provider(_provider) 

44_tracer = trace.get_tracer("py-help-service") 

45 

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

47Instrumentator().instrument(app).expose(app) 

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

49 

50 

51@app.exception_handler(HTTPException) 

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

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

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

55 

56 

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

58 

59SECRET_KEY_STR = os.getenv("INTERNAL_AUTH_SECRET") 

60if not SECRET_KEY_STR: 

61 raise RuntimeError( 

62 "CRITICAL: INTERNAL_AUTH_SECRET environment variable is missing!" 

63 ) 

64 

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

66 

67 

68async def verify_internal_hmac( 

69 request: Request, 

70 x_internal_timestamp: str = Header(None), 

71 x_internal_signature: str = Header(None), 

72): 

73 """ 

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

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

76 """ 

77 if not x_internal_timestamp or not x_internal_signature: 

78 raise HTTPException( 

79 status_code=401, 

80 detail={ 

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

82 }, 

83 ) 

84 

85 # 1. Parse incoming timestamp string context securely 

86 try: 

87 request_time = int(x_internal_timestamp) 

88 except ValueError: 

89 raise HTTPException( 

90 status_code=400, 

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

92 ) 

93 

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

95 current_time = int(time.time()) 

96 if abs(current_time - request_time) > 300: 

97 raise HTTPException( 

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

99 ) 

100 

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

102 body_bytes = await request.body() 

103 

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

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

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

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

108 hmac_context.update(b".") 

109 hmac_context.update(body_bytes) 

110 

111 expected_signature = hmac_context.hexdigest() 

112 

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

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

115 raise HTTPException( 

116 status_code=403, 

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

118 ) 

119 

120 

121def get_llm(): 

122 """ 

123 Dynamically builds the correct LLM structure using environment variables. 

124 """ 

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

126 

127 kwargs = {"timeout": 60} 

128 

129 if provider == "local": 

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

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

132 ) 

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

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

135 

136 # We explicitly enforce the underlying OpenAI integration layout 

137 provider_target = "openai" 

138 

139 elif provider == "openai": 

140 logos_key = os.getenv("LOGOS_KEY") 

141 if not logos_key: 

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

143 

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

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

146 ) 

147 kwargs["api_key"] = logos_key 

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

149 

150 provider_target = "openai" 

151 

152 else: 

153 gemini_key = os.getenv("GEMINI_HELP_SERVICE_KEY") 

154 if not gemini_key: 

155 raise RuntimeError("CRITICAL: GEMINI_HELP_SERVICE_KEY is missing!") 

156 

157 kwargs["google_api_key"] = gemini_key 

158 kwargs["thinking_level"] = ( 

159 "low" # inference time fluctuate a lot 

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

161 # medium and high seem noticeably slower than low 

162 ) 

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

164 

165 provider_target = "google_genai" 

166 

167 try: 

168 return init_chat_model( 

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

170 ) 

171 except Exception as e: 

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

173 

174 

175@app.get("/health") 

176def health_check(): 

177 return {"status": "healthy"} 

178 

179 

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

181async def generate_help( 

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

183): 

184 try: 

185 request = HelpRequestForwarded.from_dict(request_data) 

186 except Exception as e: 

187 raise HTTPException( 

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

189 ) 

190 

191 try: 

192 context = ["You are a professional culinary assistant."] 

193 

194 language = "English" 

195 if request.profile and request.profile.preferences: 

196 prefs = request.profile.preferences 

197 

198 language = LANGUAGE_NAMES.get( 

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

200 ) 

201 context.append(f"You serve a user speaking {language}.") 

202 

203 diet = prefs.diet or [] 

204 if diet: 

205 context.append( 

206 f"The user specifies the following diet: {', '.join(diet)}." 

207 ) 

208 

209 about_me = prefs.about_me or [] 

210 if about_me: 

211 context.append(f"The user specifies: {', '.join(about_me)}") 

212 

213 allergies = prefs.allergies or [] 

214 if allergies: 

215 context.append( 

216 f"Important: The user is allergic to: {', '.join(allergies)}." 

217 ) 

218 

219 if request.recipe: 

220 recipe_ctx = [ 

221 f"The user is currently looking at a recipe for '{request.recipe.title}'." 

222 ] 

223 

224 ingredients = getattr(request.recipe, "ingredients", None) 

225 if ingredients: 

226 recipe_ctx.append("\nIngredients:") 

227 for ing in ingredients: 

228 # an absent quantity or unit is Unset, which would render as "UNSET" 

229 parts = [ing.quantity, ing.unit, ing.name] 

230 line = " ".join(str(p) for p in parts if not isinstance(p, Unset)) 

231 recipe_ctx.append(f"- {line}") 

232 

233 instructions = getattr(request.recipe, "instructions", None) 

234 if instructions: 

235 recipe_ctx.append("\nInstructions:") 

236 for idx, step in enumerate(instructions, 1): 

237 recipe_ctx.append(f"{idx}. {step}") 

238 

239 if request.recipe.nutrients and not isinstance( 

240 request.recipe.nutrients, Unset 

241 ): 

242 nut = request.recipe.nutrients 

243 recipe_ctx.append("\nNutritional Information (Total recipe):") 

244 recipe_ctx.append(f"- Calories: {nut.calories} kcal") 

245 recipe_ctx.append(f"- Protein: {nut.protein}g") 

246 recipe_ctx.append(f"- Fat: {nut.fat}g") 

247 recipe_ctx.append(f"- Carbohydrates: {nut.carbs}g") 

248 

249 context.append("\n".join(recipe_ctx)) 

250 

251 # Combine into LLM prompt 

252 system_prompt = SystemMessage(content=" ".join(context)) 

253 user_prompt = HumanMessage(content=request.prompt) 

254 

255 structured_llm = llm.with_structured_output(LocalHelpResponse) 

256 

257 with _tracer.start_as_current_span("llm.generate_help"): 

258 result: LocalHelpResponse = await asyncio.wait_for( 

259 structured_llm.ainvoke([system_prompt, user_prompt]), timeout=60 

260 ) 

261 

262 return result.model_dump() 

263 

264 except asyncio.TimeoutError: 

265 raise HTTPException( 

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

267 ) 

268 except Exception as e: 

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