File zenodo.py has 293 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import calendar
import datetime as dt
import json
import os
import re

File typer_commands.py has 256 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import datetime as dt
import re
from dataclasses import asdict
from itertools import chain
from pathlib import Path

Function _parse_module has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

    def _parse_module(self, module):
        try:
            base_id = CompleteIdBase.from_module_name(module.__name__, self.name)
        except NotADzObject:
            return

• Read up
  Read up

Function _get_ns_mappers has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def _get_ns_mappers(self, data_only=True):
        f_args = (self.runtime, self.sql_meta, self.engine, self._batch_size)
        _mapped = set()
        for ns in self.runtime.metadata.namespaces.values():
            _mapped.add((self.runtime.name, ns.name))

• Read up
  Read up

Function git_run has 9 arguments (exceeds 4 allowed). Consider refactoring.

Open

def git_run(

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

Open

def _ds_cls_to_feat_dicts(ds_cls: Union[EntityClass, CompositeType]):
    feature_dict = get_feature_dict(ds_cls)
    ids = []
    props = []
    for k, cls in feature_dict.items():

• Read up
  Read up

Function __init__ has 27 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    def __init__(
        self,
        entity: type[AbstractEntity],
        entity_key_table_map: Optional[dict[str, "ScruTable"]] = None,
        partitioning_cols: Optional[list[str]] = None,

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

Open

    def add_obj(self, obj):
        for cls, l in [
            (ScruTable, self.tables),
            (SourceUrl, self.source_urls),
            (PipelineElement, self.pipeline_elements),

• Read up
  Read up

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

Open

def add_stage(cmd, name, outs_no_cache, outs, outs_persist, deps, params):

Function deposit_to_zenodo has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

def deposit_to_zenodo(

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

Open

    def get_data_envs(self, project, ns):
        if project == self.name:
            return [e.name for e in self.envs]
        for iaf in self.imported_projects:
            if iaf.name == project:

• Read up
  Read up

Function register has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

def register(

Function __call__ has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def __call__(self, df, parse=True, verbose=True, env=None, **kwargs):

Function import_dvc has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

def import_dvc(uri, path, out, rev=None, no_exec=False):

Function validate has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

def validate(

Function sql_validation has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

def sql_validation(constr, env, draw=False, batch_size=2000, verbose=False):

Function __init__ has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def __init__(self, title, version, lines, private: bool) -> None:

Function __init__ has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def __init__(self, conf: Config, private: bool, tag: str, test: bool = True):

Function feat_to_cols has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

def feat_to_cols(feat, proc_fk, wrap, init_prefix=(), open_to_fk=True) -> list:

Function _get_persistent_state_dependencies has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def _get_persistent_state_dependencies(self, conf: Config):
        for dep in self.dependencies:
            if isinstance(dep, type) and (PersistentState in dep.mro()):
                base = [CONF_KEYS.persistent_states, dep.get_full_name()]
                for k in conf.persistent_states.get(base[-1], {}).keys():

• Read up
  Read up