diff --git a/src/fidesctl/ctl/core/utils.py b/src/fidesctl/ctl/core/utils.py index 961691aadd..adff8835b3 100644 --- a/src/fidesctl/ctl/core/utils.py +++ b/src/fidesctl/ctl/core/utils.py @@ -5,6 +5,7 @@ from functools import partial from json.decoder import JSONDecodeError from typing import Dict, Iterator, List +from os.path import isfile import click import jwt @@ -12,7 +13,6 @@ import sqlalchemy from fideslang.models import DatasetField, FidesModel from fideslang.validation import FidesValidationError -from git.repo import Repo from sqlalchemy.engine import Engine from sqlalchemy.exc import SQLAlchemyError @@ -56,7 +56,9 @@ def get_db_engine(connection_string: str) -> Engine: Use SQLAlchemy to create a DB engine. """ try: - engine = sqlalchemy.create_engine(connection_string) + engine = sqlalchemy.create_engine( + connection_string, connect_args={"connect_timeout": 10} + ) except Exception as err: raise Exception("Failed to create engine!") from err @@ -82,7 +84,7 @@ def generate_request_headers(user_id: str, api_key: str) -> Dict[str, str]: return { "Content-Type": "application/json", "user-id": str(user_id), - "Authorization": "Bearer {}".format(jwt_encode(1, api_key)), + "Authorization": f"Bearer {jwt_encode(1, api_key)}", } @@ -100,8 +102,11 @@ def get_all_level_fields(fields: list) -> Iterator[DatasetField]: def get_manifest_list(manifests_dir: str) -> List[str]: """Get a list of manifest files from the manifest directory.""" - yml_endings = ["yml", "yaml"] + + if isfile(manifests_dir) and manifests_dir.split(".")[-1] in yml_endings: + return [manifests_dir] + manifest_list = [] for yml_ending in yml_endings: manifest_list += glob.glob(f"{manifests_dir}/**/*.{yml_ending}", recursive=True) @@ -136,6 +141,19 @@ def git_is_dirty(dir_to_check: str = ".") -> bool: Checks to see if the local repo has unstaged changes. Can also specify a directory to check. """ + + try: + from git.repo import Repo + from git.repo.fun import is_git_dir + except ImportError: + print("Git executable not detected, skipping git check...") + return False + + git_dir_path = ".git/" + if not is_git_dir(git_dir_path): + print(f"No git repo detected at '{git_dir_path}', skipping git check...") + return False + repo = Repo() git_session = repo.git()