util.py

import logging
import operator
import os
import time

from datetime import datetime
from functools import reduce
from pathlib import Path
from typing import Tuple, Union

import git

logger = logging.getLogger(__file__)


def remove_newline(string: str) -> str:
    return string.replace("\n", " ").strip()


def read_file_line_wise(path):
    """Linewise generator of a file """
    with open(path, "r") as file_to_read:
        for line in file_to_read.readlines():
            yield line.strip("\n")


def get_repo(repo_path: Union[Path, str] = ".") -> git.Repo:
    return git.Repo(repo_path)


def get_current_head(repo: git.Repo) -> git.Commit:
    return repo.head.commit


def get_commit_infos(commit) -> Tuple[str, str]:
    return commit.hexsha, commit.message


def get_head_info(repo_path: Path):
    return get_commit_infos(get_current_head(get_repo(repo_path)))


def get_submodule(submodule_name: str, repo: git.Repo) -> git.Submodule:
    return repo.submodule(submodule_name)


def get_submodule_path(submodule_name: str, repo_path: Union[Path, str] = ".") -> Path:
    submodule_path = get_submodule(submodule_name, get_repo(repo_path)).path
    return Path(repo_path, submodule_path)


def get_git_infos(repo_path: Union[Path, str], *, commit_key="commit", commit_msg_key="commit_message"):
    commit, commit_msg = get_commit_infos(get_current_head(get_repo(repo_path)))
    commit_msg = remove_newline(commit_msg)
    return {commit_key: commit, commit_msg_key: commit_msg}


def get_git_infos_from_env():
    """ Extract commit, commit message, branch and project id from gitlab ci environment."""
    return {"commit": os.environ["CI_COMMIT_SHA"],
            "commit_message": os.environ["CI_COMMIT_MESSAGE"],
            "project_id": os.environ["CI_PROJECT_ID"],
            "branch": os.environ["CI_COMMIT_BRANCH"],
            }


def file_time_to_sec(file_path) -> int:
    return int(os.path.getmtime(file_path))


def time_conversion(time_stamp, *, pattern="%Y-%m-%d %H:%M:%S"):
    try:
        return int(time_stamp)
    except ValueError as e:
        logger.debug(e)
    try:
        dt = datetime.strptime(time_stamp, pattern)
    except ValueError as e:
        logger.debug(f"strptime failed for {time_stamp} with: \n {e} trying with iso format")
        dt = datetime.fromisoformat(time_stamp)

    return int(dt.timestamp())


def get_time_stamp_from_env(key='CI_PIPELINE_CREATED_AT', *, fallback=lambda: int(time.time())) -> int:
    """ Try to get the time stamp from the environment and
        convert it to an int.
        If that fails fallback is called.
        The default fallback is current time.
    """
    try:
        return time_conversion(os.environ[key])
    except KeyError:
        logger.debug(f'Could not find timestamp with ${key}')
        return fallback()


def get_from_nested_dict(nested_dict, keys):
    return reduce(operator.getitem, keys, nested_dict)