airbyte_ops_mcp.cli.gh

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""CLI commands for GitHub operations.
  3
  4Commands:
  5    airbyte-ops gh workflow status - Check GitHub Actions workflow status
  6    airbyte-ops gh workflow trigger - Trigger a GitHub Actions CI workflow
  7    airbyte-ops gh connector info - Get connector metadata from GitHub
  8    airbyte-ops gh connector get-version - Get connector version from GitHub
  9    airbyte-ops gh connector list - List connectors via GitHub API
 10
 11## CLI reference
 12
 13The commands below are regenerated by `poe docs-generate` via cyclopts's
 14programmatic docs API; see `docs/generate_cli.py`.
 15
 16.. include:: ../../../docs/generated/cli/gh.md
 17   :start-line: 2
 18"""
 19
 20from __future__ import annotations
 21
 22# Hide Python-level members from the pdoc page for this module; the rendered
 23# docs for this CLI group come entirely from the grafted `.. include::` in
 24# the module docstring above.
 25__all__: list[str] = []
 26
 27import json
 28import sys
 29import time
 30from typing import Annotated, Literal
 31
 32from cyclopts import Parameter
 33
 34from airbyte_ops_mcp.airbyte_repo.list_connectors import (
 35    format_github_actions_connector_matrix,
 36    get_modified_connectors_from_github,
 37)
 38from airbyte_ops_mcp.airbyte_repo.utils import parse_pr_info
 39from airbyte_ops_mcp.cli._base import App, app
 40from airbyte_ops_mcp.cli._shared import exit_with_error, print_json
 41from airbyte_ops_mcp.connector_metadata import (
 42    ConnectorMetadataDpathError,
 43    ConnectorMetadataDpathNotFoundError,
 44    format_metadata_dpath_value,
 45    get_connector_version_from_metadata,
 46    load_raw_connector_metadata_from_github,
 47)
 48from airbyte_ops_mcp.github_api import GitHubAPIError, resolve_default_github_token
 49from airbyte_ops_mcp.mcp.github_actions import (
 50    check_ci_workflow_status,
 51    trigger_ci_workflow,
 52)
 53
 54AIRBYTE_REPO_OWNER = "airbytehq"
 55AIRBYTE_REPO_NAME = "airbyte"
 56
 57# Create the gh sub-app
 58gh_app = App(name="gh", help="GitHub operations.")
 59app.command(gh_app)
 60
 61# Create the workflow sub-app under gh
 62workflow_app = App(name="workflow", help="GitHub Actions workflow operations.")
 63gh_app.command(workflow_app)
 64
 65# Create the connector sub-app under gh
 66connector_app = App(name="connector", help="Connector operations via GitHub API.")
 67gh_app.command(connector_app)
 68
 69
 70ConnectorListOutputFormat = Literal["lines", "csv", "json", "json-gh-matrix"]
 71
 72
 73def _parse_pr_details(pr: str) -> tuple[int, str | None, str | None]:
 74    """Parse pull request details from a number or GitHub PR URL."""
 75    pr_number, pr_owner, pr_repo = parse_pr_info(pr)
 76    if pr_number is None:
 77        exit_with_error(
 78            "PR must be a pull request number or GitHub URL like "
 79            "https://github.com/airbytehq/airbyte/pull/123."
 80        )
 81    return pr_number, pr_owner, pr_repo
 82
 83
 84def _print_connector_list(
 85    connectors: list[str],
 86    output_format: ConnectorListOutputFormat,
 87) -> None:
 88    """Print connector names in the requested format."""
 89    if output_format == "lines":
 90        for connector in connectors:
 91            sys.stdout.write(connector + "\n")
 92    elif output_format == "csv":
 93        sys.stdout.write(",".join(connectors) + "\n")
 94    elif output_format == "json":
 95        sys.stdout.write(json.dumps(connectors, separators=(",", ":")) + "\n")
 96    elif output_format == "json-gh-matrix":
 97        sys.stdout.write(
 98            json.dumps(
 99                format_github_actions_connector_matrix(connectors),
100                separators=(",", ":"),
101            )
102            + "\n"
103        )
104
105
106@connector_app.command(name="info")
107def connector_info(
108    name: Annotated[
109        str,
110        Parameter(help="Connector technical name (e.g., source-github)."),
111    ],
112    ref: Annotated[
113        str,
114        Parameter(help="Git ref to read metadata.yaml from."),
115    ],
116    owner: Annotated[
117        str,
118        Parameter(help="Repository owner (e.g., airbytehq)."),
119    ] = AIRBYTE_REPO_OWNER,
120    repo: Annotated[
121        str,
122        Parameter(help="Repository name (e.g., airbyte or airbyte-enterprise)."),
123    ] = AIRBYTE_REPO_NAME,
124    gh_token: Annotated[
125        str | None,
126        Parameter(
127            help=(
128                "GitHub API token. If omitted, uses `resolve_default_github_token`; "
129                "public repos can fall back to unauthenticated requests."
130            )
131        ),
132    ] = None,
133    dpath_expression: Annotated[
134        str | None,
135        Parameter(
136            name="--dpath",
137            help=(
138                "Evaluate this dpath expression against the parsed metadata.yaml "
139                "object and print only that value (e.g., data/dockerImageTag)."
140            ),
141        ),
142    ] = None,
143) -> None:
144    """Get connector metadata from GitHub."""
145    try:
146        value = load_raw_connector_metadata_from_github(
147            name,
148            owner=owner,
149            repo=repo,
150            ref=ref,
151            gh_token=gh_token or resolve_default_github_token(allow_none=True),
152            dpath_expression=dpath_expression,
153        )
154    except (
155        ConnectorMetadataDpathError,
156        ConnectorMetadataDpathNotFoundError,
157        FileNotFoundError,
158        ValueError,
159    ) as e:
160        exit_with_error(str(e))
161
162    if dpath_expression is None:
163        print_json(value)
164        return
165
166    sys.stdout.write(format_metadata_dpath_value(value) + "\n")
167
168
169@connector_app.command(name="get-version")
170def connector_get_version(
171    name: Annotated[
172        str,
173        Parameter(help="Connector technical name (e.g., source-github)."),
174    ],
175    ref: Annotated[
176        str,
177        Parameter(help="Git ref to read metadata.yaml from."),
178    ],
179    owner: Annotated[
180        str,
181        Parameter(help="Repository owner (e.g., airbytehq)."),
182    ] = AIRBYTE_REPO_OWNER,
183    repo: Annotated[
184        str,
185        Parameter(help="Repository name (e.g., airbyte or airbyte-enterprise)."),
186    ] = AIRBYTE_REPO_NAME,
187    gh_token: Annotated[
188        str | None,
189        Parameter(
190            help=(
191                "GitHub API token. If omitted, uses `resolve_default_github_token`; "
192                "public repos can fall back to unauthenticated requests."
193            )
194        ),
195    ] = None,
196) -> None:
197    """Get connector version from GitHub.
198
199    This is analogous to `gh connector info --dpath data/dockerImageTag` and uses
200    the same dpath evaluation internally.
201    """
202    try:
203        metadata = load_raw_connector_metadata_from_github(
204            name,
205            owner=owner,
206            repo=repo,
207            ref=ref,
208            gh_token=gh_token or resolve_default_github_token(allow_none=True),
209        )
210        version = get_connector_version_from_metadata(metadata)
211    except (
212        ConnectorMetadataDpathError,
213        ConnectorMetadataDpathNotFoundError,
214        FileNotFoundError,
215        ValueError,
216    ) as e:
217        exit_with_error(str(e))
218    sys.stdout.write(version + "\n")
219
220
221@connector_app.command(name="list")
222def connector_list(
223    modified_only: Annotated[
224        bool,
225        Parameter(help="Include only modified connectors from the provided PR."),
226    ] = False,
227    pr: Annotated[
228        str | None,
229        Parameter(
230            help="Pull request number or GitHub URL to inspect for changed connector files."
231        ),
232    ] = None,
233    owner: Annotated[
234        str,
235        Parameter(help="Repository owner (e.g., airbytehq)."),
236    ] = AIRBYTE_REPO_OWNER,
237    repo: Annotated[
238        str,
239        Parameter(help="Repository name (e.g., airbyte or airbyte-enterprise)."),
240    ] = AIRBYTE_REPO_NAME,
241    gh_token: Annotated[
242        str | None,
243        Parameter(
244            help=("GitHub API token. If omitted, uses `resolve_default_github_token`.")
245        ),
246    ] = None,
247    output_format: Annotated[
248        ConnectorListOutputFormat,
249        Parameter(
250            help=(
251                'Output format: "lines", "csv", "json", or "json-gh-matrix". '
252                'The GitHub Actions matrix format is {"connector":["source-x"]} '
253                'and returns {"connector":[""]} when no connectors changed.'
254            )
255        ),
256    ] = "lines",
257) -> None:
258    """List connector names from GitHub without a local checkout."""
259    if not modified_only:
260        exit_with_error("`--modified-only` is required for GitHub connector listing.")
261    if pr is None:
262        exit_with_error("`--pr` is required when `--modified-only` is set.")
263
264    pr_number, pr_owner, pr_repo = _parse_pr_details(pr)
265    try:
266        connectors = get_modified_connectors_from_github(
267            pr_number=pr_number,
268            pr_owner=pr_owner or owner,
269            pr_repo=pr_repo or repo,
270            gh_token=gh_token or resolve_default_github_token(),
271        )
272    except (GitHubAPIError, ValueError) as e:
273        exit_with_error(str(e))
274    _print_connector_list(connectors, output_format)
275
276
277@workflow_app.command(name="status")
278def workflow_status(
279    url: Annotated[
280        str | None,
281        Parameter(
282            help="Full GitHub Actions workflow run URL "
283            "(e.g., 'https://github.com/owner/repo/actions/runs/12345')."
284        ),
285    ] = None,
286    owner: Annotated[
287        str | None,
288        Parameter(help="Repository owner (e.g., 'airbytehq')."),
289    ] = None,
290    repo: Annotated[
291        str | None,
292        Parameter(help="Repository name (e.g., 'airbyte')."),
293    ] = None,
294    run_id: Annotated[
295        int | None,
296        Parameter(help="Workflow run ID."),
297    ] = None,
298) -> None:
299    """Check the status of a GitHub Actions workflow run.
300
301    Provide either --url OR all of (--owner, --repo, --run-id).
302    """
303    # Validate input parameters
304    if url:
305        if owner or repo or run_id:
306            exit_with_error(
307                "Cannot specify --url together with --owner/--repo/--run-id. "
308                "Use either --url OR the component parts."
309            )
310    elif not (owner and repo and run_id):
311        exit_with_error(
312            "Must provide either --url OR all of (--owner, --repo, --run-id)."
313        )
314
315    result = check_ci_workflow_status(
316        workflow_url=url,
317        owner=owner,
318        repo=repo,
319        run_id=run_id,
320    )
321    print_json(result.model_dump())
322
323
324@workflow_app.command(name="trigger")
325def workflow_trigger(
326    owner: Annotated[
327        str,
328        Parameter(help="Repository owner (e.g., 'airbytehq')."),
329    ],
330    repo: Annotated[
331        str,
332        Parameter(help="Repository name (e.g., 'airbyte')."),
333    ],
334    workflow_file: Annotated[
335        str,
336        Parameter(help="Workflow file name (e.g., 'connector-regression-test.yml')."),
337    ],
338    workflow_definition_ref: Annotated[
339        str | None,
340        Parameter(
341            help="Branch name or PR number for the workflow definition to use. "
342            "If a PR number is provided, it resolves to the PR's head branch name. "
343            "Defaults to 'main' if not specified."
344        ),
345    ] = None,
346    inputs: Annotated[
347        str | None,
348        Parameter(
349            help='Workflow inputs as a JSON string (e.g., \'{"key": "value"}\').'
350        ),
351    ] = None,
352    wait: Annotated[
353        bool,
354        Parameter(help="Wait for the workflow to complete before returning."),
355    ] = False,
356    wait_seconds: Annotated[
357        int,
358        Parameter(
359            help="Maximum seconds to wait for workflow completion (default: 600)."
360        ),
361    ] = 600,
362) -> None:
363    """Trigger a GitHub Actions CI workflow via workflow_dispatch.
364
365    This command triggers a workflow in any GitHub repository that has workflow_dispatch
366    enabled. It resolves PR numbers to branch names automatically.
367    """
368    # Parse inputs JSON if provided
369    parsed_inputs: dict[str, str] | None = None
370    if inputs:
371        try:
372            parsed_inputs = json.loads(inputs)
373        except json.JSONDecodeError as e:
374            exit_with_error(f"Invalid JSON for --inputs: {e}")
375
376    # Trigger the workflow
377    result = trigger_ci_workflow(
378        owner=owner,
379        repo=repo,
380        workflow_file=workflow_file,
381        workflow_definition_ref=workflow_definition_ref,
382        inputs=parsed_inputs,
383    )
384
385    print_json(result.model_dump())
386
387    # If wait is enabled and we have a run_id, poll for completion
388    if wait and result.run_id:
389        print(f"\nWaiting for workflow to complete (timeout: {wait_seconds}s)...")
390        start_time = time.time()
391        poll_interval = 10  # seconds
392
393        while time.time() - start_time < wait_seconds:
394            status_result = check_ci_workflow_status(
395                owner=owner,
396                repo=repo,
397                run_id=result.run_id,
398            )
399
400            if status_result.status == "completed":
401                print(
402                    f"\nWorkflow completed with conclusion: {status_result.conclusion}"
403                )
404                print_json(status_result.model_dump())
405                return
406
407            elapsed = int(time.time() - start_time)
408            print(f"  Status: {status_result.status} (elapsed: {elapsed}s)")
409            time.sleep(poll_interval)
410
411        print(f"\nTimeout reached after {wait_seconds}s. Workflow still running.")
412        # Print final status
413        final_status = check_ci_workflow_status(
414            owner=owner,
415            repo=repo,
416            run_id=result.run_id,
417        )
418        print_json(final_status.model_dump())