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

Open

    def get_sqla_engine(  # pylint: disable=too-many-arguments

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

Open

    def avatar(self, user_id: int) -> Response:
        """Get a redirect to the avatar's URL for the user with the given ID.
        ---
        get:
          summary: Get the user avatar

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if "having_druid" in extras:
                            del extras["having_druid"]
                            updated = True

Avoid deeply nested control flow statements.

Open

                        if key in form_data:
                            del form_data[key]
                            updated = True

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

Open

    def _process_sql_expression(  # pylint: disable=too-many-arguments

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

Open

    def dashboard(
        self,
        dashboard_id_or_slug: str,
        add_extra_log_payload: Callable[..., None] = lambda **kwargs: None,
    ) -> FlaskResponse:

• Read up
  Read up

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

Open

    def import_(self) -> Response:
        """Import dataset(s) with associated databases.
        ---
        post:
          summary: Import dataset(s) with associated databases

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if "sqlExpression" in filt:
                            filt["sqlExpression"] = filt["sqlExpression"].replace(
                                source, target
                            )

Avoid deeply nested control flow statements.

Open

                        if op == utils.FilterOperator.LIKE.value:
                            where_clause_and.append(sqla_col.like(eq))
                        else:
                            where_clause_and.append(sqla_col.ilike(eq))
                    elif op in {utils.FilterOperator.NOT_LIKE.value}:

Avoid deeply nested control flow statements.

Open

                    for filter_config in filter_configs:
                        if filter_config["column"] == col:
                            is_multiple = filter_config["multiple"]

                            if not is_multiple and isinstance(val, list):

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

Open

    def _get_sqla_engine(  # pylint: disable=too-many-locals

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

Open

def upgrade():
    bind = op.get_bind()
    session = db.Session(bind=bind)

    for slc in session.query(Slice).filter(Slice.viz_type == "pie").all():

• Read up
  Read up

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

Open

def upgrade():
    """
    Fix databases with ``schemas_allowed_for_csv_upload`` stored as string.
    """
    bind = op.get_bind()

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if target_generic_type != GenericDataType.STRING:
                            sqla_col = sa.cast(sqla_col, sa.String)

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

Open

    def eval(self) -> datetime:
        dttm_expression, unit = self.value
        dttm = dttm_expression.eval()
        if unit == "year":
            dttm = dttm.replace(

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                    if modifier and (user := owner_dict.get(modifier.id)):
                        return executor_type, user.username
                if creator := model.created_by:

Avoid deeply nested control flow statements.

Open

                    if creator and (user := owner_dict.get(creator.id)):
                        return executor_type, user.username
                return executor_type, owners[0].username

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

Open

    def set_or_update_query_limit(self, new_limit: int, force: bool = False) -> str:
        """Returns the query with the specified limit.

        Does not change the underlying query if user did not apply the limit,
        otherwise replaces the limit with the lower value between existing limit

• Read up
  Read up

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

Open

    def _check_functions_exist_in_token(
        self, token: Token, functions: set[str]
    ) -> bool:
        if (
            isinstance(token, Function)

• Read up
  Read up

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

Open

def resample(
    df: pd.DataFrame,
    rule: str,
    method: str,
    fill_value: Optional[Union[float, int]] = None,

• Read up
  Read up