Source code for _repobee.util

"""Some general utility functions.

.. module:: util
    :synopsis: Miscellaneous utility functions that don't really belong
        anywhere else.

.. moduleauthor:: Simon Larsén
"""
import os
import sys
import pathlib
import shutil
import tempfile
from typing import Iterable, Generator, Union, Callable, TypeVar

import repobee_plug as plug

T = TypeVar("T")


[docs]def read_issue(issue_path: str) -> plug.Issue: """Attempt to read an issue from a textfile. The first line of the file is interpreted as the issue's title. Args: issue_path: Local path to textfile with an issue. """ if not os.path.isfile(issue_path): raise ValueError("{} is not a file".format(issue_path)) with open(issue_path, "r", encoding=sys.getdefaultencoding()) as file: return plug.Issue(file.readline().strip(), file.read())
[docs]def repo_name(repo_url: str) -> str: """Extract the name of the repo from its url. Args: repo_url: A url to a repo. """ repo_name = repo_url.split("/")[-1] if repo_name.endswith(".git"): return repo_name[:-4] return repo_name
[docs]def is_git_repo(path: str) -> bool: """Check if a directory has a .git subdirectory. Args: path: Path to a local directory. Returns: True if there is a .git subdirectory in the given directory. """ return os.path.isdir(path) and ".git" in os.listdir(path)
def _ends_with_ext( path: Union[str, pathlib.Path], extensions: Iterable[str] ) -> bool: _, ext = os.path.splitext(str(path)) return ext in extensions
[docs]def find_files_by_extension( root: Union[str, pathlib.Path], *extensions: str ) -> Generator[pathlib.Path, None, None]: """Find all files with the given file extensions, starting from root. Args: root: The directory to start searching. extensions: One or more file extensions to look for. Returns: a generator that yields a Path objects to the files. """ if not extensions: raise ValueError("must provide at least one extension") for cwd, _, files in os.walk(root): for file in files: if _ends_with_ext(file, extensions): yield pathlib.Path(cwd) / file
[docs]def atomic_write(content: str, dst: pathlib.Path) -> None: """Write the given contents to the destination "atomically". Achieved by writin in a temporary directory and then moving the file to the destination. Args: content: The content to write to the new file. dst: Path to the file. """ with tempfile.TemporaryDirectory() as tmpdir: with tempfile.NamedTemporaryFile( delete=False, dir=tmpdir, mode="w" ) as file: file.write(content) shutil.move(file.name, str(dst))
[docs]def call_if_defined(func: Callable[..., T], *args, **kwargs) -> T: """Call the function with the provided args and kwargs if it is defined (i.e. not None). This is mostly useful for plugin data structures that have optional functions. Args: func: A function to call. args: Positional arguments. kwargs: Keyword arguments. Returns: What ``func`` returns, or ``None`` if ``func`` is ``None``. """ return None if func is None else func(*args, **kwargs)