Source code for _repobee.command.issues

"""Top-level commands for managing issues.

This module contains the top-level functions for RepoBee's issue management
functionality. Each public function in this module is to be treated as a
self-contained program.

.. module:: issues
    :synopsis: Top-level commands for issue management.

.. moduleauthor:: Simon Larsén
"""
import os
import re
from typing import Iterable, Optional, List, Tuple, Any, Mapping

from colored import bg, fg, style  # type: ignore

import repobee_plug as plug

from _repobee.command import progresswrappers


[docs]def list_issues( repos: Iterable[plug.StudentRepo], api: plug.PlatformAPI, state: plug.IssueState = plug.IssueState.OPEN, title_regex: str = "", show_body: bool = False, author: Optional[str] = None, ) -> Mapping[str, List[plug.Result]]: """List all issues in the specified repos. Args: repos: The repos from which to fetch issues. api: An implementation of :py:class:`repobee_plug.PlatformAPI` used to interface with the platform (e.g. GitHub or GitLab) instance. state: state of the repo (open or closed). Defaults to open. title_regex: If specified, only issues with titles matching the regex are displayed. Defaults to the empty string (which matches everything). show_body: If True, the body of the issue is displayed along with the default info. author: Only show issues by this author. """ # TODO optimize by not getting all repos at once repos = list(repos) repo_names = [repo.name for repo in repos] max_repo_name_length = max(map(len, repo_names)) platform_repos = api.get_repos([repo.url for repo in repos]) issues_per_repo = _get_issue_generator( platform_repos, title_regex=title_regex, author=author, state=state, api=api, ) # _log_repo_issues exhausts the issues_per_repo iterator and # returns a list with the same information. It's important to # have issues_per_repo as an iterator as it greatly speeds # up visual feedback to the user when fetching many issues pers_issues_per_repo = _log_repo_issues( issues_per_repo, show_body, max_repo_name_length + 6 ) # for writing to JSON hook_result_mapping = { repo_name: [ plug.Result( name="list-issues", status=plug.Status.SUCCESS, msg="Fetched {} issues from {}".format(len(issues), repo_name), data={issue.number: issue.to_dict() for issue in issues}, ) ] for repo_name, issues in pers_issues_per_repo } # meta hook result hook_result_mapping["list-issues"] = [ plug.Result( name="meta", status=plug.Status.SUCCESS, msg="Meta info about the list-issues hook results", data={"state": state.value}, ) ] return hook_result_mapping
def _get_issue_generator( repos: Iterable[plug.Repo], title_regex: str, author: Optional[str], state: plug.IssueState, api: plug.PlatformAPI, ) -> Iterable[Tuple[str, Iterable[plug.Issue]]]: issues_per_repo = ( ( repo.name, [ issue for issue in api.get_repo_issues(repo) if re.match(title_regex, issue.title) and (state == plug.IssueState.ALL or state == issue.state) and (not author or issue.author == author) ], ) for repo in repos ) return issues_per_repo def _log_repo_issues( issues_per_repo: Iterable[Tuple[str, Iterable[plug.Issue]]], show_body: bool, title_alignment: int, ) -> List[Tuple[Any, list]]: """Log repo issues. Args: issues_per_repo: (repo_name, issue generator) pairs show_body: Include the body of the issue in the output. title_alignment: Where the issue title should start counting from the start of the line. """ even = True persistent_issues_per_repo = [] for repo_name, issues in issues_per_repo: issues = list(issues) persistent_issues_per_repo.append((repo_name, issues)) if not issues: plug.log.warning("{}: No matching issues".format(repo_name)) for issue in issues: color = (bg("grey_30") if even else bg("grey_15")) + fg("white") even = not even # cycle color adjusted_alignment = title_alignment + len( color ) # color takes character space id_ = "{}{}/#{}:".format(color, repo_name, issue.number).ljust( adjusted_alignment ) out = "{}{}{}{}created {!s} by {}".format( id_, issue.title, style.RESET, " ", issue.created_at, issue.author, ) if show_body: out += os.linesep * 2 + _limit_line_length(issue.body) plug.echo(out) return persistent_issues_per_repo def _limit_line_length(s: str, max_line_length: int = 100) -> str: """Return the input string with lines no longer than max_line_length. Args: s: Any string. max_line_length: Maximum allowed line length. Returns: the input string with lines no longer than max_line_length. """ lines = s.split(os.linesep) out = "" for line in lines: cur = 0 while len(line) - cur > max_line_length: # find ws closest to the line length idx = line.rfind(" ", cur, max_line_length + cur) idx = max_line_length + cur if idx <= 0 else idx if line[idx] == " ": out += line[cur:idx] else: out += line[cur : idx + 1] out += os.linesep cur = idx + 1 out += line[cur : cur + max_line_length] + os.linesep return out
[docs]def open_issue( issue: plug.Issue, assignment_names: Iterable[str], teams: Iterable[plug.StudentTeam], api: plug.PlatformAPI, ) -> None: """Open an issue in student repos. Args: assignment_names: Names of assignments. teams: Team objects specifying student groups. issue: An issue to open. api: An implementation of :py:class:`repobee_plug.PlatformAPI` used to interface with the platform (e.g. GitHub or GitLab) instance. """ repo_urls = api.get_repo_urls( team_names=[t.name for t in teams], assignment_names=assignment_names ) repos = progresswrappers.get_repos(repo_urls, api) for repo in repos: issue = api.create_issue(issue.title, issue.body, repo) msg = f"Opened issue {repo.name}/#{issue.number}-'{issue.title}'" repos.write(msg) # type: ignore plug.log.info(msg)
[docs]def close_issue( title_regex: str, repos: Iterable[plug.StudentRepo], api: plug.PlatformAPI ) -> None: """Close issues whose titles match the title_regex in student repos. Args: title_regex: A regex to match against issue titles. assignment_names: Names of assignments. teams: Team objects specifying student groups. api: An implementation of :py:class:`repobee_plug.PlatformAPI` used to interface with the platform (e.g. GitHub or GitLab) instance. """ repo_urls = (repo.url for repo in repos) platform_repos = progresswrappers.get_repos(repo_urls, api) for repo in platform_repos: to_close = [ issue for issue in api.get_repo_issues(repo) if re.match(title_regex, issue.title) and issue.state == plug.IssueState.OPEN ] for issue in to_close: api.close_issue(issue) msg = f"Closed {repo.name}/#{issue.number}='{issue.title}'" platform_repos.write(msg) # type: ignore plug.log.info(msg)