fix: resolve mypy and ruff CI errors across all source and test files

- Fix Traversable import with TYPE_CHECKING guard for Python 3.10-3.14 compat
- Add explicit dict[str, object] type annotations to all request dicts
- Fix unused variable and import warnings in connection.py and test files
- Break long lines to satisfy ruff E501 (100-char limit)
- Fix click.MultiCommand -> click.Group for mypy arg-type in repl.py
- Wrap binary_manager return in str() for mypy no-any-return
This commit is contained in:
MinaSaad1 2026-03-26 14:51:35 +02:00
parent 038e5b433c
commit 11691c8d92
25 changed files with 127 additions and 55 deletions

View file

@ -82,7 +82,10 @@ def create(
"translatedDescription": translated_description,
},
)
run_tool(ctx, "object_translation_operations", {"operation": "Create", "definitions": [definition]})
run_tool(ctx, "object_translation_operations", {
"operation": "Create",
"definitions": [definition],
})
# --- Function ---

View file

@ -31,7 +31,10 @@ def create(ctx: PbiContext, name: str, description: str | None, precedence: int
required={"name": name},
optional={"description": description, "calculationGroupPrecedence": precedence},
)
run_tool(ctx, "calculation_group_operations", {"operation": "CreateGroup", "definitions": [definition]})
run_tool(ctx, "calculation_group_operations", {
"operation": "CreateGroup",
"definitions": [definition],
})
@calc_group.command()
@ -47,7 +50,10 @@ def delete(ctx: PbiContext, name: str) -> None:
@pass_context
def list_items(ctx: PbiContext, group_name: str) -> None:
"""List calculation items in a group."""
run_tool(ctx, "calculation_group_operations", {"operation": "ListItems", "calculationGroupName": group_name})
run_tool(ctx, "calculation_group_operations", {
"operation": "ListItems",
"calculationGroupName": group_name,
})
@calc_group.command(name="create-item")
@ -56,7 +62,9 @@ def list_items(ctx: PbiContext, group_name: str) -> None:
@click.option("--expression", "-e", required=True, help="DAX expression.")
@click.option("--ordinal", type=int, default=None, help="Item ordinal.")
@pass_context
def create_item(ctx: PbiContext, item_name: str, group: str, expression: str, ordinal: int | None) -> None:
def create_item(
ctx: PbiContext, item_name: str, group: str, expression: str, ordinal: int | None
) -> None:
"""Create a calculation item in a group."""
definition = build_definition(
required={"name": item_name, "expression": expression},

View file

@ -33,7 +33,9 @@ def get(ctx: PbiContext, name: str, table: str) -> None:
@column.command()
@click.argument("name")
@click.option("--table", "-t", required=True, help="Table name.")
@click.option("--data-type", required=True, help="Data type (string, int64, double, datetime, etc.).")
@click.option(
"--data-type", required=True, help="Data type (string, int64, double, datetime, etc.)."
)
@click.option("--source-column", default=None, help="Source column name (for Import mode).")
@click.option("--expression", default=None, help="DAX expression (for calculated columns).")
@click.option("--format-string", default=None, help="Format string.")
@ -101,4 +103,8 @@ def rename(ctx: PbiContext, old_name: str, new_name: str, table: str) -> None:
@pass_context
def export_tmdl(ctx: PbiContext, name: str, table: str) -> None:
"""Export a column as TMDL."""
run_tool(ctx, "column_operations", {"operation": "ExportTMDL", "name": name, "tableName": table})
run_tool(ctx, "column_operations", {
"operation": "ExportTMDL",
"name": name,
"tableName": table,
})

View file

@ -25,14 +25,20 @@ from pbi_cli.main import PbiContext, pass_context
@click.command()
@click.option("--data-source", "-d", required=True, help="Data source (e.g., localhost:54321).")
@click.option("--catalog", "-C", default="", help="Initial catalog / dataset name.")
@click.option("--name", "-n", default=None, help="Name for this connection (auto-generated if omitted).")
@click.option("--connection-string", default="", help="Full connection string (overrides data-source).")
@click.option(
"--name", "-n", default=None, help="Name for this connection (auto-generated if omitted)."
)
@click.option(
"--connection-string", default="", help="Full connection string (overrides data-source)."
)
@pass_context
def connect(ctx: PbiContext, data_source: str, catalog: str, name: str | None, connection_string: str) -> None:
def connect(
ctx: PbiContext, data_source: str, catalog: str, name: str | None, connection_string: str
) -> None:
"""Connect to a Power BI instance via data source."""
conn_name = name or _auto_name(data_source)
request: dict = {
request: dict[str, object] = {
"operation": "Connect",
"connectionName": conn_name,
"dataSource": data_source,
@ -75,11 +81,13 @@ def connect(ctx: PbiContext, data_source: str, catalog: str, name: str | None, c
@click.option("--name", "-n", default=None, help="Name for this connection.")
@click.option("--tenant", default="myorg", help="Tenant name for B2B scenarios.")
@pass_context
def connect_fabric(ctx: PbiContext, workspace: str, model: str, name: str | None, tenant: str) -> None:
def connect_fabric(
ctx: PbiContext, workspace: str, model: str, name: str | None, tenant: str
) -> None:
"""Connect to a Fabric workspace semantic model."""
conn_name = name or f"{workspace}/{model}"
request: dict = {
request: dict[str, object] = {
"operation": "ConnectFabric",
"connectionName": conn_name,
"workspaceName": workspace,
@ -116,7 +124,9 @@ def connect_fabric(ctx: PbiContext, workspace: str, model: str, name: str | None
@click.command()
@click.option("--name", "-n", default=None, help="Connection name to disconnect (defaults to active).")
@click.option(
"--name", "-n", default=None, help="Connection name to disconnect (defaults to active)."
)
@pass_context
def disconnect(ctx: PbiContext, name: str | None) -> None:
"""Disconnect from the active or named connection."""
@ -130,7 +140,7 @@ def disconnect(ctx: PbiContext, name: str | None) -> None:
repl = ctx.repl_mode
client = get_client(repl_mode=repl)
try:
result = client.call_tool("connection_operations", {
client.call_tool("connection_operations", {
"operation": "Disconnect",
"connectionName": target,
})

View file

@ -56,7 +56,7 @@ def export_tmsl(ctx: PbiContext) -> None:
@pass_context
def deploy(ctx: PbiContext, workspace: str, new_name: str | None, tenant: str | None) -> None:
"""Deploy the model to a Fabric workspace."""
deploy_request: dict = {"targetWorkspaceName": workspace}
deploy_request: dict[str, object] = {"targetWorkspaceName": workspace}
if new_name:
deploy_request["newDatabaseName"] = new_name
if tenant:

View file

@ -18,10 +18,14 @@ def dax() -> None:
@dax.command()
@click.argument("query", default="")
@click.option("--file", "-f", "query_file", type=click.Path(exists=True), help="Read query from file.")
@click.option(
"--file", "-f", "query_file", type=click.Path(exists=True), help="Read query from file."
)
@click.option("--max-rows", type=int, default=None, help="Maximum rows to return.")
@click.option("--metrics", is_flag=True, default=False, help="Include execution metrics.")
@click.option("--metrics-only", is_flag=True, default=False, help="Return metrics without row data.")
@click.option(
"--metrics-only", is_flag=True, default=False, help="Return metrics without row data."
)
@click.option("--timeout", type=int, default=200, help="Query timeout in seconds.")
@pass_context
def execute(
@ -48,7 +52,7 @@ def execute(
print_error("No query provided. Pass as argument, --file, or stdin.")
raise SystemExit(1)
request: dict = {
request: dict[str, object] = {
"operation": "Execute",
"query": resolved_query,
"timeoutSeconds": timeout,
@ -73,7 +77,9 @@ def execute(
@dax.command()
@click.argument("query", default="")
@click.option("--file", "-f", "query_file", type=click.Path(exists=True), help="Read query from file.")
@click.option(
"--file", "-f", "query_file", type=click.Path(exists=True), help="Read query from file."
)
@click.option("--timeout", type=int, default=10, help="Validation timeout in seconds.")
@pass_context
def validate(ctx: PbiContext, query: str, query_file: str | None, timeout: int) -> None:
@ -83,7 +89,7 @@ def validate(ctx: PbiContext, query: str, query_file: str | None, timeout: int)
print_error("No query provided.")
raise SystemExit(1)
request: dict = {
request: dict[str, object] = {
"operation": "Validate",
"query": resolved_query,
"timeoutSeconds": timeout,
@ -106,7 +112,7 @@ def validate(ctx: PbiContext, query: str, query_file: str | None, timeout: int)
@pass_context
def clear_cache(ctx: PbiContext) -> None:
"""Clear the DAX query cache."""
request: dict = {"operation": "ClearCache"}
request: dict[str, object] = {"operation": "ClearCache"}
if ctx.connection:
request["connectionName"] = ctx.connection

View file

@ -39,7 +39,9 @@ def create(ctx: PbiContext, name: str, expression: str, description: str | None)
required={"name": name, "expression": expression},
optional={"description": description},
)
run_tool(ctx, "named_expression_operations", {"operation": "Create", "definitions": [definition]})
run_tool(ctx, "named_expression_operations", {
"operation": "Create", "definitions": [definition],
})
@expression.command()
@ -61,4 +63,6 @@ def create_param(ctx: PbiContext, name: str, expression: str, description: str |
required={"name": name, "expression": expression},
optional={"description": description},
)
run_tool(ctx, "named_expression_operations", {"operation": "CreateParameter", "definitions": [definition]})
run_tool(ctx, "named_expression_operations", {
"operation": "CreateParameter", "definitions": [definition],
})

View file

@ -18,7 +18,7 @@ def hierarchy() -> None:
@pass_context
def hierarchy_list(ctx: PbiContext, table: str | None) -> None:
"""List hierarchies."""
request: dict = {"operation": "List"}
request: dict[str, object] = {"operation": "List"}
if table:
request["tableName"] = table
run_tool(ctx, "user_hierarchy_operations", request)
@ -30,7 +30,11 @@ def hierarchy_list(ctx: PbiContext, table: str | None) -> None:
@pass_context
def get(ctx: PbiContext, name: str, table: str) -> None:
"""Get hierarchy details."""
run_tool(ctx, "user_hierarchy_operations", {"operation": "Get", "name": name, "tableName": table})
run_tool(ctx, "user_hierarchy_operations", {
"operation": "Get",
"name": name,
"tableName": table,
})
@hierarchy.command()
@ -53,4 +57,8 @@ def create(ctx: PbiContext, name: str, table: str, description: str | None) -> N
@pass_context
def delete(ctx: PbiContext, name: str, table: str) -> None:
"""Delete a hierarchy."""
run_tool(ctx, "user_hierarchy_operations", {"operation": "Delete", "name": name, "tableName": table})
run_tool(ctx, "user_hierarchy_operations", {
"operation": "Delete",
"name": name,
"tableName": table,
})

View file

@ -20,7 +20,7 @@ def measure() -> None:
@pass_context
def measure_list(ctx: PbiContext, table: str | None) -> None:
"""List all measures."""
request: dict = {"operation": "List"}
request: dict[str, object] = {"operation": "List"}
if table:
request["tableName"] = table
run_tool(ctx, "measure_operations", request)

View file

@ -28,7 +28,11 @@ def stats(ctx: PbiContext) -> None:
@model.command()
@click.option("--type", "refresh_type", type=click.Choice(["Automatic", "Full", "Calculate", "DataOnly", "Defragment"]), default="Automatic", help="Refresh type.")
@click.option(
"--type", "refresh_type",
type=click.Choice(["Automatic", "Full", "Calculate", "DataOnly", "Defragment"]),
default="Automatic", help="Refresh type.",
)
@pass_context
def refresh(ctx: PbiContext, refresh_type: str) -> None:
"""Refresh the model."""

View file

@ -27,7 +27,9 @@ def partition_list(ctx: PbiContext, table: str) -> None:
@click.option("--expression", "-e", default=None, help="M/Power Query expression.")
@click.option("--mode", type=click.Choice(["Import", "DirectQuery", "Dual"]), default=None)
@pass_context
def create(ctx: PbiContext, name: str, table: str, expression: str | None, mode: str | None) -> None:
def create(
ctx: PbiContext, name: str, table: str, expression: str | None, mode: str | None
) -> None:
"""Create a partition."""
definition = build_definition(
required={"name": name, "tableName": table},
@ -51,4 +53,8 @@ def delete(ctx: PbiContext, name: str, table: str) -> None:
@pass_context
def refresh(ctx: PbiContext, name: str, table: str) -> None:
"""Refresh a partition."""
run_tool(ctx, "partition_operations", {"operation": "Refresh", "name": name, "tableName": table})
run_tool(ctx, "partition_operations", {
"operation": "Refresh",
"name": name,
"tableName": table,
})

View file

@ -34,7 +34,11 @@ def get(ctx: PbiContext, name: str) -> None:
@click.option("--from-column", required=True, help="Source column.")
@click.option("--to-table", required=True, help="Target (one-side) table.")
@click.option("--to-column", required=True, help="Target column.")
@click.option("--cross-filter", type=click.Choice(["OneDirection", "BothDirections", "Automatic"]), default="OneDirection", help="Cross-filtering behavior.")
@click.option(
"--cross-filter",
type=click.Choice(["OneDirection", "BothDirections", "Automatic"]),
default="OneDirection", help="Cross-filtering behavior.",
)
@click.option("--active/--inactive", default=True, help="Whether the relationship is active.")
@pass_context
def create(

View file

@ -5,6 +5,10 @@ from __future__ import annotations
import importlib.resources
import shutil
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from importlib.abc import Traversable
import click
@ -13,10 +17,10 @@ from pbi_cli.main import pass_context
SKILLS_TARGET_DIR = Path.home() / ".claude" / "skills"
def _get_bundled_skills() -> dict[str, importlib.resources.abc.Traversable]:
def _get_bundled_skills() -> dict[str, Traversable]:
"""Return a mapping of skill-name -> Traversable for each bundled skill."""
skills_pkg = importlib.resources.files("pbi_cli.skills")
result: dict[str, importlib.resources.abc.Traversable] = {}
result: dict[str, Traversable] = {}
for item in skills_pkg.iterdir():
if item.is_dir() and (item / "SKILL.md").is_file():
result[item.name] = item

View file

@ -32,7 +32,10 @@ def get(ctx: PbiContext, name: str) -> None:
@table.command()
@click.argument("name")
@click.option("--mode", type=click.Choice(["Import", "DirectQuery", "Dual"]), default="Import", help="Table mode.")
@click.option(
"--mode", type=click.Choice(["Import", "DirectQuery", "Dual"]),
default="Import", help="Table mode.",
)
@click.option("--m-expression", default=None, help="M/Power Query expression (use - for stdin).")
@click.option("--dax-expression", default=None, help="DAX expression for calculated tables.")
@click.option("--sql-query", default=None, help="SQL query for DirectQuery.")
@ -82,7 +85,11 @@ def delete(ctx: PbiContext, name: str) -> None:
@table.command()
@click.argument("name")
@click.option("--type", "refresh_type", type=click.Choice(["Full", "Automatic", "Calculate", "DataOnly"]), default="Automatic", help="Refresh type.")
@click.option(
"--type", "refresh_type",
type=click.Choice(["Full", "Automatic", "Calculate", "DataOnly"]),
default="Automatic", help="Refresh type.",
)
@pass_context
def refresh(ctx: PbiContext, name: str, refresh_type: str) -> None:
"""Refresh a table."""

View file

@ -25,7 +25,7 @@ def begin(ctx: PbiContext) -> None:
@pass_context
def commit(ctx: PbiContext, transaction_id: str) -> None:
"""Commit the active or specified transaction."""
request: dict = {"operation": "Commit"}
request: dict[str, object] = {"operation": "Commit"}
if transaction_id:
request["transactionId"] = transaction_id
run_tool(ctx, "transaction_operations", request)
@ -36,7 +36,7 @@ def commit(ctx: PbiContext, transaction_id: str) -> None:
@pass_context
def rollback(ctx: PbiContext, transaction_id: str) -> None:
"""Rollback the active or specified transaction."""
request: dict = {"operation": "Rollback"}
request: dict[str, object] = {"operation": "Rollback"}
if transaction_id:
request["transactionId"] = transaction_id
run_tool(ctx, "transaction_operations", request)

View file

@ -126,7 +126,7 @@ def query_latest_version() -> str:
if not versions:
raise RuntimeError(f"No versions found for {EXTENSION_ID}")
return versions[0]["version"]
return str(versions[0]["version"])
def download_and_extract(version: str | None = None) -> Path:

View file

@ -43,7 +43,9 @@ def load_connections() -> ConnectionStore:
raw = json.loads(CONNECTIONS_FILE.read_text(encoding="utf-8"))
conns = {}
for name, data in raw.get("connections", {}).items():
conns[name] = ConnectionInfo(name=name, **{k: v for k, v in data.items() if k != "name"})
conns[name] = ConnectionInfo(
name=name, **{k: v for k, v in data.items() if k != "name"}
)
return ConnectionStore(
last_used=raw.get("last_used", ""),
connections=conns,
@ -79,7 +81,9 @@ def remove_connection(store: ConnectionStore, name: str) -> ConnectionStore:
return ConnectionStore(last_used=new_last, connections=new_conns)
def get_active_connection(store: ConnectionStore, override: str | None = None) -> ConnectionInfo | None:
def get_active_connection(
store: ConnectionStore, override: str | None = None
) -> ConnectionInfo | None:
"""Get the active connection: explicit override, or last-used."""
name = override or store.last_used
if not name:

View file

@ -25,8 +25,13 @@ pass_context = click.make_pass_decorator(PbiContext, ensure=True)
@click.group()
@click.option("--json", "json_output", is_flag=True, default=False, help="Output raw JSON for agent consumption.")
@click.option("--connection", "-c", default=None, help="Named connection to use (defaults to last-used).")
@click.option(
"--json", "json_output", is_flag=True, default=False,
help="Output raw JSON for agent consumption.",
)
@click.option(
"--connection", "-c", default=None, help="Named connection to use (defaults to last-used)."
)
@click.version_option(version=__version__, prog_name="pbi-cli")
@click.pass_context
def cli(ctx: click.Context, json_output: bool, connection: str | None) -> None:

View file

@ -45,7 +45,7 @@ class PbiRepl:
words: list[str] = []
for name, cmd in cli.commands.items():
words.append(name)
if isinstance(cmd, click.MultiCommand):
if isinstance(cmd, click.Group):
sub_names = cmd.list_commands(click.Context(cmd))
for sub in sub_names:
words.append(f"{name} {sub}")

View file

@ -8,7 +8,6 @@ from typing import Any
import pytest
from click.testing import CliRunner
# ---------------------------------------------------------------------------
# Canned MCP responses used by the mock client
# ---------------------------------------------------------------------------
@ -114,7 +113,9 @@ def mock_client() -> MockPbiMcpClient:
@pytest.fixture
def patch_get_client(monkeypatch: pytest.MonkeyPatch, mock_client: MockPbiMcpClient) -> MockPbiMcpClient:
def patch_get_client(
monkeypatch: pytest.MonkeyPatch, mock_client: MockPbiMcpClient
) -> MockPbiMcpClient:
"""Monkeypatch get_client in _helpers and connection modules."""
factory = lambda repl_mode=False: mock_client # noqa: E731

View file

@ -3,8 +3,8 @@
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
import pytest
from click.testing import CliRunner
from pbi_cli.main import cli
@ -74,10 +74,9 @@ def test_repl_execute_line_quit() -> None:
def test_repl_execute_line_strips_pbi_prefix(
monkeypatch: "pytest.MonkeyPatch",
monkeypatch: pytest.MonkeyPatch,
tmp_connections: Path,
) -> None:
import pytest
from tests.conftest import MockPbiMcpClient
mock = MockPbiMcpClient()

View file

@ -2,7 +2,6 @@
from __future__ import annotations
import json
from pathlib import Path
from pbi_cli.core.config import PbiConfig, load_config, save_config

View file

@ -11,7 +11,6 @@ import sys
import pytest
pytestmark = pytest.mark.e2e

View file

@ -2,8 +2,6 @@
from __future__ import annotations
from pathlib import Path
import pytest
from pbi_cli.commands._helpers import build_definition, run_tool

View file

@ -2,8 +2,6 @@
from __future__ import annotations
import json
from pbi_cli.core.mcp_client import (
_extract_text,
_parse_content,
@ -11,7 +9,6 @@ from pbi_cli.core.mcp_client import (
get_shared_client,
)
# ---------------------------------------------------------------------------
# _parse_content tests
# ---------------------------------------------------------------------------