Linting Errors

Author: areeb1501

examples/agent_wait_until_ready.py

@@ -24,32 +24,32 @@
 if agent_id:
     print(f"Agent created with ID: {agent_id}")
     print("Waiting for agent to be ready...")
-    
+
     try:
         # Wait for the agent to be deployed and ready
         # This will poll the agent status every 5 seconds (default)
         # and wait up to 5 minutes (default timeout=300 seconds)
         ready_agent = client.agents.wait_until_ready(
             agent_id,
             poll_interval=5.0,  # Check every 5 seconds
-            timeout=300.0,      # Wait up to 5 minutes
+            timeout=300.0,  # Wait up to 5 minutes
         )
-        
+
         if ready_agent.agent and ready_agent.agent.deployment:
             print(f"Agent is ready! Status: {ready_agent.agent.deployment.status}")
             print(f"Agent URL: {ready_agent.agent.url}")
-        
+
         # Now you can use the agent
         # ...
-        
+
     except AgentDeploymentError as e:
         print(f"Agent deployment failed: {e}")
         print(f"Failed status: {e.status}")
-        
+
     except AgentDeploymentTimeoutError as e:
         print(f"Agent deployment timed out: {e}")
         print(f"Agent ID: {e.agent_id}")
-        
+
     except Exception as e:
         print(f"Unexpected error: {e}")
 
@@ -60,37 +60,37 @@
 
 async def main() -> None:
     async_client = AsyncGradient()
-    
+
     # Create a new agent
     agent_response = await async_client.agents.create(
         name="My Async Agent",
         instruction="You are a helpful assistant",
         model_uuid="<your-model-uuid>",
         region="nyc1",
     )
-    
+
     agent_id = agent_response.agent.uuid if agent_response.agent else None
-    
+
     if agent_id:
         print(f"Agent created with ID: {agent_id}")
         print("Waiting for agent to be ready...")
-        
+
         try:
             # Wait for the agent to be deployed and ready (async)
             ready_agent = await async_client.agents.wait_until_ready(
                 agent_id,
                 poll_interval=5.0,
                 timeout=300.0,
             )
-            
+
             if ready_agent.agent and ready_agent.agent.deployment:
                 print(f"Agent is ready! Status: {ready_agent.agent.deployment.status}")
                 print(f"Agent URL: {ready_agent.agent.url}")
-            
+
         except AgentDeploymentError as e:
             print(f"Agent deployment failed: {e}")
             print(f"Failed status: {e.status}")
-            
+
         except AgentDeploymentTimeoutError as e:
             print(f"Agent deployment timed out: {e}")
             print(f"Agent ID: {e.agent_id}")

examples/knowledge_base_indexing_wait.py

@@ -19,37 +19,35 @@ def main():
     # Example 1: Basic usage - wait for indexing job to complete
     print("Example 1: Basic usage")
     print("-" * 50)
-    
+
     # Create an indexing job (replace with your actual knowledge base UUID)
     knowledge_base_uuid = os.getenv("KNOWLEDGE_BASE_UUID", "your-kb-uuid-here")
-    
+
     print(f"Creating indexing job for knowledge base: {knowledge_base_uuid}")
     indexing_job = client.knowledge_bases.indexing_jobs.create(
         knowledge_base_uuid=knowledge_base_uuid,
     )
-    
+
     job_uuid = indexing_job.job.uuid if indexing_job.job else None
     if not job_uuid:
         print("Error: Could not create indexing job")
         return
-    
+
     print(f"Indexing job created with UUID: {job_uuid}")
     print("Waiting for indexing job to complete...")
-    
+
     try:
         # Wait for the job to complete (polls every 5 seconds by default)
-        completed_job = client.knowledge_bases.indexing_jobs.wait_for_completion(
-            job_uuid
-        )
-        
+        completed_job = client.knowledge_bases.indexing_jobs.wait_for_completion(job_uuid)
+
         print("\n✅ Indexing job completed successfully!")
         if completed_job.job:
             print(f"Phase: {completed_job.job.phase}")
             print(f"Total items indexed: {completed_job.job.total_items_indexed}")
             print(f"Total items failed: {completed_job.job.total_items_failed}")
             print(f"Total datasources: {completed_job.job.total_datasources}")
             print(f"Completed datasources: {completed_job.job.completed_datasources}")
-        
+
     except IndexingJobTimeoutError as e:
         print(f"\n⏱️ Timeout: {e}")
     except IndexingJobError as e:
