Fix all remaining linter errors and enable linter in CI

Co-authored-by: kevinbackhouse <4358136+kevinbackhouse@users.noreply.github.com>

Author: Copilot

.github/workflows/ci.yml

@@ -34,9 +34,7 @@ jobs:
       run: pip install --upgrade hatch
 
     - name: Run static analysis
-      run: |
-        # hatch fmt --check
-        echo linter errors will be fixed in a separate PR
+      run: hatch fmt --check
 
     - name: Run tests
       run: hatch test --python ${{ matrix.python-version }} --cover --randomize --parallel --retries 2 --retry-delay 1

src/seclab_taskflows/mcp_servers/alert_results_models.py

@@ -3,17 +3,16 @@
 
 from __future__ import annotations
 
-from typing import Optional
-
 from sqlalchemy import Column, ForeignKey, Integer, Text
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 
 
 class Base(DeclarativeBase):
     pass
 
+
 class AlertResults(Base):
-    __tablename__ = 'alert_results'
+    __tablename__ = "alert_results"
 
     canonical_id: Mapped[int] = mapped_column(primary_key=True)
     alert_id: Mapped[str]
@@ -22,29 +21,33 @@ class AlertResults(Base):
     language: Mapped[str]
     location: Mapped[str]
     result: Mapped[str] = mapped_column(Text)
-    created: Mapped[Optional[str]]
+    created: Mapped[str | None]
     valid: Mapped[bool] = mapped_column(nullable=False, default=True)
     completed: Mapped[bool] = mapped_column(nullable=False, default=False)
 
-    relationship('AlertFlowGraph', cascade='all, delete')
+    relationship("AlertFlowGraph", cascade="all, delete")
 
     def __repr__(self):
-        return (f"<AlertResults(alert_id={self.alert_id}, repo={self.repo}, "
-                f"rule={self.rule}, language={self.language}, location={self.location}, "
-                f"result={self.result}, created_at={self.created}, valid={self.valid}, completed={self.completed})>")
+        return (
+            f"<AlertResults(alert_id={self.alert_id}, repo={self.repo}, "
+            f"rule={self.rule}, language={self.language}, location={self.location}, "
+            f"result={self.result}, created_at={self.created}, valid={self.valid}, completed={self.completed})>"
+        )
+
 
 class AlertFlowGraph(Base):
-    __tablename__ = 'alert_flow_graph'
+    __tablename__ = "alert_flow_graph"
 
     id: Mapped[int] = mapped_column(primary_key=True)
-    alert_canonical_id = Column(Integer, ForeignKey('alert_results.canonical_id', ondelete='CASCADE'))
+    alert_canonical_id = Column(Integer, ForeignKey("alert_results.canonical_id", ondelete="CASCADE"))
     flow_data: Mapped[str] = mapped_column(Text)
     repo: Mapped[str]
-    prev: Mapped[Optional[str]]
-    next: Mapped[Optional[str]]
+    prev: Mapped[str | None]
+    next: Mapped[str | None]
     started: Mapped[bool] = mapped_column(nullable=False, default=False)
 
     def __repr__(self):
-        return (f"<AlertFlowGraph(alert_canonical_id={self.alert_canonical_id}, "
-                f"flow_data={self.flow_data}, repo={self.repo}, prev={self.prev}, next={self.next}, started={self.started})>")
-
+        return (
+            f"<AlertFlowGraph(alert_canonical_id={self.alert_canonical_id}, "
+            f"flow_data={self.flow_data}, repo={self.repo}, prev={self.prev}, next={self.next}, started={self.started})>"
+        )

src/seclab_taskflows/mcp_servers/codeql_python/codeql_sqlite_models.py

@@ -3,8 +3,6 @@
 
 from __future__ import annotations
 
-from typing import Optional
-
 from sqlalchemy import Text
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
@@ -14,16 +12,18 @@ class Base(DeclarativeBase):
 
 
 class Source(Base):
