format

Author: taylorlynn

app/api/endpoints/analysis_endpoints.py

@@ -113,7 +113,7 @@ async def stream_claim_analysis_exp(
         claim = await claim_service.get_claim(claim_id=claim_id, user_id=current_user.id)
 
         session = claim_service._claim_repo._session
-        
+
         await session.rollback()
 
         async def event_generator():
@@ -128,59 +128,57 @@ async def event_generator():
                 # THE HEALTH CHECK LOOP
                 # ---------------------------------------------------------
                 next_event_task = None
-                
+
                 while True:
                     # Only create a new task if we don't already have one waiting
                     if next_event_task is None:
                         next_event_task = asyncio.create_task(anext(orchestrator_stream))
 
                     # Wait for the task to finish, but only wait 15 seconds
                     done, pending = await asyncio.wait(
-                        [next_event_task], 
-                        timeout=15.0, 
-                        return_when=asyncio.FIRST_COMPLETED
+                        [next_event_task], timeout=15.0, return_when=asyncio.FIRST_COMPLETED
                     )
-                    
+
                     if next_event_task in done:
                         # The LLM yielded a chunk! Let's process it.
                         try:
                             event = next_event_task.result()
                             if isinstance(event, dict):
                                 yield f"data: {json.dumps(event)}\n\n"
-                                
+
                             # Reset the task so we grab the next chunk on the next loop
-                            next_event_task = None 
-                            
+                            next_event_task = None
+
                         except StopAsyncIteration:
                             # The stream finished normally!
                             break
                         except Exception as e:
                             # If the orchestrator crashed, catch it here
                             raise e
-                            
+
                     else:
                         # The task is in 'pending'. 15 seconds passed, but the LLM is still thinking.
-                        # We yield a heartbeat, but we DO NOT reset next_event_task. 
+                        # We yield a heartbeat, but we DO NOT reset next_event_task.
                         # It will keep running safely in the background on the next loop!
                         logger.debug("Stream idle for 15s. Sending health check ping...")
                         yield ": healthcheck\n\n"
-                        
+
                 yield "data: [DONE]\n\n"
 
             except asyncio.CancelledError:
                 logger.warning(f"Client disconnected during stream for claim {claim_id}")
                 raise
-                
+
             # async for event in analysis_orchestrator.analyze_claim_stream(
-                #     claim=claim, user_id=current_user.id, default=False
-                # ):
-                #     if isinstance(event, dict):
-                #         yield f"data: {json.dumps(event)}\n\n"
+            #     claim=claim, user_id=current_user.id, default=False
+            # ):
+            #     if isinstance(event, dict):
+            #         yield f"data: {json.dumps(event)}\n\n"
 
-                # yield "data: [DONE]\n\n"
+            # yield "data: [DONE]\n\n"
 
             # except asyncio.CancelledError:
-            #     # THE FIX: The user closed their browser! 
+            #     # THE FIX: The user closed their browser!
             #     logger.info(f"Client disconnected during stream for claim {claim_id}")
             #     await session.rollback()  # Explicitly release the lock!
             #     raise
@@ -196,7 +194,7 @@ async def event_generator():
                 #         logger.error(f"Force rollback failed: {e}")
                 #     finally:
                 #         await session.close()
-                
+
                 # # Fire and forget. FastAPI cannot cancel this!
                 # asyncio.create_task(force_cleanup())
                 if next_event_task and not next_event_task.done():

app/repositories/implementations/source_repository.py

@@ -44,17 +44,21 @@ async def create_with_domain(self, source: SourceModel) -> Optional[SourceModel]
         try:
             self._session.add(source)
             await self._session.commit()
-            
+
             # --- THE FIX ---
             # Re-fetch to guarantee it is fully loaded and not expired by the commit
-            stmt = select(self._model_class).where(self._model_class.id == source.id).options(selectinload(self._model_class.domain))
+            stmt = (
+                select(self._model_class)
+                .where(self._model_class.id == source.id)
+                .options(selectinload(self._model_class.domain))
+            )
             result = await self._session.execute(stmt)
             loaded_source = result.scalar_one()
-            
+
             self._session.expunge(loaded_source)
             if loaded_source.domain:
                 self._session.expunge(loaded_source.domain)
-                
+
             return loaded_source
         except Exception as e:
             await self._session.rollback()
@@ -65,17 +69,21 @@ async def update(self, source: SourceModel) -> SourceModel:
         try:
             merged = await self._session.merge(source)
             await self._session.commit()
-            
+
             # --- THE FIX ---
             # Same treatment: refetch eagerly, then detach
-            stmt = select(self._model_class).where(self._model_class.id == merged.id).options(selectinload(self._model_class.domain))
+            stmt = (
+                select(self._model_class)
+                .where(self._model_class.id == merged.id)
+                .options(selectinload(self._model_class.domain))
+            )
             result = await self._session.execute(stmt)
             loaded_source = result.scalar_one()
-            
+
             self._session.expunge(loaded_source)
             if loaded_source.domain:
                 self._session.expunge(loaded_source.domain)
-                
+
             return loaded_source
         except Exception as e:
             await self._session.rollback()

app/services/implementations/serper_web_search_service.py

@@ -64,7 +64,7 @@ async def search_and_create_sources(
                             logger.error(f"Error processing search result: {str(e)}", exc_info=True)
                             continue
             await self.source_repository._session.commit()
-            
+
             return sources
 
         except Exception as e: