Source code for data_pipelines_cli.looker_utils

import glob
import os
import pathlib
from shutil import copy, rmtree
from typing import Any, Dict, Tuple

import requests
import yaml
from git import Repo

from .cli_constants import BUILD_DIR
from .cli_utils import echo_info, subprocess_run
from .config_generation import (
    generate_profiles_yml,
    read_dictionary_from_config_directory,
)
from .dbt_utils import run_dbt_command

LOOKML_DEST_PATH: pathlib.Path = BUILD_DIR.joinpath("lookml")
LOOKML_VIEWS_SUBDIR: str = "views"


def read_looker_config(env: str) -> Dict[str, Any]:
    """
    Read Looker configuration.

    :param env: Name of the environment
    :type env: str
    :return: Compiled dictionary
    :rtype: Dict[str, Any]
    """
    return read_dictionary_from_config_directory(BUILD_DIR.joinpath("dag"), env, "looker.yml")


def generate_lookML_model() -> None:
    """
    Generate lookML codes based on compiled dbt project.
    """
    subprocess_run(["dbt2looker", "--output-dir", str(LOOKML_DEST_PATH)])


def deploy_lookML_model(key_path: str, env: str) -> None:
    """
    Write compiled lookML to Looker's repository and deploy project to production

    :param key_path: Path to the key with write access to git repository
    :type key_path: str
    :param env: Name of the environment
    :type env: str
    """
    profiles_path = generate_profiles_yml(env, False)
    run_dbt_command(("docs", "generate"), env, profiles_path)

    looker_config = read_looker_config(env)
    local_repo_path = BUILD_DIR.joinpath("looker_project_repo")

    if local_repo_path.exists():
        echo_info(f"Removing {local_repo_path}")
        rmtree(local_repo_path)

    ssh_command_with_key = f"ssh -i {key_path}"
    repo = Repo.clone_from(
        looker_config["looker_repository"],
        local_repo_path,
        branch=looker_config["looker_repository_branch"],
        env={"GIT_SSH_COMMAND": ssh_command_with_key},
    )

    project_name, project_version = _get_project_name_and_version()
    with repo.git.custom_environment(GIT_SSH_COMMAND=ssh_command_with_key):
        _prepare_repo_changes(LOOKML_DEST_PATH, local_repo_path)
        _configure_git_env(repo, looker_config)
        _commit_and_push_changes(repo, project_name, project_version)
        _deploy_looker_project_to_production(
            looker_config["looker_instance_url"],
            looker_config["looker_project_id"],
            looker_config["looker_repository_branch"],
            looker_config["looker_webhook_secret"],
        )


def _prepare_repo_changes(src: pathlib.Path, local_repo_gen_path: pathlib.Path) -> None:
    _clear_repo_before_writing_lookml(local_repo_gen_path)
    _copy_all_files_by_extention(src, local_repo_gen_path, "model.lkml")
    _copy_all_files_by_extention(
        src, local_repo_gen_path.joinpath(LOOKML_VIEWS_SUBDIR), "view.lkml"
    )

    with open(f"{local_repo_gen_path}/readme.txt", "w") as readme:
        readme.write(
            """models and views with extention '.dp.[view|model].lkml' are generated by data-pipelines-cli.
            Do not edit manually! Your changes could be overwrite!
        """
        )


def _clear_repo_before_writing_lookml(local_repo_gen_path: pathlib.Path) -> None:
    if local_repo_gen_path.exists():
        _remove_dp_files_from_repo(local_repo_gen_path, ".dp.model.lkml")

        if local_repo_gen_path.joinpath(LOOKML_VIEWS_SUBDIR).exists():
            _remove_dp_files_from_repo(
                local_repo_gen_path.joinpath(LOOKML_VIEWS_SUBDIR), ".dp.view.lkml"
            )


def _remove_dp_files_from_repo(dir_path: pathlib.Path, files_extention: str) -> None:
    for file in os.listdir(dir_path):
        if file.endswith(files_extention):
            os.remove(dir_path.joinpath(file))


def _configure_git_env(repo: Repo, config: Dict[str, Any]) -> None:
    repo.config_writer().set_value("user", "name", config["looker_repository_username"]).release()
    repo.config_writer().set_value("user", "email", config["looker_repository_email"]).release()


def _copy_all_files_by_extention(
    src: pathlib.Path, dest: pathlib.Path, files_extention: str
) -> None:
    os.makedirs(dest, exist_ok=True)
    for file_path in glob.glob(os.path.join(src, "**", "*." + files_extention), recursive=True):
        file_path_with_dp_ext = "{0}.dp.{1}.{2}".format(*file_path.rsplit(".", 2))
        new_path = os.path.join(dest, os.path.basename(file_path_with_dp_ext))
        copy(file_path, new_path)


def _get_project_name_and_version() -> Tuple[str, str]:
    with open(pathlib.Path.cwd().joinpath("dbt_project.yml"), "r") as f:
        dbt_project_config = yaml.safe_load(f)
        return dbt_project_config["name"], dbt_project_config["version"]


def _commit_and_push_changes(repo: Repo, project_name: str, project_version: str) -> None:
    echo_info("Publishing BI codes to Looker repository")
    repo.git.add(all=True)
    repo.index.commit(f"Publication from project {project_name}, version: {project_version}")
    origin = repo.remote(name="origin")
    origin.push()


def _deploy_looker_project_to_production(
    looker_instance_url: str, project_id: str, branch: str, webhook_secret: str
) -> None:
    echo_info("Deploying Looker project to production")
    headers = {"X-Looker-Deploy-Secret": webhook_secret}
    requests.post(
        url=f"{looker_instance_url}/webhooks/projects/{project_id}/deploy/branch/{branch}",
        headers=headers,
    )