Source code for gerrit_to_platform.github

# SPDX-License-Identifier: Apache-2.0
##############################################################################
# Copyright (c) 2023 The Linux Foundation and others.
#
# All rights reserved. This program and the accompanying materials are made
# available under the terms of the Apache-2.0 license which accompanies this
# distribution, and is available at
# https://opensource.org/licenses/Apache-2.0
##############################################################################
"""Github connection module."""

from typing import Any, Dict, List

from fastcore.net import HTTP404NotFoundError  # type: ignore
from ghapi.all import GhApi  # type: ignore

from gerrit_to_platform.config import get_setting


[docs]def dispatch_workflow( owner: str, repository: str, workflow_id: str, ref: str, inputs: Dict[str, str] ) -> Any: """ Trigger target workflow on GitHub. Args: owner (str): GitHub owner (user or organization) repository (str): target repository workflow_id (str): ID of the workflow to trigger ref (str): the commit ref to trigger with inputs (Dict[str, str]): dictionary of key / value pairs to pass as inputs to the workflow """ github_token = get_setting("github.com", "token") api = GhApi(token=github_token) return api.actions.create_workflow_dispatch( owner, repository, workflow_id, ref, inputs )
[docs]def filter_path(search_filter: str, workflow: Dict[str, str]) -> bool: """ Case insensitive path filter for use in lambda filters. Args: search_filter (str): string to search workflow file names for workflow (Dict[str, str]): dictionary containing all data related to the workflow being evaluated Returns: bool: True if search_filter matches in workflow["path"], False otherwise """ path = workflow["path"].lower() if path.find(search_filter.lower()) >= 0: return True return False
[docs]def filter_workflows( owner: str, repository: str, search_filter: str, search_required: bool = False ) -> List[Dict[str, str]]: """ Return a case insensitive filtered list of workflows. All workflows must meet the basic filtering of containing the search_filter in the name as well as "gerrit" in the name. If search_required is true, then the list will contiain all workflows that contain "required" in the name If search_required is false (the default), then the list will _exclude_ all workflows that contain "required" in the name. Args: owner (str): GitHub owner (entity or organization) repository (str): target repository search_filter (str): the substring to search for in workflow filenames search_required (bool): if workflows with "required" in the filename are to be returned. If false, then required workflows will be filtered out, if true only required workflows will be returned. Returns: List[Dict[str, str]]: list of dictionaries containing all workflows that meet the search criteria of having "gerrit" in the name, the search_filter substring, and either required or not required according to the search_required argument. """ workflows = get_workflows(owner, repository) filtered_workflows: List[Dict[str, str]] = [] filtered_workflows = list( filter(lambda workflow: filter_path(search_filter, workflow), workflows) ) filtered_workflows = list( filter(lambda workflow: filter_path("gerrit", workflow), filtered_workflows) ) if search_required: filtered_workflows = list( filter( lambda workflow: filter_path("required", workflow), filtered_workflows ) ) else: filtered_workflows = list( filter( lambda workflow: not filter_path("required", workflow), filtered_workflows, ) ) return filtered_workflows
[docs]def get_workflows(owner: str, repository: str) -> List[Dict[str, str]]: """ Get all active workflows for specific repository. Args: owner (str): GitHub owner (entity or organization) repository (str): target repository Returns: List[Dict[str, str]]: list of dictionaries containing data related to active workflows in the target repository. """ github_token = get_setting("github.com", "token") api = GhApi(token=github_token) try: workflows = api.actions.list_repo_workflows(owner, repository)["workflows"] except HTTP404NotFoundError: return [] workflows = [workflow for workflow in workflows if workflow["state"] == "active"] key_ids = [ "node_id", "created_at", "updated_at", "url", "html_url", "badge_url", "state", ] for workflow in workflows: for del_key in key_ids: del workflow[del_key] return workflows