Capstone Project - Full Deployment

1# app/main.py
2from fastapi import FastAPI
3from fastapi.middleware.cors import CORSMiddleware
4from contextlib import asynccontextmanager
5from app.middleware import RateLimitMiddleware, MetricsMiddleware
6from app.routes import chat, health
7from app.config import settings
8
9@asynccontextmanager
10async def lifespan(app: FastAPI):
11    # Startup
12    print("Starting AI service...")
13    yield
14    # Shutdown
15    print("Shutting down...")
16
17app = FastAPI(
18    title="AI Deployment Capstone",
19    version="1.0.0",
20    lifespan=lifespan
21)
22
23app.add_middleware(
24    CORSMiddleware,
25    allow_origins=settings.CORS_ORIGINS,
26    allow_methods=["*"],
27    allow_headers=["*"],
28)
29app.add_middleware(RateLimitMiddleware, max_requests=60)
30app.add_middleware(MetricsMiddleware)
31
32app.include_router(health.router, tags=["health"])
33app.include_router(chat.router, prefix="/api/v1", tags=["chat"])

1# app/config.py
2from pydantic_settings import BaseSettings
3
4class Settings(BaseSettings):
5    OPENAI_API_KEY: str
6    REDIS_URL: str = "redis://redis:6379"
7    LANGSMITH_API_KEY: str = ""
8    API_KEYS: list[str] = []
9    CORS_ORIGINS: list[str] = ["http://localhost:3000"]
10    RATE_LIMIT: int = 60
11    DAILY_BUDGET: float = 10.0
12    
13    class Config:
14        env_file = ".env"
15
16settings = Settings()

1# app/services/ai_service.py
2from langchain_openai import ChatOpenAI
3from langchain_core.prompts import ChatPromptTemplate
4from langchain_core.output_parsers import StrOutputParser
5from app.services.cache_service import CacheService
6
7class AIService:
8    def __init__(self):
9        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
10        self.cache = CacheService()
11        
12        self.chat_prompt = ChatPromptTemplate.from_messages([
13            ("system", "You are a helpful AI assistant."),
14            ("human", "{message}")
15        ])
16        
17        self.chain = self.chat_prompt | self.llm | StrOutputParser()
18    
19    async def chat(self, message: str) -> str:
20        # Check cache
21        cached = await self.cache.get(message)
22        if cached:
23            return cached
24        
25        # Invoke chain
26        result = await self.chain.ainvoke({"message": message})
27        
28        # Cache result
29        await self.cache.set(message, result)
30        
31        return result
32    
33    async def stream_chat(self, message: str):
34        async for chunk in self.chain.astream({"message": message}):
35            yield chunk

1# tests/test_chat.py
2import pytest
3from httpx import AsyncClient, ASGITransport
4from app.main import app
5
6@pytest.mark.asyncio
7async def test_chat_endpoint():
8    transport = ASGITransport(app=app)
9    async with AsyncClient(transport=transport, base_url="http://test") as ac:
10        response = await ac.post(
11            "/api/v1/chat",
12            json={"message": "Hello"},
13            headers={"X-API-Key": "test-key"}
14        )
15        assert response.status_code == 200
16        assert "response" in response.json()
17
18@pytest.mark.asyncio
19async def test_health_check():
20    transport = ASGITransport(app=app)
21    async with AsyncClient(transport=transport, base_url="http://test") as ac:
22        response = await ac.get("/health")
23        assert response.status_code == 200
24        assert response.json()["status"] == "healthy"