def prepare_xy()

in svinfer/processor/commons.py [0:0]

• 28 lines of code
• 8 McCabe index (conditional complexity)

    def prepare_xy(self, x_columns, y_column, fit_intercept=True):
        x_columns = [sqlalchemy.Column(j, sqlalchemy.FLOAT) for j in x_columns]
        y_column = sqlalchemy.Column(y_column, sqlalchemy.FLOAT)
        filters_columns = {}
        if self.filters is not None:
            for k in self.filters:
                filters_columns[k] = sqlalchemy.Column(k, sqlalchemy.String)

        sqlalchemy.Table(
            self.table_name,
            sqlalchemy.MetaData(schema=self.database),
            y_column,
            *x_columns,
            *filters_columns.values(),
        )

        need = [y_column.label("y")]
        if fit_intercept:
            need.append(sqlalchemy.literal(1.0).label("x0"))
        for i in range(len(x_columns)):
            need.append(x_columns[i].label("x" + str(i + 1)))

        where_clauses = (
            [filters_columns[k].in_(self.filters[k]) for k in filters_columns.keys()]
            if self.filters is not None
            else []
        )

        work = sqlalchemy.select(need).where(sqlalchemy.and_(*where_clauses))
        y_part = list(work.columns)[:1]
        x_part = list(work.columns)[1:]
        return SqlMatrix(x_part), SqlMatrix(y_part)