todonotifier.connect

  1"""This module provides interface for connecting to the respective repository to allow
  2cloning the repository to be able to run TODO Notifier.
  3"""
  4
  5import logging
  6import os
  7from enum import Enum
  8from shutil import copy, copytree, ignore_patterns
  9from typing import TypeVar, Union
 10
 11from git.repo import Repo
 12
 13from todonotifier.constants import DEFAULT_EXCLUDE_DIRS
 14
 15P = TypeVar("P")
 16
 17# logging configuration
 18logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(process)d - %(name)s - %(levelname)s - %(message)s")
 19logger = logging.getLogger(__name__)
 20
 21
 22class ConnectException(Exception):
 23    """Raised if any exception in `connect` module"""
 24
 25    pass
 26
 27
 28class CONNECT_METHOD(Enum):
 29    """Enum values defining the connection method to pull a repository"""
 30
 31    GIT_CLONE = "GIT_CLONE"
 32    DRY_RUN_FILE = "DRY_RUN_FILE"
 33    DRY_RUN_DIR = "DRY_RUN_DIR"
 34
 35
 36class Connect:
 37    """Provides a common interface to pull repositories from different sources"""
 38
 39    def __init__(self, connect_method: CONNECT_METHOD, project_dir_name: str, url: str, branch_name: Union[str, None] = None) -> None:
 40        """Initializer for `Connect` class
 41
 42        Args:
 43            connect_method (CONNECT_METHOD): Method to be used to pull a repository into the target location
 44            project_dir_name (str): Name of the project. Should match the name of project main directory
 45            url (str): Url or file address or directory address that needs to be pulled
 46            branch_name (optional, Union[str, None]): Branch name is specific branch to be checked out
 47                                                    after cloning the repository. Useful for `CONNECT_METHOD.GIT_CLONE`. Defaults to None.
 48        """
 49        self._connect_method = connect_method
 50        self._project_dir_name = project_dir_name
 51        self._file_dir_url = url
 52        self._branch_name = branch_name
 53
 54    @property
 55    def project_dir_name(self) -> str:
 56        """Getter for project directory name
 57
 58        Returns:
 59            str: Returns value of `self._project_dir_name
 60        """
 61        return self._project_dir_name
 62
 63    def __str__(self) -> str:
 64        """Returns the string representation of the class `Connect`
 65
 66        Returns:
 67            str: Returns string representation of the class `Connect`
 68        """
 69        return f"{repr(self)} connect_method: {self._connect_method} project_dir_name: {self._project_dir_name} url: {self._file_dir_url}"
 70
 71    def pull_repository(self, target_dir: str, branch_name: Union[str, None] = None) -> P:
 72        """Provides a one point access to pull repository into the target location. This method simply relegates
 73        the call to respective methods set during class initialization
 74
 75        Args:
 76            target_dir (str): Directory into which the data from given `url` needs to be copied into
 77            branch_name (optional, Union[str, None]): Branch name is specific branch to be checked out
 78                                                    after cloning the repository. Useful for `CONNECT_METHOD.GIT_CLONE`. Defaults to None.
 79
 80        Returns:
 81            P: Returns whatever is returned by the respective method to which call is delegated to
 82        """
 83        try:
 84            logger.info(f"Pulling repository: {self._project_dir_name} via {self._connect_method}")
 85            if self._connect_method == CONNECT_METHOD.GIT_CLONE:
 86                return self._pull_using_git_clone(target_dir, branch_name=self._branch_name)
 87            elif self._connect_method == CONNECT_METHOD.DRY_RUN_FILE:
 88                return self._pull_file_for_dry_run(target_dir)
 89            elif self._connect_method == CONNECT_METHOD.DRY_RUN_DIR:
 90                return self._pull_dir_for_dry_run(target_dir)
 91            else:
 92                raise ConnectException("Unsupported connect method passed")
 93        except Exception:
 94            logger.exception(f"Error in pulling repository via {self._connect_method}")
 95            raise ConnectException(f"Error in pulling repository via {self._connect_method}")
 96
 97    def _pull_using_git_clone(self, target_dir: str, branch_name: Union[str, None] = None) -> Repo:
 98        """Pulls the repository using GIT_CLONE method
 99
100        NOTE: This method used GitPython library that required Git to be installed on the system
101
102        Args:
103            target_dir (str): Directory into which the data from given `url` needs to be copied into
104            branch_name (optional, Union[str, None]): Branch name is specific branch to be checked out
105                                                    after cloning the repository. Defaults to None.
106
107            Returns:
108                Repo: Returns handle to the repository cloned
109        """
110        return Repo.clone_from(self._file_dir_url, target_dir, branch=branch_name)
111
112    def _pull_file_for_dry_run(self, target_dir: str) -> None:
113        """Copies the local file `test_file` into `target_dir` directory
114
115        Args:
116            target_dir (str): Directory into which the file from given `url` needs to be copied into
117        """
118        target_dir = os.path.join(target_dir, self._project_dir_name)
119        if not os.path.isdir(target_dir):
120            os.mkdir(target_dir)
121
122        copy(self._file_dir_url, target_dir)
123
124    def _pull_dir_for_dry_run(self, target_dir: str) -> None:
125        """Copies the local file `test_file` into `target_dir` directory. it automatically ignores the directories in `constants.DEFAULT_EXCLUDE_DIRS`
126
127        Args:
128            target_dir (str): Directory into which the folder from given `url` needs to be copied into
129        """
130        test_dir_base_name = os.path.basename(self._file_dir_url)
131        ignore_list = DEFAULT_EXCLUDE_DIRS["PATTERN"] + DEFAULT_EXCLUDE_DIRS["NAME"]
132        target_dir = os.path.join(target_dir, test_dir_base_name)
133        copytree(self._file_dir_url, target_dir, ignore=ignore_patterns(*ignore_list))