@@ -62,42 +60,44 @@ def example_with_custom_polling():
     """Example with custom polling interval and timeout"""
     print("\n\nExample 2: Custom polling interval and timeout")
     print("-" * 50)
-    
+
     client = Gradient()
     knowledge_base_uuid = os.getenv("KNOWLEDGE_BASE_UUID", "your-kb-uuid-here")
-    
+
     print(f"Creating indexing job for knowledge base: {knowledge_base_uuid}")
     indexing_job = client.knowledge_bases.indexing_jobs.create(
         knowledge_base_uuid=knowledge_base_uuid,
     )
-    
+
     job_uuid = indexing_job.job.uuid if indexing_job.job else None
     if not job_uuid:
         print("Error: Could not create indexing job")
         return
-    
+
     print(f"Indexing job created with UUID: {job_uuid}")
     print("Waiting for indexing job to complete (polling every 10 seconds, 5 minute timeout)...")
-    
+
     try:
         # Wait with custom poll interval (10 seconds) and timeout (5 minutes = 300 seconds)
         completed_job = client.knowledge_bases.indexing_jobs.wait_for_completion(
             job_uuid,
             poll_interval=10,  # Poll every 10 seconds
-            timeout=300,       # Timeout after 5 minutes
+            timeout=300,  # Timeout after 5 minutes
         )
-        
+
         print("\n✅ Indexing job completed successfully!")
         if completed_job.job:
             print(f"Phase: {completed_job.job.phase}")
-        
+
     except IndexingJobTimeoutError:
         print("\n⏱️ Job did not complete within 5 minutes")
         # You can still check the current status
         current_status = client.knowledge_bases.indexing_jobs.retrieve(job_uuid)
         if current_status.job:
             print(f"Current phase: {current_status.job.phase}")
-            print(f"Completed datasources: {current_status.job.completed_datasources}/{current_status.job.total_datasources}")
+            print(
+                f"Completed datasources: {current_status.job.completed_datasources}/{current_status.job.total_datasources}"
+            )
     except IndexingJobError as e:
         print(f"\n❌ Job failed: {e}")
 
@@ -106,31 +106,31 @@ def example_manual_polling():
     """Example of the old manual polling approach (for comparison)"""
     print("\n\nExample 3: Manual polling (old approach)")
     print("-" * 50)
-    
+
     client = Gradient()
     knowledge_base_uuid = os.getenv("KNOWLEDGE_BASE_UUID", "your-kb-uuid-here")
-    
+
     indexing_job = client.knowledge_bases.indexing_jobs.create(
         knowledge_base_uuid=knowledge_base_uuid,
     )
-    
+
     job_uuid = indexing_job.job.uuid if indexing_job.job else None
     if not job_uuid:
         print("Error: Could not create indexing job")
         return
-    
+
     print(f"Indexing job created with UUID: {job_uuid}")
     print("Manual polling (old approach)...")
-    
+
     import time
-    
+
     while True:
         indexing_job = client.knowledge_bases.indexing_jobs.retrieve(job_uuid)
-        
+
         if indexing_job.job and indexing_job.job.phase:
             phase = indexing_job.job.phase
             print(f"Current phase: {phase}")
-            
+
             if phase in ["BATCH_JOB_PHASE_UNKNOWN", "BATCH_JOB_PHASE_PENDING", "BATCH_JOB_PHASE_RUNNING"]:
                 time.sleep(5)
                 continue
@@ -146,36 +146,36 @@ async def example_async():
     """Example using async/await"""
     print("\n\nExample 4: Async usage")
     print("-" * 50)
-    
+
     from gradient import AsyncGradient
-    
+
     client = AsyncGradient()
     knowledge_base_uuid = os.getenv("KNOWLEDGE_BASE_UUID", "your-kb-uuid-here")
-    
+
     print(f"Creating indexing job for knowledge base: {knowledge_base_uuid}")
     indexing_job = await client.knowledge_bases.indexing_jobs.create(
         knowledge_base_uuid=knowledge_base_uuid,
     )
-    
+
     job_uuid = indexing_job.job.uuid if indexing_job.job else None
     if not job_uuid:
         print("Error: Could not create indexing job")
         return
-    
+
     print(f"Indexing job created with UUID: {job_uuid}")
     print("Waiting for indexing job to complete (async)...")
