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 __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 _do_write has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    def _do_write(self, out_f: BinaryIO) -> Tuple[bool, str]:
        all_metric_component_names = set()
        # capture the list of all metric component paths
        for col in self._columns.values():
            all_metric_component_names.update(col.get_metric_component_paths())

• Read up
  Read up

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

Open

    def _do_match(cls, dtype_or_type: Any, maybe_type: Optional[Any]) -> bool:
        if maybe_type:
            dtype_or_type = maybe_type  # type: ignore

        if not isinstance(dtype_or_type, type):

• Read up
  Read up

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

Open

    def get_component_paths(self) -> List[str]:
        res = []
        for sub_name, metrics in self.submetrics.items():
            for namespace, metric in metrics.items():
                for comp_name in metric.get_component_paths():

• Read up
  Read up

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

Open

    def submetrics_from_protobuf(cls: Type[COMPOUND_METRIC], msg: MetricMessage) -> Dict[str, Metric]:
        submetrics: Dict[str, Metric] = {}
        submetric_msgs: Dict[str, Dict[str, MetricComponentMessage]] = {}
        for key, comp_msg in msg.metric_components.items():
            submetric_name, comp_name = key.split("/")

• 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,
        types: Optional[Dict[str, Any]] = None,
        default_configs: Optional[MetricConfig] = None,
        type_mapper: Optional[TypeMapper] = None,

• Read up
  Read up

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

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 _log_segment has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def _log_segment(
    partition: SegmentationPartition,
    schema: DatasetSchema,
    obj: Any = None,
    pandas: Optional[pd.DataFrame] = None,

• 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 _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 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 __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

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

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)

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 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