-    __tablename__ = 'source'
+    __tablename__ = "source"
 
     id: Mapped[int] = mapped_column(primary_key=True)
     repo: Mapped[str]
     source_location: Mapped[str]
     line: Mapped[int]
     source_type: Mapped[str]
-    notes: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    notes: Mapped[str | None] = mapped_column(Text, nullable=True)
 
     def __repr__(self):
-        return (f"<Source(id={self.id}, repo={self.repo}, "
-                f"location={self.source_location}, line={self.line}, source_type={self.source_type}, "
-                f"notes={self.notes})>")
+        return (
+            f"<Source(id={self.id}, repo={self.repo}, "
+            f"location={self.source_location}, line={self.line}, source_type={self.source_type}, "
+            f"notes={self.notes})>"
+        )

src/seclab_taskflows/mcp_servers/codeql_python/mcp_server.py

@@ -11,35 +11,34 @@
 import subprocess
 from pathlib import Path
 
-#from mcp.server.fastmcp import FastMCP, Context
 from fastmcp import FastMCP  # use FastMCP 2.0
 from pydantic import Field
 from seclab_taskflow_agent.mcp_servers.codeql.client import _debug_log, run_query
 from seclab_taskflow_agent.path_utils import log_file_name, mcp_data_dir
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
 
-from seclab_taskflows.mcp_servers.utils import process_repo
 from seclab_taskflows.mcp_servers.codeql_python.codeql_sqlite_models import Base, Source
+from seclab_taskflows.mcp_servers.utils import process_repo
 
 logging.basicConfig(
     level=logging.DEBUG,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    filename=log_file_name('mcp_codeql_python.log'),
-    filemode='a'
+    format="%(asctime)s - %(levelname)s - %(message)s",
+    filename=log_file_name("mcp_codeql_python.log"),
+    filemode="a",
 )
 
-MEMORY = mcp_data_dir('seclab-taskflows', 'codeql', 'DATA_DIR')
-CODEQL_DBS_BASE_PATH = mcp_data_dir('seclab-taskflows', 'codeql', 'CODEQL_DBS_BASE_PATH')
+logger = logging.getLogger(__name__)
+
+MEMORY = mcp_data_dir("seclab-taskflows", "codeql", "DATA_DIR")
+CODEQL_DBS_BASE_PATH = mcp_data_dir("seclab-taskflows", "codeql", "CODEQL_DBS_BASE_PATH")
 
 mcp = FastMCP("CodeQL-Python")
 
 # tool name -> templated query lookup for supported languages
 TEMPLATED_QUERY_PATHS = {
     # to add a language, port the templated query pack and add its definition here
-    'python': {
-        'remote_sources': 'queries/mcp-python/remote_sources.ql'
-    }
+    "python": {"remote_sources": "queries/mcp-python/remote_sources.ql"}
 }
 
 
@@ -50,9 +49,10 @@ def source_to_dict(result):
         "source_location": result.source_location,
         "line": result.line,
         "source_type": result.source_type,
-        "notes": result.notes
+        "notes": result.notes,
     }
 
+
 def _resolve_query_path(language: str, query: str) -> Path:
     if language not in TEMPLATED_QUERY_PATHS:
         msg = f"Error: Language `{language}` not supported!"
@@ -67,7 +67,7 @@ def _resolve_query_path(language: str, query: str) -> Path:
 def _resolve_db_path(relative_db_path: str | Path):
     # path joins will return "/B" if "/A" / "////B" etc. as well
     # not windows compatible and probably needs additional hardening
-    relative_db_path = str(relative_db_path).strip().lstrip('/')
+    relative_db_path = str(relative_db_path).strip().lstrip("/")
     relative_db_path = Path(relative_db_path)
     absolute_path = (CODEQL_DBS_BASE_PATH / relative_db_path).resolve()
     if not absolute_path.is_relative_to(CODEQL_DBS_BASE_PATH.resolve()):
@@ -79,37 +79,38 @@ def _resolve_db_path(relative_db_path: str | Path):
         raise RuntimeError(msg)
     return str(absolute_path)
 