-    
+
     try:
         completed_job = await client.knowledge_bases.indexing_jobs.wait_for_completion(
             job_uuid,
             poll_interval=5,
             timeout=600,  # 10 minute timeout
         )
-        
+
         print("\n✅ Indexing job completed successfully!")
         if completed_job.job:
             print(f"Phase: {completed_job.job.phase}")
-        
+
     except IndexingJobTimeoutError as e:
         print(f"\n⏱️ Timeout: {e}")
     except IndexingJobError as e:
@@ -187,11 +187,11 @@ async def example_async():
 if __name__ == "__main__":
     # Run the basic example
     main()
-    
+
     # Uncomment to run other examples:
     # example_with_custom_polling()
     # example_manual_polling()
-    
+
     # For async example, you would need to run:
     # import asyncio
     # asyncio.run(example_async())

tests/api_resources/knowledge_bases/test_indexing_jobs.py

@@ -282,7 +282,9 @@ def test_wait_for_completion_raises_indexing_job_error_on_error(self, client: Gr
         assert "error" in str(exc_info.value).lower()
 
     @parametrize
-    def test_wait_for_completion_raises_indexing_job_error_on_cancelled(self, client: Gradient, respx_mock: Any) -> None:
+    def test_wait_for_completion_raises_indexing_job_error_on_cancelled(
+        self, client: Gradient, respx_mock: Any
+    ) -> None:
         """Test that IndexingJobError is raised when job phase is CANCELLED"""
         job_uuid = "test-job-uuid"
         respx_mock.get(f"{base_url}/v2/gen-ai/indexing_jobs/{job_uuid}").mock(
@@ -321,9 +323,7 @@ def test_wait_for_completion_raises_timeout_error(self, client: Gradient, respx_
         )
 
         with pytest.raises(IndexingJobTimeoutError) as exc_info:
-            client.knowledge_bases.indexing_jobs.wait_for_completion(
-                job_uuid, poll_interval=0.1, timeout=0.2
-            )
+            client.knowledge_bases.indexing_jobs.wait_for_completion(job_uuid, poll_interval=0.1, timeout=0.2)
 
         assert exc_info.value.uuid == job_uuid
         assert exc_info.value.phase == "BATCH_JOB_PHASE_RUNNING"
@@ -567,7 +567,9 @@ async def test_path_params_update_cancel(self, async_client: AsyncGradient) -> N
             )
 
     @parametrize
-    async def test_wait_for_completion_raises_indexing_job_error_on_failed(self, async_client: AsyncGradient, respx_mock: Any) -> None:
+    async def test_wait_for_completion_raises_indexing_job_error_on_failed(
+        self, async_client: AsyncGradient, respx_mock: Any
+    ) -> None:
         """Test that IndexingJobError is raised when job phase is FAILED"""
         job_uuid = "test-job-uuid"
         respx_mock.get(f"{base_url}/v2/gen-ai/indexing_jobs/{job_uuid}").mock(
@@ -592,7 +594,9 @@ async def test_wait_for_completion_raises_indexing_job_error_on_failed(self, asy
         assert "failed" in str(exc_info.value).lower()
 
     @parametrize
-    async def test_wait_for_completion_raises_indexing_job_error_on_error(self, async_client: AsyncGradient, respx_mock: Any) -> None:
+    async def test_wait_for_completion_raises_indexing_job_error_on_error(
+        self, async_client: AsyncGradient, respx_mock: Any
+    ) -> None:
         """Test that IndexingJobError is raised when job phase is ERROR"""
         job_uuid = "test-job-uuid"
         respx_mock.get(f"{base_url}/v2/gen-ai/indexing_jobs/{job_uuid}").mock(
@@ -615,7 +619,9 @@ async def test_wait_for_completion_raises_indexing_job_error_on_error(self, asyn
         assert "error" in str(exc_info.value).lower()
 
     @parametrize
-    async def test_wait_for_completion_raises_indexing_job_error_on_cancelled(self, async_client: AsyncGradient, respx_mock: Any) -> None:
+    async def test_wait_for_completion_raises_indexing_job_error_on_cancelled(
+        self, async_client: AsyncGradient, respx_mock: Any
+    ) -> None:
         """Test that IndexingJobError is raised when job phase is CANCELLED"""
         job_uuid = "test-job-uuid"
         respx_mock.get(f"{base_url}/v2/gen-ai/indexing_jobs/{job_uuid}").mock(