[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

integration_tests/base_routes.py

@@ -1060,91 +1060,105 @@ def create_item(request, body: CreateItemBody, query: CreateItemQueryParamsParam
 
 # ===== Server-Sent Events (SSE) Routes =====
 
+
 @app.get("/sse/basic")
 def sse_basic(request):
     """Basic SSE endpoint that sends 3 messages"""
+
     def event_generator():
         for i in range(3):
             yield f"data: Test message {i}\n\n"
+
     return sse_response(event_generator())
 
 
 @app.get("/sse/formatted")
 def sse_formatted(request):
     """SSE endpoint using sse_message formatter"""
+
     def event_generator():
         for i in range(3):
             yield sse_message(f"Formatted message {i}", event="test", id=str(i))
+
     return sse_response(event_generator())
 
 
 @app.get("/sse/json")
 def sse_json(request):
     """SSE endpoint that sends JSON data"""
     import json
+
     def event_generator():
         for i in range(3):
             data = {"id": i, "message": f"JSON message {i}", "type": "test"}
             yield f"data: {json.dumps(data)}\n\n"
+
     return sse_response(event_generator())
 
 
 @app.get("/sse/named_events")
 def sse_named_events(request):
     """SSE endpoint with different event types"""
+
     def event_generator():
-        events = [
-            ("start", "Test started"),
-            ("progress", "Test in progress"),
-            ("end", "Test completed")
-        ]
+        events = [("start", "Test started"), ("progress", "Test in progress"), ("end", "Test completed")]
         for event_type, message in events:
             yield sse_message(message, event=event_type)
+
     return sse_response(event_generator())
 
 
 @app.get("/sse/async")
 def sse_async(request):
     """Async SSE endpoint (sync implementation)"""
+
     def event_generator():
         # Simple generator for testing
         for i in range(3):
             yield f"data: Async message {i}\n\n"
-    
+
     return sse_response(event_generator())
 
 
 @app.get("/sse/single")
 def sse_single(request):
     """SSE endpoint that sends a single message and closes"""
+
     def event_generator():
         yield "data: Single message\n\n"
+
     return sse_response(event_generator())
 
 
 @app.get("/sse/empty")
 def sse_empty(request):
     """SSE endpoint that sends no messages"""
+
     def event_generator():
         return
         yield  # This will never be reached
+
     return sse_response(event_generator())
 
 
 @app.get("/sse/with_headers")
 def sse_with_headers(request):
     """SSE endpoint with custom headers"""
     headers = Headers({"X-Custom-Header": "custom-value"})
+
     def event_generator():
         yield "data: Message with custom headers\n\n"
+
     return sse_response(event_generator(), headers=headers)
 
 
 @app.get("/sse/status_code")
 def sse_status_code(request):
     """SSE endpoint with custom status code"""
+
     def event_generator():
         yield "data: Message with custom status\n\n"
+
     return sse_response(event_generator(), status_code=201)
 
 
@@ -1170,7 +1184,7 @@ def authenticate(self, request: Request) -> Optional[Identity]:
             return None
 
     app.configure_authentication(BasicAuthHandler(token_getter=BearerGetter()))
-    
+
     # Read port from environment variable if set, otherwise default to 8080
     port = int(os.getenv("ROBYN_PORT", "8080"))
     app.start(port=port, _check_port=False)

integration_tests/conftest.py

@@ -43,12 +43,12 @@ def start_server(domain: str, port: int, is_dev: bool = False) -> subprocess.Pop
     command = ["python3", base_routes]
     if is_dev:
         command.append("--dev")
-    
+
     # Ensure environment variables are properly set for the subprocess
     env = os.environ.copy()
     env["ROBYN_HOST"] = domain
     env["ROBYN_PORT"] = str(port)
-    
+
     if platform.system() == "Windows":
         command[0] = "python"
         process = subprocess.Popen(command, shell=True, creationflags=subprocess.CREATE_NEW_PROCESS_GROUP, env=env)
@@ -70,7 +70,7 @@ def start_server(domain: str, port: int, is_dev: bool = False) -> subprocess.Pop
             break  # We were able to reach the server, exit the loop
         except Exception:
             time.sleep(0.5)  # Longer delay before retrying
-    
+
     # Give the server a moment to fully initialize after accepting connections
     time.sleep(1)
     return process

integration_tests/test_base_url.py

@@ -28,4 +28,3 @@ def test_global_index_request(global_session):
     res = requests.get(f"{BASE_URL}")
     assert os.getenv("ROBYN_HOST") == f"{host}"
     assert res.status_code == 200
-