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

Open

    def __init__(
        self,
        op: Relation = Relation.no_op,
        value: Union[str, int, float, ValueGetter] = 0,
        udf: Optional[Callable[[Any], bool]] = None,

• Read up
  Read up

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

Open

    def __post_init__(self):
        self.range_definitions["UNKNOWN"] = (0, 0)  # catchall for characters not in a defined range
        for key, range in self.range_definitions.items():
            if range[0] > range[1]:
                raise ValueError(f"Invalid codepoint range {key}")

• Read up
  Read up

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

Open

def initialize_logger() -> None:
    # Initialize session
    n_attempts = 3
    while n_attempts > 0:
        # Initialize logger

• Read up
  Read up

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

Open

    def _get_time_tuple(self) -> time.struct_time:
        if self.utc:
            time_tuple = time.gmtime(self._current_batch_timestamp)
        else:
            time_tuple = time.localtime(self._current_batch_timestamp)

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                    if _is_nan(val):
                        evaluations.append((pandas[col].isna()))
                    else:
                        evaluations.append((pandas[col] == val))
                mask = reduce(lambda x, y: x & y, evaluations)

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

Open

def _segmented_performance_metrics(
    log_full_data: bool,
    schema: DatasetSchema,
    data: pd.DataFrame,
    performance_column_mapping: Dict[str, Optional[str]],

• Read up
  Read up

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

Open

    def calc_non_numeric_relevance(self, row_dict: pd.core.series.Series) -> Tuple[List[int], List[int]]:
        prediction_relevance = []
        ideal_relevance = []
        for target_val in row_dict[self.prediction_column]:
            ideal_relevance.append(1 if target_val in row_dict[self.target_column] else 0)

• Read up
  Read up

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

Open

    def write(
        self,
        file: Writable,
        dest: Optional[str] = None,
        **kwargs: Any,

• Read up
  Read up

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

Open

    def __init__(
        self,
        aggregate_by: TimeGranularity = TimeGranularity.Hour,
        write_schedule: Optional[Schedule] = Schedule(cadence=TimeGranularity.Minute, interval=10),
        schema: Optional[DatasetSchema] = None,

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                    if ref_column_summary["drift_from_ref"]:
                        ref_column_summary["drift_from_ref"].update(
                            {"primary_value": col_drift_value["pvalue"] or col_drift_value["statistic"]}
                        )
            target_dist = target_col_view.get_metric("distribution")

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

Open

    def __init__(
        self,
        svd_class: Optional[type] = None,  # TODO: maybe make this updatable: bool = False
        svd_config: Optional[SvdMetricConfig] = None,
        svd_state: Optional[MetricMessage] = None,

• Read up
  Read up

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

Open

    def _create_client(self, cache_config: ClientCacheConfig) -> Tuple[ApiClient, KeyRefresher]:
        """
        Refresh the API client by comparing various configs. We try to
        re-use the client as much as we can since using a new client
        every time can be expensive.

• Read up
  Read up

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

Open

    def log_debug_event(
        self,
        debug_event: Optional[Dict[str, Any]] = None,
        *,
        trace_id: str,

• Read up
  Read up

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

Open

    def _load_or_prompt(
        self,
        env_name: EnvVariableName,
        config_name: ConfigVariableName,
        persist: bool = False,

• Read up
  Read up

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

Open

    def __init__(
        self,
        types: Optional[Dict[str, Any]] = None,
        default_configs: Optional[MetricConfig] = None,
        type_mapper: Optional[TypeMapper] = None,

• Read up
  Read up

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

Open

    def write(
        self,
        file: Writable,
        dest: Optional[str] = None,  # TODO: this should be used as object_name
        **kwargs: Any,

• Read up
  Read up

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

Open

    def columnar_update(self, data: PreprocessedColumn) -> OperationResult:
        # Should be data.list.objs  [ List[str] ] from scalar
        #           data.pandas.obj Series[List[str]] from apply
        doc_lengths = list()
        if data.list.objs and isinstance(data.list.objs[0], list) and _all_strings(data.list.objs[0]):

• Read up
  Read up

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

Open

    def option(self, **kwargs) -> "WhyLabsClient":  # type: ignore
        """

        Parameters
        ----------

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if spec.metric.get_namespace() in result:
                            logger.warning(
                                f"Conflicting resolvers for {spec.metric.get_namespace()} metric in column '{name}' of type {why_type.return_type.__name__}"
                            )
                        result[spec.metric.get_namespace()] = spec.metric.zero(cfg)

Avoid deeply nested control flow statements.

Open

                        if validate_with_row_id:
                            action(self.name, cond_name, x, identity_list[index])  # type: ignore
                        else:
                            action(self.name, cond_name, x)  # type: ignore