forms.py

# -*- coding: utf-8 -*-
# pylint: disable=C,R,W
"""Contains the logic to create cohesive forms on the explore view"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

from flask_appbuilder.fieldwidgets import BS3TextFieldWidget
from flask_appbuilder.forms import DynamicForm
from flask_babel import lazy_gettext as _
from flask_wtf.file import FileAllowed, FileField, FileRequired
from wtforms import (
    BooleanField, Field, IntegerField, SelectField, StringField)
from wtforms.ext.sqlalchemy.fields import QuerySelectField
from wtforms.validators import DataRequired, NumberRange, Optional

from superset import app, db
from superset.models import core as models

config = app.config


class CommaSeparatedListField(Field):
    widget = BS3TextFieldWidget()

    def _value(self):
        if self.data:
            return u', '.join(self.data)
        else:
            return u''

    def process_formdata(self, valuelist):
        if valuelist:
            self.data = [x.strip() for x in valuelist[0].split(',')]
        else:
            self.data = []


def filter_not_empty_values(value):
    """Returns a list of non empty values or None"""
    if not value:
        return None
    data = [x for x in value if x]
    if not data:
        return None
    return data


class CsvToDatabaseForm(DynamicForm):
    # pylint: disable=E0211
    def all_db_items():
        return db.session.query(models.Database)

    name = StringField(
        _('Table Name'),
        description=_('Name of table to be created from csv data.'),
        validators=[DataRequired()],
        widget=BS3TextFieldWidget())
    csv_file = FileField(
        _('CSV File'),
        description=_('Select a CSV file to be uploaded to a database.'),
        validators=[
            FileRequired(), FileAllowed(['csv'], _('CSV Files Only!'))])
    con = QuerySelectField(
        _('Database'),
        query_factory=all_db_items,
        get_pk=lambda a: a.id, get_label=lambda a: a.database_name)
    sep = StringField(
        _('Delimiter'),
        description=_('Delimiter used by CSV file (for whitespace use \s+).'),
        validators=[DataRequired()],
        widget=BS3TextFieldWidget())
    if_exists = SelectField(
        _('Table Exists'),
        description=_(
            'If table exists do one of the following: '
            'Fail (do nothing), Replace (drop and recreate table) '
            'or Append (insert data).'),
        choices=[
            ('fail', _('Fail')), ('replace', _('Replace')),
            ('append', _('Append'))],
        validators=[DataRequired()])
    schema = StringField(
        _('Schema'),
        description=_('Specify a schema (if database flavour supports this).'),
        validators=[Optional()],
        widget=BS3TextFieldWidget(),
        filters=[lambda x: x or None])
    header = IntegerField(
        _('Header Row'),
        description=_(
            'Row containing the headers to use as '
            'column names (0 is first line of data). '
            'Leave empty if there is no header row.'),
        validators=[Optional()],
        widget=BS3TextFieldWidget(),
        filters=[lambda x: x or None])
    index_col = IntegerField(
        _('Index Column'),
        description=_(
            'Column to use as the row labels of the '
            'dataframe. Leave empty if no index column.'),
        validators=[Optional(), NumberRange(0, 1E+20)],
        widget=BS3TextFieldWidget(),
        filters=[lambda x: x or None])
    mangle_dupe_cols = BooleanField(
        _('Mangle Duplicate Columns'),
        description=_('Specify duplicate columns as "X.0, X.1".'))
    skipinitialspace = BooleanField(
        _('Skip Initial Space'),
        description=_('Skip spaces after delimiter.'))
    skiprows = IntegerField(
        _('Skip Rows'),
        description=_('Number of rows to skip at start of file.'),
        validators=[Optional(), NumberRange(0, 1E+20)],
        widget=BS3TextFieldWidget(),
        filters=[lambda x: x or None])
    nrows = IntegerField(
        _('Rows to Read'),
        description=_('Number of rows of file to read.'),
        validators=[Optional(), NumberRange(0, 1E+20)],
        widget=BS3TextFieldWidget(),
        filters=[lambda x: x or None])
    skip_blank_lines = BooleanField(
        _('Skip Blank Lines'),
        description=_(
            'Skip blank lines rather than interpreting them '
            'as NaN values.'))
    parse_dates = CommaSeparatedListField(
        _('Parse Dates'),
        description=_(
            'A comma separated list of columns that should be '
            'parsed as dates.'),
        filters=[filter_not_empty_values])
    infer_datetime_format = BooleanField(
        _('Infer Datetime Format'),
        description=_(
            'Use Pandas to interpret the datetime format '
            'automatically.'))
    decimal = StringField(
        _('Decimal Character'),
        description=_('Character to interpret as decimal point.'),
        validators=[Optional()],
        widget=BS3TextFieldWidget(),
        filters=[lambda x: x or '.'])
    index = BooleanField(
        _('Dataframe Index'),
        description=_('Write dataframe index as a column.'))
    index_label = StringField(
        _('Column Label(s)'),
        description=_(
            'Column label for index column(s). If None is given '
            'and Dataframe Index is True, Index Names are used.'),
        validators=[Optional()],
        widget=BS3TextFieldWidget(),
        filters=[lambda x: x or None])