style: reformat imports and strings in knowledge CLI

Author: mdesmet

src/datapilot/cli/main.py

@@ -6,9 +6,9 @@
 import click
 from dotenv import load_dotenv
 
+from datapilot.core.knowledge.cli import cli as knowledge
 from datapilot.core.mcp_utils.mcp import mcp
 from datapilot.core.platforms.dbt.cli.cli import dbt
-from datapilot.core.knowledge.cli import cli as knowledge
 
 
 def load_config_from_file():

src/datapilot/core/knowledge/cli.py

@@ -1,10 +1,14 @@
-import click
 import json
-from http.server import HTTPServer, BaseHTTPRequestHandler
-from urllib.parse import urlparse
-from urllib.request import Request, urlopen
-from urllib.error import URLError, HTTPError
 import re
+from http.server import BaseHTTPRequestHandler
+from http.server import HTTPServer
+from urllib.error import HTTPError
+from urllib.error import URLError
+from urllib.parse import urlparse
+from urllib.request import Request
+from urllib.request import urlopen
+
+import click
 
 
 @click.group(name="knowledge")
@@ -13,17 +17,19 @@ def cli():
 
 
 @cli.command()
-@click.option('--port', default=4000, help='Port to run the server on')
+@click.option("--port", default=4000, help="Port to run the server on")
 @click.pass_context
 def serve(ctx, port):
     """Serve knowledge bases via HTTP server."""
     # Get configuration from parent context
-    token = ctx.parent.obj.get('token')
-    instance_name = ctx.parent.obj.get('instance_name')
-    backend_url = ctx.parent.obj.get('backend_url')
+    token = ctx.parent.obj.get("token")
+    instance_name = ctx.parent.obj.get("instance_name")
+    backend_url = ctx.parent.obj.get("backend_url")
 
     if not token or not instance_name:
-        click.echo("Error: API token and instance name are required. Use --token and --instance-name options or set them in config.", err=True)
+        click.echo(
+            "Error: API token and instance name are required. Use --token and --instance-name options or set them in config.", err=True
+        )
         ctx.exit(1)
 
     class KnowledgeBaseHandler(BaseHTTPRequestHandler):
@@ -32,12 +38,12 @@ def do_GET(self):
             path = urlparse(self.path).path
 
             # Match /knowledge_bases/{uuid} pattern
-            match = re.match(r'^/knowledge_bases/([a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12})$', path)
+            match = re.match(r"^/knowledge_bases/([a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12})$", path)
 
             if match:
                 public_id = match.group(1)
                 self.handle_knowledge_base(public_id)
-            elif path == '/health':
+            elif path == "/health":
                 self.handle_health()
             else:
                 self.send_error(404, "Not Found")
@@ -46,46 +52,42 @@ def handle_knowledge_base(self, public_id):
             """Fetch and return knowledge base data."""
             url = f"{backend_url}/knowledge_bases/public/{public_id}"
 
-            headers = {
-                'Authorization': f'Bearer {token}',
-                'X-Tenant': instance_name,
-                'Content-Type': 'application/json'
-            }
+            headers = {"Authorization": f"Bearer {token}", "X-Tenant": instance_name, "Content-Type": "application/json"}
 
             req = Request(url, headers=headers)
 
             try:
                 with urlopen(req) as response:
                     data = response.read()
                     self.send_response(200)
-                    self.send_header('Content-Type', 'application/json')
+                    self.send_header("Content-Type", "application/json")
                     self.end_headers()
                     self.wfile.write(data)
             except HTTPError as e:
-                error_data = e.read().decode('utf-8') if e.read() else '{"error": "HTTP Error"}'
+                error_data = e.read().decode("utf-8") if e.read() else '{"error": "HTTP Error"}'
                 self.send_response(e.code)
-                self.send_header('Content-Type', 'application/json')
+                self.send_header("Content-Type", "application/json")
                 self.end_headers()
-                self.wfile.write(error_data.encode('utf-8'))
+                self.wfile.write(error_data.encode("utf-8"))
             except URLError as e:
                 self.send_response(500)
-                self.send_header('Content-Type', 'application/json')
+                self.send_header("Content-Type", "application/json")
                 self.end_headers()
-                error_msg = json.dumps({'error': str(e)})
-                self.wfile.write(error_msg.encode('utf-8'))
+                error_msg = json.dumps({"error": str(e)})
+                self.wfile.write(error_msg.encode("utf-8"))
 
         def handle_health(self):
             """Handle health check endpoint."""
             self.send_response(200)
-            self.send_header('Content-Type', 'application/json')
+            self.send_header("Content-Type", "application/json")
             self.end_headers()
-            self.wfile.write(json.dumps({'status': 'ok'}).encode('utf-8'))
+            self.wfile.write(json.dumps({"status": "ok"}).encode("utf-8"))
 
         def log_message(self, format, *args):
             """Override to use click.echo for logging."""
             click.echo(f"{self.address_string()} - {format % args}")
 
-    server_address = ('', port)
+    server_address = ("", port)
     httpd = HTTPServer(server_address, KnowledgeBaseHandler)
 
     click.echo(f"Starting knowledge base server on port {port}...")
@@ -98,4 +100,3 @@ def log_message(self, format, *args):
     except KeyboardInterrupt:
         click.echo("\nShutting down server...")
         httpd.shutdown()
-