+
 # This sqlite database is specifically made for CodeQL for Python MCP.
 class CodeqlSqliteBackend:
     def __init__(self, memcache_state_dir: str):
         self.memcache_state_dir = memcache_state_dir
         if not Path(self.memcache_state_dir).exists():
-            db_dir = 'sqlite://'
+            db_dir = "sqlite://"
         else:
-            db_dir = f'sqlite:///{self.memcache_state_dir}/codeql_sqlite.db'
+            db_dir = f"sqlite:///{self.memcache_state_dir}/codeql_sqlite.db"
         self.engine = create_engine(db_dir, echo=False)
         Base.metadata.create_all(self.engine, tables=[Source.__table__])
 
-
-    def store_new_source(self, repo, source_location, line, source_type, notes, *, update = False):
+    def store_new_source(self, repo, source_location, line, source_type, notes, *, update=False):
         with Session(self.engine) as session:
-            existing = session.query(Source).filter_by(repo = repo, source_location = source_location, line = line).first()
+            existing = session.query(Source).filter_by(repo=repo, source_location=source_location, line=line).first()
             if existing:
                 existing.notes = (existing.notes or "") + notes
                 session.commit()
                 return f"Updated notes for source at {source_location}, line {line} in {repo}."
             if update:
                 return f"No source exists at repo {repo}, location {source_location}, line {line} to update."
-            new_source = Source(repo = repo,  source_location = source_location, line = line, source_type = source_type, notes = notes)
+            new_source = Source(
+                repo=repo, source_location=source_location, line=line, source_type=source_type, notes=notes
+            )
             session.add(new_source)
             session.commit()
             return f"Added new source for {source_location} in {repo}."
 
     def get_sources(self, repo):
         with Session(self.engine) as session:
-            results = session.query(Source).filter_by(repo = repo).all()
-            sources = [source_to_dict(source) for source in results]
-        return sources
+            results = session.query(Source).filter_by(repo=repo).all()
+            return [source_to_dict(source) for source in results]
 
 
 # our query result format is: "human readable template {val0} {val1},'key0,key1',val0,val1"
@@ -121,8 +122,8 @@ def _csv_parse(raw):
             if i == 0:
                 continue
             # col1 has what we care about, but offer flexibility
-            keys = row[1].split(',')
-            this_obj = {'description': row[0].format(*row[2:])}
+            keys = row[1].split(",")
+            this_obj = {"description": row[0].format(*row[2:])}
             for j, k in enumerate(keys):
                 this_obj[k.strip()] = row[j + 2]
             results.append(this_obj)
