"""Top-level commands for working with repos.
This module contains high level functions for administrating repositories, such
as creating student repos from some master repo template. All functions follow
the conventions specified in :ref:`conventions`.
Each public function in this module is to be treated as a self-contained
program.
.. module:: repos
:synopsis: Top-level commands for working with repos.
.. moduleauthor:: Simon Larsén
"""
import itertools
import pathlib
import re
import os
import sys
import tempfile
from typing import Iterable, List, Optional, Mapping, Union, Tuple
import repobee_plug as plug
import _repobee.command.teams
from _repobee import git
from _repobee import util
from _repobee import exception
from _repobee import config
from _repobee import plugin
from _repobee.git import Push
from _repobee.command import progresswrappers
[docs]def setup_student_repos(
template_repo_urls: Iterable[str],
teams: Iterable[plug.StudentTeam],
api: plug.PlatformAPI,
) -> Mapping[str, List[plug.Result]]:
"""Setup student repositories based on master repo templates. Performs three
primary tasks:
1. Create the specified teams on the target platform and add the
specified members to their teams. If a team already exists, it is left
as-is. If a student is already in a team they are assigned to, nothing
happens. If no account exists for some specified username, that
particular student is ignored, but any associated teams are still
created (even if a missing user is the only member of that team).
2. For each master repository, create one student repo per team and add
it to the corresponding student team. If a repository already exists,
it is skipped.
3. Push files from the master repos to the corresponding student repos.
Args:
template_repo_urls: URLs to master repos.
teams: An iterable of student teams specifying the teams to be setup.
api: An implementation of :py:class:`repobee_plug.PlatformAPI` used to
interface with the platform (e.g. GitHub or GitLab) instance.
"""
teams = list(teams)
with tempfile.TemporaryDirectory() as tmpdir:
workdir = pathlib.Path(tmpdir)
template_repos = [
plug.TemplateRepo(
name=util.repo_name(url),
url=url,
_path=workdir / api.extract_repo_name(url),
)
for url in template_repo_urls
]
plug.log.info("Cloning into master repos ...")
_clone_all(template_repos, cwd=tmpdir, api=api)
hook_results = plugin.execute_setup_tasks(
template_repos, api, cwd=pathlib.Path(tmpdir)
)
created_teams = list(
plug.cli.io.progress_bar(
_repobee.command.teams.create_teams(
teams, plug.TeamPermission.PUSH, api
),
total=len(teams),
desc="Creating student teams",
)
)
push_tuple_iter = _create_push_tuples(
created_teams, template_repos, api
)
push_tuple_iter_progress = plug.cli.io.progress_bar(
push_tuple_iter,
desc="Setting up student repos",
total=len(teams) * len(template_repos),
)
git.push(push_tuples=push_tuple_iter_progress)
return hook_results
def _create_push_tuples(
teams: Iterable[plug.Team],
template_repos: Iterable[plug.TemplateRepo],
api: plug.PlatformAPI,
) -> Iterable[Push]:
"""Create push tuples for newly created repos. Repos that already exist are
ignored.
Args:
teams: An iterable of teams.
template_repos: Template repositories.
api: A platform API instance.
Returns:
A list of Push namedtuples for all student repo urls that relate to
any of the master repo urls.
"""
for team, template_repo in itertools.product(teams, template_repos):
repo_name = plug.generate_repo_name(team, template_repo.name)
created, repo = _create_or_fetch_repo(
name=repo_name,
description=f"{repo_name} created for {team.name}",
private=True,
team=team,
api=api,
)
if created:
yield git.Push(
local_path=template_repo.path,
repo_url=api.insert_auth(repo.url),
branch=git.active_branch(template_repo.path),
)
def _create_or_fetch_repo(
name: str,
description: str,
private: bool,
api: plug.PlatformAPI,
team: Optional[plug.Team] = None,
) -> Tuple[bool, plug.Repo]:
try:
return (
True,
api.create_repo(
name, description=description, private=private, team=team
),
)
except plug.PlatformError:
team_name = team.name if team else None
repo = api.get_repo(repo_name=name, team_name=team_name)
if team:
api.assign_repo(
team=team, repo=repo, permission=plug.TeamPermission.PUSH
)
return False, repo
def _clone_all(
repos: plug.TemplateRepo, cwd: pathlib.Path, api: plug.PlatformAPI
):
"""Attempts to clone all repos sequentially.
Args:
repos: Repos to clone.
cwd: Working directory. Use temporary directory for automatic cleanup.
api: An instance of the platform API.
"""
try:
for repo in plug.cli.io.progress_bar(
repos, desc="Cloning template repositories", unit="repos"
):
url = _try_insert_auth(repo, api)
plug.log.info(f"Cloning into '{url}'")
git.clone_single(url, cwd=str(cwd))
except exception.CloneFailedError:
plug.log.error(f"Error cloning into {url}, aborting ...")
raise
def _try_insert_auth(
repo: Union[plug.TemplateRepo, plug.StudentRepo], api: plug.PlatformAPI
) -> str:
"""Try to insert authentication into the URL."""
try:
return api.insert_auth(repo.url)
except plug.InvalidURL:
return repo.url
[docs]def update_student_repos(
template_repo_urls: Iterable[str],
teams: Iterable[plug.StudentTeam],
api: plug.PlatformAPI,
issue: Optional[plug.Issue] = None,
) -> Mapping[str, List[plug.Result]]:
"""Attempt to update all student repos related to one of the master repos.
Args:
template_repo_urls: URLs to master repos. Must be in the organization
that the api is set up for.
teams: An iterable of student teams.
api: An implementation of :py:class:`repobee_plug.PlatformAPI` used to
interface with the platform (e.g. GitHub or GitLab) instance.
issue: An optional issue to open in repos to which pushing fails.
"""
if len(set(template_repo_urls)) != len(template_repo_urls):
raise ValueError("template_repo_urls contains duplicates")
with tempfile.TemporaryDirectory() as tmpdir:
workdir = pathlib.Path(tmpdir)
template_repos = [
plug.TemplateRepo(
name=util.repo_name(url),
url=url,
_path=workdir / api.extract_repo_name(url),
)
for url in template_repo_urls
]
plug.log.info("Cloning into master repos ...")
_clone_all(template_repos, cwd=tmpdir, api=api)
hook_results = plugin.execute_setup_tasks(
template_repos, api, cwd=pathlib.Path(tmpdir)
)
push_tuple_iter = _create_update_push_tuples(
teams, template_repos, api
)
push_tuple_iter_progress = plug.cli.io.progress_bar(
push_tuple_iter,
desc="Setting up student repos",
total=len(teams) * len(template_repos),
)
failed_urls = git.push(push_tuples=push_tuple_iter_progress)
if failed_urls and issue:
plug.echo("Opening issue in repos to which push failed")
urls_without_auth = [
re.sub("https://.*?@", "https://", url) for url in failed_urls
]
_open_issue_by_urls(urls_without_auth, issue, api)
plug.log.info("Done!")
return hook_results
def _create_update_push_tuples(
teams: Iterable[plug.StudentTeam],
template_repos: Iterable[plug.TemplateRepo],
api: plug.PlatformAPI,
) -> Iterable[Push]:
"""Create push tuples for existing repos. Repos that don't exist are
ignored.
Args:
teams: An iterable of teams.
template_repos: Template repositories.
api: A platform API instance.
Returns:
A list of Push namedtuples for all student repo urls that relate to
any of the master repo urls.
"""
urls_to_templates = {}
for team, template_repo in itertools.product(teams, template_repos):
repo_url, *_ = api.get_repo_urls(
[template_repo.name], team_names=[team.name]
)
urls_to_templates[repo_url] = template_repo
for repo in api.get_repos(urls_to_templates.keys()):
template = urls_to_templates[repo.url]
branch = git.active_branch(template.path)
yield git.Push(template.path, api.insert_auth(repo.url), branch)
def _open_issue_by_urls(
repo_urls: Iterable[str], issue: plug.Issue, api: plug.PlatformAPI
) -> None:
"""Open issues in the repos designated by the repo_urls.
Args:
repo_urls: URLs to repos in which to open an issue.
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.
"""
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)
plug.log.info(msg)
[docs]def clone_repos(
repos: Iterable[plug.StudentRepo], api: plug.PlatformAPI
) -> Mapping[str, List[plug.Result]]:
"""Clone all student repos related to the provided master repos and student
teams.
Args:
repos: The repos to be cloned. This function does not use the
``implementation`` attribute, so it does not need to be set.
api: An implementation of :py:class:`repobee_plug.PlatformAPI` used to
interface with the platform (e.g. GitHub or GitLab) instance.
Returns:
A mapping from repo name to a list of hook results.
"""
plug.echo("Cloning into student repos ...")
with tempfile.TemporaryDirectory() as tmpdir:
local_repos = _clone_repos_no_check(repos, pathlib.Path(tmpdir), api)
for p in plug.manager.get_plugins():
if "post_clone" in dir(p):
local_repos_progress = plug.cli.io.progress_bar(
local_repos, desc="Executing post_clone hooks"
)
return plugin.execute_clone_tasks(local_repos_progress, api)
return {}
def _clone_repos_no_check(
repos: Iterable[plug.StudentRepo],
dst_dirpath: pathlib.Path,
api: plug.PlatformAPI,
) -> Iterable[plug.StudentRepo]:
"""Clone the specified repo urls into the destination directory without
making any sanity checks; they must be done in advance.
Return a list of cloned and previously existing repos.
"""
cur_dir = pathlib.Path(".").resolve()
pathed_repos = [
repo.with_path(cur_dir / repo.team.name / repo.name) for repo in repos
]
cloned_repos = git.clone_student_repos(pathed_repos, dst_dirpath, api)
return [repo for _, repo in cloned_repos]
[docs]def migrate_repos(
template_repo_urls: Iterable[str], api: plug.PlatformAPI
) -> None:
"""Migrate a repository from an arbitrary URL to the target organization.
The new repository is added to the master_repos team, which is created if
it does not already exist.
Args:
template_repo_urls: Local urls to repos to migrate.
api: An implementation of :py:class:`repobee_plug.PlatformAPI` used to
interface with the platform (e.g. GitHub or GitLab) instance.
"""
local_templates = [
plug.TemplateRepo(name=util.repo_name(url), url=url)
for url in template_repo_urls
]
create_repo_it = plug.cli.io.progress_bar(
(
_create_or_fetch_repo(
local.name, description="", private=True, api=api
)
for local in local_templates
),
desc="Creating remote repos",
total=len(template_repo_urls),
)
with tempfile.TemporaryDirectory() as tmpdir:
workdir = pathlib.Path(tmpdir)
_clone_all(local_templates, cwd=workdir, api=api)
remote_templates = [
plug.TemplateRepo(
name=repo.name, url=repo.url, _path=workdir / repo.name
)
for _, repo in create_repo_it
]
git.push(
[
git.Push(
local_path=template_repo.path,
repo_url=api.insert_auth(template_repo.url),
branch=git.active_branch(template_repo.path),
)
for template_repo in remote_templates
]
)
plug.echo("Done!")
[docs]def show_config(config_file: pathlib.Path, show_secrets: bool) -> None:
"""Echo the config file.
Args:
config_file: The config file to echo.
show_secrets: Whether or not to show configured secrets.
"""
config.check_config_integrity(config_file)
plug.echo(f"Found valid config file at {config_file}")
with config_file.open(encoding=sys.getdefaultencoding()) as f:
config_contents = "".join(f.readlines())
output = (
os.linesep
+ "BEGIN CONFIG FILE".center(50, "-")
+ os.linesep
+ config_contents
+ "END CONFIG FILE".center(50, "-")
)
sanitized_output = re.sub(
r"\s*token\s*=\s*.*", "token = xxxxxxxxxx", output
)
plug.echo(output if show_secrets else sanitized_output)