refactor: reformat using black

Author: ksarang90

chatbot-api/__init__.py

@@ -1 +1 @@
-"""ChatBot API package""" 
+"""ChatBot API package"""

chatbot-api/conftest.py

@@ -9,15 +9,15 @@ def setup_test_env():
     """Automatically set up test environment variables before each test"""
     # Load test environment variables
     load_dotenv()
-    
+
     # Set up test environment variables
-    os.environ['OPENAI_API_KEY'] = 'test-api-key'
-    os.environ['OPENAI_API_BASE'] = 'https://test-api-base.com'
-    
+    os.environ["OPENAI_API_KEY"] = "test-api-key"
+    os.environ["OPENAI_API_BASE"] = "https://test-api-base.com"
+
     yield
-    
+
     # Clean up after tests
-    if 'OPENAI_API_KEY' in os.environ:
-        del os.environ['OPENAI_API_KEY']
-    if 'OPENAI_API_BASE' in os.environ:
-        del os.environ['OPENAI_API_BASE'] 
+    if "OPENAI_API_KEY" in os.environ:
+        del os.environ["OPENAI_API_KEY"]
+    if "OPENAI_API_BASE" in os.environ:
+        del os.environ["OPENAI_API_BASE"]

chatbot-api/src/chat/models.py

@@ -2,25 +2,29 @@
 from typing import List, Optional, Union, Dict
 from datetime import datetime
 
+
 @dataclass
 class ImageContent:
     type: str = "image_url"
     image_url: Dict[str, str] = None
 
+
 @dataclass
 class TextContent:
     type: str = "text"
     text: str = ""
 
+
 @dataclass
 class Message:
     role: str
     content: Union[str, List[Union[TextContent, ImageContent]]]
     model: Optional[str] = None
     timestamp: datetime = datetime.utcnow()
 
+
 @dataclass
 class ChatResponse:
     content: str
     model: str
-    timestamp: datetime = datetime.utcnow() 
+    timestamp: datetime = datetime.utcnow()

chatbot-api/src/chat/provider.py

@@ -6,67 +6,71 @@
 
 logger = logging.getLogger(__name__)
 
+
 class AIProvider(Protocol):
     """Protocol for AI providers"""
+
     def generate_response(self, messages: List[Message]) -> ChatResponse: ...
     def generate_stream(self, messages: List[Message]) -> AsyncIterator[str]: ...
 
+
 class OpenAIProvider:
     """OpenAI implementation of AIProvider"""
+
     def __init__(self, api_key: str, api_base: str | None = None):
         self.client = OpenAI(api_key=api_key, base_url=api_base)
 
     def _format_message(self, message: Message) -> dict:
         """Format message for OpenAI API"""
         if isinstance(message.content, str):
             return {"role": message.role, "content": message.content}
-        
+
         # For messages with text and images
         formatted_content = []
         for item in message.content:
             if isinstance(item, TextContent):
                 formatted_content.append({"type": "text", "text": item.text})
             elif isinstance(item, ImageContent):
-                formatted_content.append({
-                    "type": "image_url",
-                    "image_url": item.image_url
-                })
+                formatted_content.append(
+                    {"type": "image_url", "image_url": item.image_url}
+                )
         return {"role": message.role, "content": formatted_content}
 
     def generate_response(self, messages: List[Message]) -> ChatResponse:
         """Generate a response for messages"""
         # Use the model from the last message, or default to claude-3-5-sonnet
-        model = next((m.model for m in reversed(messages) if m.model), "claude-3-5-sonnet")
-        
+        model = next(
+            (m.model for m in reversed(messages) if m.model), "claude-3-5-sonnet"
+        )
+
         formatted_messages = [self._format_message(m) for m in messages]
         response = self.client.chat.completions.create(
-            model=model,
-            messages=formatted_messages
+            model=model, messages=formatted_messages
         )
-        
+
         return ChatResponse(
             content=response.choices[0].message.content,
             model=response.model,
-            timestamp=datetime.utcnow()
+            timestamp=datetime.utcnow(),
         )
 
     async def generate_stream(self, messages: List[Message]) -> AsyncIterator[str]:
         """Stream response for messages"""
         try:
             # Use the model from the last message, or default to claude-3-5-sonnet
-            model = next((m.model for m in reversed(messages) if m.model), "claude-3-5-sonnet")
-            
+            model = next(
+                (m.model for m in reversed(messages) if m.model), "claude-3-5-sonnet"
+            )
+
             formatted_messages = [self._format_message(m) for m in messages]
             stream = self.client.chat.completions.create(
-                model=model,
-                messages=formatted_messages,
-                stream=True
+                model=model, messages=formatted_messages, stream=True
             )
-            
+
             for chunk in stream:
                 if chunk.choices[0].delta.content:
                     content = chunk.choices[0].delta.content
                     yield content
         except Exception as e:
             logger.error(f"Error in stream: {str(e)}")
-            raise 
+            raise

chatbot-api/src/chat/routes.py

@@ -1,7 +1,13 @@
 from fastapi import APIRouter, Depends
 from fastapi.responses import StreamingResponse
 from typing import List, Union, Dict
-from .schemas import ChatRequestSchema, ChatResponseSchema, MessageSchema, TextContentSchema, ImageContentSchema
+from .schemas import (
+    ChatRequestSchema,
+    ChatResponseSchema,
+    MessageSchema,
+    TextContentSchema,
+    ImageContentSchema,
+)
 from .service import ChatService
 from .models import Message, TextContent, ImageContent
 from datetime import datetime
