Function parse_args has a Cognitive Complexity of 296 (exceeds 5 allowed). Consider refactoring.

Open

    def parse_args(self, cli_args: List[str] = list()) -> None:
        args = sys.argv[1:]

        # TODO: FIXME: This is a temporary logic for when --dbt-args is followed by a single-word string,
        # like cases of --dbt-args "--no-compile",

• Read up
  Read up

File airbyte.py has 514 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import json
import os
import pathlib
from copy import copy

File base.py has 417 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import csv
import fnmatch
from pathlib import Path

import questionary

Function render_property_files has a Cognitive Complexity of 35 (exceeds 5 allowed). Consider refactoring.

Open

    def render_property_files(
        self,
        context,
        options,
        templates_folder,

• Read up
  Read up

File airflow_dags.py has 383 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import datetime
import importlib
import textwrap
from glob import glob
from pathlib import Path

Function discover_dbt_connections has a Cognitive Complexity of 34 (exceeds 5 allowed). Consider refactoring.

Open

    def discover_dbt_connections(self) -> Set[str]:
        """
        Discover DBT source(s)' Airbyte/Fivetran connection IDs based on params
        """
        if self.virtualenv_path:

• Read up
  Read up

File flags.py has 358 lines of code (exceeds 250 allowed). Consider refactoring.

Open

"""Flags module containing the DbtCovesFlags "Factory"."""

import os
import sys
from argparse import ArgumentParser

File sources.py has 358 lines of code (exceeds 250 allowed). Consider refactoring.

Open

from __future__ import nested_scopes

import json
from pathlib import Path

File fivetran.py has 346 lines of code (exceeds 250 allowed). Consider refactoring.

Open

from pathlib import Path

import questionary
from rich.console import Console

Function run_git_clone has a Cognitive Complexity of 28 (exceeds 5 allowed). Consider refactoring.

Open

    def run_git_clone(self, workspace_path):
        repo_url = ""
        cloned_status = "[red]MISSING[/red]"
        cloned_exists = Path(workspace_path, ".git").exists()
        if cloned_exists:

• Read up
  Read up

Function get_pipeline_connection_ids has a Cognitive Complexity of 25 (exceeds 5 allowed). Consider refactoring.

Open

    def get_pipeline_connection_ids(
        self, source_db: str, source_schema: str, source_table: str
    ) -> str:
        """
        Given a table name, schema and db, returns the corresponding Fivetran Connection ID

• Read up
  Read up

Function get_relations has a Cognitive Complexity of 24 (exceeds 5 allowed). Consider refactoring.

Open

    def get_relations(self, filtered_schemas):
        rel_name_selectors = [relation for relation in self.get_config_value("select_relations")]

        rel_excludes = [relation for relation in self.get_config_value("exclude_relations")]

• Read up
  Read up

Function run has a Cognitive Complexity of 22 (exceeds 5 allowed). Consider refactoring.

Open

    def run(self) -> int:
        self.load_results = {
            "groups": {"created": set()},
            "destinations": {"created": set(), "updated": set()},
            "connectors": {"created": set(), "updated": set()},

• Read up
  Read up

Function generate has a Cognitive Complexity of 22 (exceeds 5 allowed). Consider refactoring.

Open

    def generate(self, rels):
        self.flatten_json = self.get_config_value("flatten_json_fields")
        self.overwrite_sqls = self.get_config_value("overwrite_staging_models")
        models_destination = self.get_config_value("models_destination")
        options = {

• Read up
  Read up

File config.py has 294 lines of code (exceeds 250 allowed). Consider refactoring.

Open

"""Holds config for dbt-coves."""

import os
import re
from pathlib import Path

Function generate has a Cognitive Complexity of 21 (exceeds 5 allowed). Consider refactoring.

Open

    def generate(self, rels):
        destination = self.get_config_value("destination")
        options = {"append_all": False, "recreate_files": False}

        for rel in rels:

• Read up
  Read up

Function run_git_config has a Cognitive Complexity of 20 (exceeds 5 allowed). Consider refactoring.

Open

    def run_git_config(self):
        config_status = "[red]MISSING[/red]"

        email_output = run_and_capture(["git", "config", "--global", "--get", "user.email"])
        email_exists = email_output.returncode == 0 and email_output.stdout

• Read up
  Read up

Function get_schemas has a Cognitive Complexity of 20 (exceeds 5 allowed). Consider refactoring.

Open

    def get_schemas(self):
        # get schema names selectors
        schema_name_selectors = [schema for schema in self.get_config_value("schemas")]

        schema_wildcard_selectors = []

• Read up
  Read up

Function get_pipeline_connection_ids has a Cognitive Complexity of 19 (exceeds 5 allowed). Consider refactoring.

Open

    def get_pipeline_connection_ids(self, db: str, schema: str, table: str) -> str:
        """
        Given a table name, schema and db, returns the corresponding Airbyte Connection ID
        """
        airbyte_tables = []

• Read up
  Read up

Function run has a Cognitive Complexity of 18 (exceeds 5 allowed). Consider refactoring.

Open

    def run(self):
        options = {"overwrite_all": False}
        dbtcoves_templates_path = Path(dbt_coves.__file__).parent / "templates"

        dbt_project_path = self.config.project_root

• Read up
  Read up