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"
[docs]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")
[docs]def generate_lookML_model() -> None:
"""
Generate lookML codes based on compiled dbt project.
"""
subprocess_run(["dbt2looker", "--output-dir", str(LOOKML_DEST_PATH)])
[docs]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,
)