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

Open

    def run(self) -> int:
        self.extraction_results = set()

        extract_destination = self.get_config_value("path")
        self.api_key = self.get_config_value("api_key")

• Read up
  Read up

Function _dbt_schema_table_in_connector has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def _dbt_schema_table_in_connector(self, connector_schemas, dbt_schema, dbt_table):
        for schema_details in connector_schemas.values():
            if schema_details.get("name_in_destination", "").lower() == dbt_schema:
                for table_details in schema_details.get("tables", {}).values():
                    if table_details.get("name_in_destination", "").lower() == dbt_table:

• Read up
  Read up

Function dbt_deps has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def dbt_deps(self, config_folder=None):
        if not config_folder:
            config_folder = self.get_config_folder(mandatory=False)

        if config_folder:

• Read up
  Read up

Function _get_connection_schema has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def _get_connection_schema(self, conn, destination_config):
        """Given an airybte connection, returns a schema name"""
        namespace_definition = conn["namespaceDefinition"]

        if namespace_definition == "source" or (

• Read up
  Read up

Function replace_env_vars has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def replace_env_vars(self, yaml_dict: Dict) -> Dict:
        # regex -> {{ env_var('ENV_VAR_NAME', 'DEFAULT_VALUE') }}
        env_var_pattern = re.compile(
            r"\{\{\s*env_var\s*\(\s*['\"]?\s*([^'\"]+)['\"]?\s*(?:,\s*['\"]?\s*([^'\"]+)['\"]?\s*)?\)\s*\}\}"
        )

• Read up
  Read up

Function _load_fivetran_datacoves_secrets has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def _load_fivetran_datacoves_secrets(self, fivetran_object):
        for secret in self.secret_manager_data:
            for destination_data in fivetran_object.values():
                object_details = destination_data["details"]
                if object_details["id"] == secret.get("slug", ""):

• Read up
  Read up

Function make_query has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

    def make_query(self) -> SelectAny:
        table = self.table
        query = table.select()
        if not self.incremental:
            return query

• Read up
  Read up

Function modify_property_file has 8 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def modify_property_file(

Function sql_database has 8 arguments (exceeds 4 allowed). Consider refactoring.

Open

def sql_database(

Function _update_or_create_fivetran_destination has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    def _update_or_create_fivetran_destination(self, exported_destination, target_group_id):
        """
        Given exported destination data
        - update if ID exists
        - create a new one if doesn't

• Read up
  Read up

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

Open

    def run(self):
        self.loading_results = {
            "sources": {"updated": [], "created": []},
            "destinations": {"updated": [], "created": []},
            "connections": {"updated": [], "created": []},

• Read up
  Read up

Function get_metadata has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    def get_metadata(self):
        """
        If metadata path is configured, returns a dictionary with column keys
        and their corresponding values.
        If metadata is already set, do not load again and return the existing value.

• Read up
  Read up

Function table_rows has 7 arguments (exceeds 4 allowed). Consider refactoring.

Open

def table_rows(

Function render_property_files has 7 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def render_property_files(

Avoid deeply nested control flow statements.

Open

                        for k, v in secret_data.items():
                            self._replace_dict_key(object_details, k, v)

Function main has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def main(parser: argparse.ArgumentParser = parser, test_cli_args: List[str] = list()) -> int:
    tracking.do_not_track()

    exit_code = 0
    cli_args = test_cli_args or []

• Read up
  Read up

Function integrated has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def integrated(self):
        """
        Returns the values read from the config file plus the overrides from cli flags
        """
        config_copy = self._config.dict()

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if (airbyte_schema == schema or not airbyte_schema) and conn.get(
                            "connectionId"
                        ) not in connection_ids:
                            connection_ids.append(conn["connectionId"])
        if connection_ids:

Function list_from_dbt_ls has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def list_from_dbt_ls(self, output):
        user_selectors = list()
        if self.get_config_value("select"):
            user_selectors.append("--select")
            user_selectors.append(self.get_config_value("select"))

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        for schema_id, schema_data in connector_data.get("schemas", {}).items():
                            if (
                                self._dbt_schema_table_in_connector(
                                    {schema_id: schema_data},
                                    source_schema.lower(),