@@ -12,30 +18,32 @@
 
 router = APIRouter(prefix="/chat", tags=["chat"])
 
+
 async def get_chat_service() -> ChatService:
     # This will be overridden in main.py
     raise NotImplementedError("Chat service not configured")
 
+
 def extract_message_content(message: MessageSchema) -> tuple[str, List[str]]:
     """Extract text and image URLs from a message"""
     if isinstance(message.content, str):
         return message.content, []
-    
+
     text = ""
     images = []
-    
+
     for content in message.content:
         if content.type == "text":
             text = content.text
         elif content.type == "image_url":
             images.append(content.image_url.url)
-    
+
     return text, images
 
+
 @router.post("/", response_model=ChatResponseSchema)
 async def chat(
-    request: ChatRequestSchema,
-    chat_service: ChatService = Depends(get_chat_service)
+    request: ChatRequestSchema, chat_service: ChatService = Depends(get_chat_service)
 ) -> ChatResponseSchema:
     # Convert request messages to domain models
     messages = []
@@ -44,36 +52,37 @@ async def chat(
             content = msg.content
         else:
             content = [
-                TextContent(text=c.text) if c.type == "text"
-                else ImageContent(image_url={"url": c.image_url.url})
+                (
+                    TextContent(text=c.text)
+                    if c.type == "text"
+                    else ImageContent(image_url={"url": c.image_url.url})
+                )
                 for c in msg.content
             ]
-        messages.append(Message(
-            role=msg.role,
-            content=content,
-            model=msg.model,
-            timestamp=msg.timestamp or datetime.utcnow()
-        ))
-    
+        messages.append(
+            Message(
+                role=msg.role,
+                content=content,
+                model=msg.model,
+                timestamp=msg.timestamp or datetime.utcnow(),
+            )
+        )
+
     # Extract content from the last message
     text, images = extract_message_content(request.messages[-1])
-    
+
     response = await chat_service.process_message(
-        text=text,
-        images=images,
-        conversation_messages=messages
+        text=text, images=images, conversation_messages=messages
     )
-    
+
     return ChatResponseSchema(
-        reply=response.content,
-        model=response.model,
-        timestamp=response.timestamp
+        reply=response.content, model=response.model, timestamp=response.timestamp
     )
 
+
 @router.post("/stream")
 async def stream_chat(
-    request: ChatRequestSchema,
-    chat_service: ChatService = Depends(get_chat_service)
+    request: ChatRequestSchema, chat_service: ChatService = Depends(get_chat_service)
 ):
     # Convert request messages to domain models
     messages = []
@@ -82,26 +91,29 @@ async def stream_chat(
             content = msg.content
         else:
             content = [
-                TextContent(text=c.text) if c.type == "text"
-                else ImageContent(image_url={"url": c.image_url.url})
+                (
+                    TextContent(text=c.text)
+                    if c.type == "text"
+                    else ImageContent(image_url={"url": c.image_url.url})
+                )
                 for c in msg.content
             ]
-        messages.append(Message(
-            role=msg.role,
-            content=content,
-            model=msg.model,
-            timestamp=msg.timestamp or datetime.utcnow()
-        ))
-    
+        messages.append(
+            Message(
+                role=msg.role,
+                content=content,
+                model=msg.model,
+                timestamp=msg.timestamp or datetime.utcnow(),
+            )
+        )
+
     # Extract content from the last message
     text, images = extract_message_content(request.messages[-1])
 
     async def generate():
         try:
             async for chunk in chat_service.stream_response(
-                text=text,
-                images=images,
-                conversation_messages=messages
+                text=text, images=images, conversation_messages=messages
             ):
                 yield f"data: {json.dumps({'content': chunk})}\n\n"
         except Exception as e:
@@ -114,6 +126,6 @@ async def generate():
         headers={
             "Cache-Control": "no-cache",
             "Connection": "keep-alive",
-            "X-Accel-Buffering": "no"
-        }
-    ) 
+            "X-Accel-Buffering": "no",
+        },
+    )

chatbot-api/src/chat/schemas.py

@@ -2,33 +2,39 @@
 from typing import List, Optional, Union, Dict
 from datetime import datetime
 
+
 class ImageUrlSchema(BaseModel):
     url: str
     model_config = ConfigDict(from_attributes=True)
 
+
 class ImageContentSchema(BaseModel):
     type: str = "image_url"
     image_url: ImageUrlSchema
     model_config = ConfigDict(from_attributes=True)
 
+
 class TextContentSchema(BaseModel):
     type: str = "text"
     text: str
     model_config = ConfigDict(from_attributes=True)
 
+
 class MessageSchema(BaseModel):
     role: str
     content: Union[str, List[Union[TextContentSchema, ImageContentSchema]]]
     model: Optional[str] = None
     timestamp: Optional[datetime] = None
     model_config = ConfigDict(from_attributes=True)
 
+
 class ChatRequestSchema(BaseModel):
     messages: List[MessageSchema]
     model_config = ConfigDict(from_attributes=True)
 
+
 class ChatResponseSchema(BaseModel):
     reply: str
     model: str
     timestamp: datetime
-    model_config = ConfigDict(from_attributes=True) 
+    model_config = ConfigDict(from_attributes=True)