@@ -143,27 +144,32 @@ def _run_query(query_name: str, database_path: str, language: str, template_valu
     except RuntimeError:
         return f"The query {query_name} is not supported for language: {language}"
     try:
-        csv = run_query(Path(__file__).parent.resolve() /
-                        query_path,
-                        database_path,
-                        fmt='csv',
-                        template_values=template_values,
-                        log_stderr=True)
+        csv = run_query(
+            Path(__file__).parent.resolve() / query_path,
+            database_path,
+            fmt="csv",
+            template_values=template_values,
+            log_stderr=True,
+        )
         return _csv_parse(csv)
-    except Exception as e:
+    except (subprocess.CalledProcessError, FileNotFoundError) as e:
         return f"The query {query_name} encountered an error: {e}"
 
+
 backend = CodeqlSqliteBackend(MEMORY)
 
+
 @mcp.tool()
-def remote_sources(owner: str = Field(description="The owner of the GitHub repository"),
-                   repo: str = Field(description="The name of the GitHub repository"),
-                   database_path: str = Field(description="The CodeQL database path."),
-                   language: str = Field(description="The language used for the CodeQL database.")):
+def remote_sources(
+    owner: str = Field(description="The owner of the GitHub repository"),
+    repo: str = Field(description="The name of the GitHub repository"),
+    database_path: str = Field(description="The CodeQL database path."),
+    language: str = Field(description="The language used for the CodeQL database."),
+):
     """List all remote sources and their locations in a CodeQL database, then store the results in a database."""
 
     repo = process_repo(owner, repo)
-    results = _run_query('remote_sources', database_path, language, {})
+    results = _run_query("remote_sources", database_path, language, {})
 
     # Check if results is an error (list of strings) or valid data (list of dicts)
     if isinstance(results, str):
@@ -174,53 +180,67 @@ def remote_sources(owner: str = Field(description="The owner of the GitHub repos
     for result in results:
         backend.store_new_source(
             repo=repo,
-            source_location=result.get('location', ''),
-            source_type=result.get('source', ''),
-            line=int(result.get('line', '0')),
-            notes=None, #result.get('description', ''),
-            update=False
+            source_location=result.get("location", ""),
+            source_type=result.get("source", ""),
+            line=int(result.get("line", "0")),
+            notes=None,  # result.get('description', ''),
+            update=False,
         )
         stored_count += 1
 
     return f"Stored {stored_count} remote sources in {repo}."
 
+
 @mcp.tool()
-def fetch_sources(owner: str = Field(description="The owner of the GitHub repository"),
-                     repo: str = Field(description="The name of the GitHub repository")):
+def fetch_sources(
+    owner: str = Field(description="The owner of the GitHub repository"),
+    repo: str = Field(description="The name of the GitHub repository"),
+):
     """
     Fetch all sources from the repo
     """
     repo = process_repo(owner, repo)
     return json.dumps(backend.get_sources(repo))
 
+
 @mcp.tool()
-def add_source_notes(owner: str = Field(description="The owner of the GitHub repository"),
-                     repo: str = Field(description="The name of the GitHub repository"),
-                     source_location: str = Field(description="The path to the file"),
-                     line: int = Field(description="The line number of the source"),
-                     notes: str = Field(description="The notes to append to this source")):
+def add_source_notes(
+    owner: str = Field(description="The owner of the GitHub repository"),
+    repo: str = Field(description="The name of the GitHub repository"),
+    source_location: str = Field(description="The path to the file"),
+    line: int = Field(description="The line number of the source"),
+    notes: str = Field(description="The notes to append to this source"),
+):
     """
     Add new notes to an existing source. The notes will be appended to any existing notes.
     """
     repo = process_repo(owner, repo)
-    return backend.store_new_source(repo = repo, source_location = source_location, line = line, source_type = "", notes = notes, update=True)
+    return backend.store_new_source(
+        repo=repo, source_location=source_location, line=line, source_type="", notes=notes, update=True
+    )
+
 
 @mcp.tool()
-def clear_codeql_repo(owner: str = Field(description="The owner of the GitHub repository"),
-                     repo: str = Field(description="The name of the GitHub repository")):
+def clear_codeql_repo(
+    owner: str = Field(description="The owner of the GitHub repository"),
+    repo: str = Field(description="The name of the GitHub repository"),
+):
     """
     Clear all data for a given repo from the database
     """
     repo = process_repo(owner, repo)
     with Session(backend.engine) as session:
-        deleted_sources = session.query(Source).filter_by(repo = repo).delete()
+        deleted_sources = session.query(Source).filter_by(repo=repo).delete()
         session.commit()
     return f"Cleared {deleted_sources} sources from repo {repo}."
 
+
 if __name__ == "__main__":
     # Check if codeql/python-all pack is installed, if not install it
-    if not os.path.isdir('/.codeql/packages/codeql/python-all'):
-        pack_path = importlib.resources.files('seclab_taskflows.mcp_servers.codeql_python.queries').joinpath('mcp-python')
-        print(f"Installing CodeQL pack from {pack_path}")
-        subprocess.run(["codeql", "pack", "install", pack_path], check=False)
+    if not os.path.isdir("/.codeql/packages/codeql/python-all"):
+        pack_path = importlib.resources.files("seclab_taskflows.mcp_servers.codeql_python.queries").joinpath(
+            "mcp-python"
+        )
+        logger.info("Installing CodeQL pack from %s", pack_path)
+        subprocess.run(["/usr/local/bin/codeql", "pack", "install", pack_path], check=False)
     mcp.run(show_banner=False, transport="http", host="127.0.0.1", port=9998)