@task(default_retry_delay=5, ignore_result=True)
def calculate_updates(dataset, new_data, new_dframe_raw=None,
                      parent_dataset_id=None, update_id=None):
    """Update dataset with `new_data`.

    This can result in race-conditions when:

    - deleting ``controllers.Datasets.DELETE``
    - updating ``controllers.Datasets.POST([dataset_id])``

    Therefore, perform these actions asychronously.

    :param new_data: Data to update this dataset with.
    :param new_dframe_raw: DataFrame to update this dataset with.
    :param parent_dataset_id: If passed add ID as parent ID to column,
        default is None.
    """
    if not __update_is_valid(dataset, new_dframe_raw):
        dataset.remove_pending_update(update_id)
        return

    __ensure_ready(dataset, update_id)

    if new_dframe_raw is None:
        new_dframe_raw = dframe_from_update(dataset, new_data)

    new_dframe = recognize_dates(new_dframe_raw, dataset.schema)

    new_dframe = __add_calculations(dataset, new_dframe)

    # set parent id if provided
    if parent_dataset_id:
        new_dframe = add_parent_column(new_dframe, parent_dataset_id)

    dataset.append_observations(new_dframe)
    dataset.clear_summary_stats()

    propagate(dataset, new_dframe=new_dframe, update={'add': new_dframe_raw})

    dataset.update_complete(update_id) def dframe_from_update(dataset, new_data):
    """Make a DataFrame for the `new_data`.

    :param new_data: Data to add to dframe.
    :type new_data: List.
    """
    filtered_data = []
    columns = dataset.columns
    labels_to_slugs = dataset.schema.labels_to_slugs
    num_columns = len(columns)
    num_rows = dataset.num_rows
    dframe_empty = not num_columns

    if dframe_empty:
        columns = dataset.schema.keys()

    for row in new_data:
        filtered_row = dict()
        for col, val in row.iteritems():
            # special case for reserved keys (e.g. _id)
            if col == MONGO_ID:
                if (not num_columns or col in columns) and\
                        col not in filtered_row.keys():
                    filtered_row[col] = val
            else:
                # if col is a label take slug, if it's a slug take col
                slug = labels_to_slugs.get(
                    col, col if col in labels_to_slugs.values() else None)

                # if slug is valid or there is an empty dframe
                if (slug or col in labels_to_slugs.keys()) and (
                        dframe_empty or slug in columns):
                    filtered_row[slug] = dataset.schema.convert_type(
                        slug, val)

        filtered_data.append(filtered_row)

    index = range(num_rows, num_rows + len(filtered_data))
    new_dframe = DataFrame(filtered_data, index=index)

    return new_dframe @task(default_retry_delay=5, ignore_result=True)
def propagate(dataset, new_dframe=None, update=None):
    """Propagate changes in a modified dataset."""
    __update_aggregate_datasets(dataset, new_dframe, update=update)

    if update:
        __update_merged_datasets(dataset, update)
        __update_joined_datasets(dataset, update) BAMBOO_RESERVED_KEY_PREFIX = '^^' DATASET_ID = BAMBOO_RESERVED_KEY_PREFIX + 'dataset_id' INDEX = BAMBOO_RESERVED_KEY_PREFIX + 'index' def join_dataset(left, other, on):
    """Left join an `other` dataset.

    :param other: Other dataset to join.
    :param on: Column or 2 comma seperated columns to join on.

    :returns: Joined DataFrame.

    :raises: `KeyError` if join columns not in datasets.
    """
    on_lhs, on_rhs = (on.split(',') * 2)[:2]

    right_dframe = other.dframe(padded=True)

    if on_lhs not in left.columns:
        raise KeyError('No item named "%s" in left hand side dataset' % on_lhs)

    if on_rhs not in right_dframe.columns:
        raise KeyError('No item named "%s" in right hand side dataset' %
                       on_rhs)

    right_dframe = right_dframe.set_index(on_rhs)

    if len(right_dframe.index) != len(right_dframe.index.unique()):
        msg = 'Join column "%s" of the right hand side dataset is not unique'
        raise NonUniqueJoinError(msg % on_rhs)

    shared_columns = left.columns.intersection(right_dframe.columns)

    if len(shared_columns):
        rename_map = [{c: '%s.%s' % (c, v) for c in shared_columns} for v
                      in ['x', 'y']]
        left.rename(columns=rename_map[0], inplace=True)
        right_dframe.rename(columns=rename_map[1], inplace=True)

    return left.join(right_dframe, on=on_lhs) PARENT_DATASET_ID = BAMBOO_RESERVED_KEY_PREFIX + 'parent_dataset_id' def remove_reserved_keys(df, exclude=[]):
    """Remove reserved internal columns in this DataFrame.

    :param keep_parent_ids: Keep parent column if True, default False.
    """
    reserved_keys = __column_intersect(
        df, BAMBOO_RESERVED_KEYS).difference(set(exclude))

    return df.drop(reserved_keys, axis=1) def summarize(dataset, dframe, groups, no_cache, update=False):
    """Raises a ColumnTypeError if grouping on a non-dimensional column."""
    # do not allow group by numeric types
    for group in groups:
        if not dataset.is_factor(group):
            raise ColumnTypeError("group: '%s' is not a dimension." % group)

    group_str = dataset.join_groups(groups) or dataset.ALL

    # check cached stats for group and update as necessary
    stats = dataset.stats
    group_stats = stats.get(group_str)

    if no_cache or not group_stats or update:
        group_stats = summarize_with_groups(dframe, groups, dataset) if\
            groups else summarize_df(dframe, dataset)

        if not no_cache:
            if update:
                original_group_stats = stats.get(group_str, {})
                group_stats = combine_dicts(original_group_stats, group_stats)

            stats.update({group_str: group_stats})
            dataset.update({dataset.STATS: dict_for_mongo(stats)})

    stats_dict = dict_from_mongo(group_stats)

    if groups:
        stats_dict = {group_str: stats_dict}

    return stats_dict def call_async(function, *args, **kwargs):
    """Potentially asynchronously call `function` with the arguments.

    :param function: The function to call.
    :param args: Arguments for the function.
    :param kwargs: Keyword arguments for the function.
    """
    countdown = kwargs.pop('countdown', 0)

    if is_async():
        function.__getattribute__('apply_async')(
            countdown=countdown, args=args, kwargs=kwargs)
    else:  # pragma: no cover
        function(*args, **kwargs) class ArgumentError(Exception):
    pass def df_mongo_decode(df, keep_mongo_keys=False):
    """Decode MongoDB reserved keys in this DataFrame."""
    rename_dict = {}

    if MONGO_ID in df.columns:
        if keep_mongo_keys:
            df.rename(columns={MONGO_ID: MONGO_ID_ENCODED,
                               MONGO_ID_ENCODED: MONGO_ID}, inplace=True)
        else:
            del df[MONGO_ID]
            if MONGO_ID_ENCODED in df.columns:
                rename_dict[MONGO_ID_ENCODED] = MONGO_ID

    if rename_dict:
        df.rename(columns={MONGO_ID_ENCODED: MONGO_ID}, inplace=True)

    return df class ImportableDataset(object):
    def import_from_url(self, url, na_values=[], allow_local_file=False):
        """Load a URL, read from a CSV, add data to dataset.

        :param url: URL to load file from.
        :param allow_local_file: Allow URL to refer to a local file.

        :raises: `IOError` for an unreadable file or a bad URL.

        :returns: The created dataset.
        """
        if not allow_local_file and isinstance(url, basestring)\
                and url[0:4] == 'file':
            raise IOError

        call_async(
            import_dataset, self, partial(
                csv_file_reader, url, na_values=na_values))

        return self

    def import_from_csv(self, csv_file, na_values=[]):
        """Import data from a CSV file.

        .. note::

            Write to a named tempfile in order  to get a handle for pandas'
            `read_csv` function.

        :param csv_file: The CSV File to create a dataset from.

        :returns: The created dataset.
        """
        if 'file' in dir(csv_file):
            csv_file = csv_file.file

        tmpfile = tempfile.NamedTemporaryFile(delete=False)
        tmpfile.write(csv_file.read())

        # pandas needs a closed file for *read_csv*
        tmpfile.close()

        call_async(import_dataset, self, partial(
            csv_file_reader, tmpfile.name, na_values=na_values, delete=True))

        return self

    def import_from_json(self, json_file):
        """Impor data from a JSON file.

        :param json_file: JSON file to import.
        """
        content = json_file.file.read()
        call_async(import_dataset, self, partial(json_file_reader, content))

        return self

    def import_schema(self, schema):
        """Create a dataset from a SDF schema file (JSON).

        :param schema: The SDF (JSON) file to create a dataset from.

        :returns: The created dataset.
        """
        try:
            schema = json.loads(schema.file.read())
        except AttributeError:
            schema = json.loads(schema)

        self.set_schema(filter_schema(schema))
        self.ready()

        return self class QueryArgs(object):
    def __init__(self, query=None, select=None, distinct=None, limit=0,
                 order_by=None, dataset=None):
        """A holder for query arguments.

        :param query: An optional query.
        :param select: An optional select to limit the fields in the dframe.
        :param distinct: Return distinct entries for this field.
        :param limit: Limit on the number of rows in the returned dframe.
        :param order_by: Sort resulting rows according to a column value and
            sign indicating ascending or descending.

        Example of `order_by`:

          - ``order_by='mycolumn'``
          - ``order_by='-mycolumn'``
        """
        self.query = parse_dates_from_query(query, dataset)
        self.select = select
        self.distinct = distinct
        self.limit = limit
        self.order_by = parse_order_by(order_by)

    def encode(self, encoding, query):
        """Encode query, order_by, and select given an encoding.

        The query will be combined with the existing query.

        :param encoding: A dict to encode the QueryArgs fields with.
        :param query: An additional dict to combine with the existing query.
        """
        self.query = replace_keys(combine_dicts(self.query, query), encoding)
        self.order_by = self.order_by and replace_keys(dict(self.order_by),
                                                       encoding).items()
        self.select = self.select and replace_keys(self.select, encoding)

    def __nonzero__(self):
        return bool(self.query or self.select or self.distinct or self.limit
                    or self.order_by) class Schema(dict):
    @classmethod
    def safe_init(cls, arg):
        """Make schema with potential arg of None."""
        return cls() if arg is None else cls(arg)

    @property
    def labels_to_slugs(self):
        """Build dict from column labels to slugs."""
        return {
            column_attrs[LABEL]: reserve_encoded(column_name) for
            (column_name, column_attrs) in self.items()
        }

    @property
    def numerics(self):
        return [slug for slug, col_schema in self.items()
                if col_schema[SIMPLETYPE] in [INTEGER, FLOAT]]

    @property
    def numerics_select(self):
        return {col: 1 for col in self.numerics}

    def cardinality(self, column):
        if self.is_dimension(column):
            return self[column].get(CARDINALITY)

    def convert_type(self, slug, value):
        column_schema = self.get(slug)
        if column_schema:
            type_func = SIMPLETYPE_TO_DTYPE.get(column_schema[SIMPLETYPE])
            if type_func:
                value = type_func(value)

        return value

    def datetimes(self, intersect_with):
        return [slug for slug, col in self.items()
                if col[SIMPLETYPE] == DATETIME and slug in intersect_with]

    def is_date_simpletype(self, column):
        return self[column][SIMPLETYPE] == DATETIME

    def is_dimension(self, column):
        col_schema = self.get(column)

        return col_schema and col_schema[OLAP_TYPE] == DIMENSION

    def rebuild(self, dframe, overwrite=False):
        """Rebuild a schema for a dframe.

        :param dframe: The DataFrame whose schema to merge with the current
            schema.
        :param overwrite: If true replace schema, otherwise update.
        """
        current_schema = self
        new_schema = schema_from_dframe(dframe, self)

        if current_schema and not overwrite:
            # merge new schema with existing schema
            current_schema.update(new_schema)
            new_schema = current_schema

        return new_schema

    def rename_map_for_dframe(self, dframe):
        """Return a map from dframe columns to slugs.

        :param dframe: The DataFrame to produce the map for.
        """
        labels_to_slugs = self.labels_to_slugs

        return {
            column: labels_to_slugs[column] for column in
            dframe.columns.tolist() if self._resluggable_column(
                column, labels_to_slugs, dframe)
        }

    def set_olap_type(self, column, olap_type):
        """Set the OLAP Type for this `column` of schema.

        Only columns with an original OLAP Type of 'measure' can be modified.
        This includes columns with Simple Type integer, float, and datetime.

        :param column: The column to set the OLAP Type for.
        :param olap_type: The OLAP Type to set. Must be 'dimension' or
            'measure'.
        :raises: `ArgumentError` if trying to set the OLAP Type of an column
          whose OLAP Type was not originally a 'measure'.
        """
        self[column][OLAP_TYPE] = olap_type

    def _resluggable_column(self, column, labels_to_slugs, dframe):
        """Test if column should be slugged.

        A column should be slugged if:
            1. The `column` is a key in `labels_to_slugs` and
            2. The `column` is not a value in `labels_to_slugs` or
                1. The `column` label is not equal to the `column` slug and
                2. The slug is not in the `dframe`'s columns

        :param column: The column to reslug.
        :param labels_to_slugs: The labels to slugs map (only build once).
        :param dframe: The DataFrame that column is in.
        """
        return (column in labels_to_slugs.keys() and (
                not column in labels_to_slugs.values() or (
                    labels_to_slugs[column] != column and
                    labels_to_slugs[column] not in dframe.columns))) def combine_dicts(*dicts):
    """Combine dicts with keys in later dicts taking precedence."""
    return dict(chain(*[_dict.iteritems() for _dict in dicts])) def to_list(maybe_list):
    return maybe_list if isinstance(maybe_list, list) else [maybe_list] class AbstractModel(object):
    """An abstact class for all MongoDB models.

    Attributes:

    - __collection__: The MongoDB collection to communicate with.
    - STATE: A key with which to store state.
    - STATE_PENDING: A value for the pending state.
    - STATE_READY: A value for the ready state.

    """

    __collection__ = None
    __collectionname__ = None

    DB_READ_BATCH_SIZE = 1000
    DB_SAVE_BATCH_SIZE = 2000
    ERROR_MESSAGE = 'error_message'
    GROUP_DELIMITER = ','  # delimiter when passing multiple groups as a string
    MIN_BATCH_SIZE = 50
    STATE = 'state'
    STATE_FAILED = 'failed'
    STATE_PENDING = 'pending'
    STATE_READY = 'ready'

    @property
    def state(self):
        return self.record[self.STATE]

    @property
    def error_message(self):
        return self.record.get(self.ERROR_MESSAGE)

    @property
    def is_pending(self):
        return self.state == self.STATE_PENDING

    @property
    def is_ready(self):
        return self.state == self.STATE_READY

    @property
    def record_ready(self):
        return self.record is not None and self.state == self.STATE_READY

    @property
    def clean_record(self):
        """Remove reserved keys from records."""
        _dict = {
            key: value for (key, value) in self.record.items() if not key in
            BAMBOO_RESERVED_KEYS
        }
        return remove_mongo_reserved_keys(_dict)

    @classmethod
    def set_collection(cls, collection_name):
        """Return a MongoDB collection for the passed name.

        :param collection_name: The name of collection to return.

        :returns: A MongoDB collection from the current database.
        """
        return Database.db()[collection_name]

    @classproperty
    @classmethod
    def collection(cls):
        """Set the internal collection to the class' collection name."""
        if not cls.__collection__:
            cls.__collection__ = AbstractModel.set_collection(
                cls.__collectionname__)

        return cls.__collection__

    @classmethod
    def create(cls, *args):
        model = cls()
        return model.save(*args)

    @classmethod
    def find(cls, query_args, as_dict=False, as_cursor=False):
        """An interface to MongoDB's find functionality.

        :param query_args: An optional QueryArgs to hold the query arguments.
        :param as_cursor: If True, return the cursor.
        :param as_dict: If True, return dicts and not model instances.

        :returns: A list of dicts or model instances for each row returned.
        """
        cursor = cls.collection.find(query_args.query,
                                     query_args.select,
                                     sort=query_args.order_by,
                                     limit=query_args.limit)

        if as_cursor:
            return cursor
        else:
            return [record for record in cursor] if as_dict else [
                cls(record) for record in cursor
            ]

    @classmethod
    def find_one(cls, query, select=None, as_dict=False):
        """Return the first row matching `query` and `select` from MongoDB.

        :param query: A query to pass to MongoDB.
        :param select: An optional select to pass to MongoDB.
        :param as_dict: If true, return dicts and not model instances.

        :returns: A model instance of the row returned for this query and
            select.
        """
        record = cls.collection.find_one(query, select)

        return record if as_dict else cls(record)

    @classmethod
    def unset(cls, query, unset_query):
        """Call unset with the spec `query` the unset document `unset_query`.

        :param query: The spec restrict updates to.
        :param unset_query: The query to pass to unset.
        """
        cls.collection.update(query, {"$unset": unset_query}, multi=True)

    def __init__(self, record=None):
        """Instantiate with data in `record`."""
        self.record = record

    def __nonzero__(self):
        return self.record is not None

    def failed(self, message=None):
        """Perist the state of the current instance to `STATE_FAILED`.

        :params message: A string store as the error message, default None.
        """
        doc = {self.STATE: self.STATE_FAILED}

        if message:
            doc.update({self.ERROR_MESSAGE: message})

        self.update(doc)

    def pending(self):
        """Perist the state of the current instance to `STATE_PENDING`"""
        self.update({self.STATE: self.STATE_PENDING})

    def ready(self):
        """Perist the state of the current instance to `STATE_READY`"""
        self.update({self.STATE: self.STATE_READY})

    def delete(self, query):
        """Delete rows matching query.

        :param query: The query for rows to delete.
        """
        self.collection.remove(query)

    def save(self, record):
        """Save `record` in this model's collection.

        Save the record in the model instance's collection and set the internal
        record of this instance to the passed in record.

        :param record: The dict to save in the model's collection.

        :returns: The record passed in.
        """
        self.collection.insert(record)
        self.record = record

        return self

    def update(self, record):
        """Update the current instance with `record`.

        Update the current model instance based on its `_id`, set it to the
        passed in `record`.

        :param record: The record to replace the instance's data with.
        """
        record = dict_for_mongo(record)
        id_dict = {'_id': self.record['_id']}
        self.collection.update(id_dict, {'$set': record})

        # Set record to the latest record from the database
        self.record = self.__class__.collection.find_one(id_dict)

    def split_groups(self, groups):
        """Split a string based on the group delimiter"""
        return groups.split(self.GROUP_DELIMITER) if groups else []

    def join_groups(self, groups):
        return self.GROUP_DELIMITER.join(groups) class Calculation(AbstractModel):

    __collectionname__ = 'calculations'

    AGGREGATION = 'aggregation'
    AGGREGATION_ID = 'aggregation_id'
    DEPENDENCIES = 'dependencies'
    DEPENDENT_CALCULATIONS = 'dependent_calculations'
    FORMULA = 'formula'
    GROUP = 'group'
    NAME = 'name'

    @property
    def aggregation(self):
        return self.record[self.AGGREGATION]

    @property
    def aggregation_id(self):
        return self.record.get(self.AGGREGATION_ID)

    @property
    def dataset_id(self):
        return self.record[DATASET_ID]

    @property
    def dependencies(self):
        return self.record.get(self.DEPENDENCIES, [])

    @property
    def dependent_calculations(self):
        return self.record.get(self.DEPENDENT_CALCULATIONS, [])

    @property
    def formula(self):
        return self.record[self.FORMULA]

    @property
    def group(self):
        return self.record[self.GROUP]

    @property
    def groups_as_list(self):
        return self.split_groups(self.group)

    @property
    def name(self):
        return self.record[self.NAME]

    @classmethod
    def create(cls, dataset, formula, name, group=None):
        calculation = super(cls, cls).create(dataset, formula, name, group)
        call_async(calculate_task, [calculation], dataset.clear_cache())

        return calculation

    @classmethod
    def create_from_list_or_dict(cls, dataset, calculations):
        calculations = to_list(calculations)

        if not len(calculations) or not isinstance(calculations, list) or\
                any([not isinstance(e, dict) for e in calculations]):
            raise ArgumentError('Improper format for JSON calculations.')

        parsed_calculations = []

        # Pull out args to check JSON format
        try:
            for c in calculations:
                groups = c.get("groups")

                if not isinstance(groups, list):
                    groups = [groups]

                for group in groups:
                    parsed_calculations.append(
                        [c[cls.FORMULA], c[cls.NAME], group])
        except KeyError as e:
            raise ArgumentError('Required key %s not found in JSON' % e)

        # Save instead of create so that we calculate on all at once.
        calculations = [cls().save(dataset, formula, name, group)
                        for formula, name, group in parsed_calculations]
        call_async(calculate_task, calculations, dataset.clear_cache())

    @classmethod
    def find(cls, dataset, include_aggs=True, only_aggs=False):
        """Return the calculations for`dataset`.

        :param dataset: The dataset to retrieve the calculations for.
        :param include_aggs: Include aggregations, default True.
        :param only_aggs: Exclude non-aggregations, default False.
        """
        query = {DATASET_ID: dataset.dataset_id}

        if not include_aggs:
            query[cls.AGGREGATION] = None

        if only_aggs:
            query[cls.AGGREGATION] = {'$ne': None}

        query_args = QueryArgs(query=query, order_by='name')
        return super(cls, cls).find(query_args)

    @classmethod
    def find_one(cls, dataset_id, name, group=None):
        query = {DATASET_ID: dataset_id, cls.NAME: name}

        if group:
            query[cls.GROUP] = group

        return super(cls, cls).find_one(query)

    def add_dependencies(self, dataset, dependent_columns):
        """Store calculation dependencies."""
        calculations = dataset.calculations()
        names_to_calcs = {calc.name: calc for calc in calculations}

        for column_name in dependent_columns:
            calc = names_to_calcs.get(column_name)
            if calc:
                self.add_dependency(calc.name)
                calc.add_dependent_calculation(self.name)

    def add_dependency(self, name):
        self.__add_and_update_set(self.DEPENDENCIES, self.dependencies, name)

    def add_dependent_calculation(self, name):
        self.__add_and_update_set(self.DEPENDENT_CALCULATIONS,
                                  self.dependent_calculations, name)

    def delete(self, dataset):
        """Delete this calculation.

        First ensure that there are no other calculations which depend on this
        one. If not, start a background task to delete the calculation.

        :param dataset: Dataset for this calculation.

        :raises: `DependencyError` if dependent calculations exist.
        :raises: `ArgumentError` if group is not in DataSet or calculation does
            not exist for DataSet.
        """
        if len(self.dependent_calculations):
            msg = 'Cannot delete, calculations %s depend on this calculation.'
            raise DependencyError(msg % self.dependent_calculations)

        if not self.group is None:
            # it is an aggregate calculation
            dataset = dataset.aggregated_dataset(self.group)

            if not dataset:
                msg = 'Aggregation with group "%s" does not exist for dataset.'
                raise ArgumentError(msg % self.group)

        call_async(delete_task, self, dataset)

    def remove_dependent_calculation(self, name):
        new_dependent_calcs = self.dependent_calculations
        new_dependent_calcs.remove(name)
        self.update({self.DEPENDENT_CALCULATIONS: new_dependent_calcs})

    def remove_dependencies(self):
        for name in self.dependencies:
            calculation = self.find_one(self.dataset_id, name)
            calculation.remove_dependent_calculation(self.name)

    def restart_if_has_pending(self, dataset, current_calcs=[]):
        current_names = sorted([self.name] + [c.name for c in current_calcs])
        unfinished = [c for c in dataset.calculations() if c.is_pending]
        unfinished_names = [c.name for c in unfinished[:len(current_names)]]

        if len(unfinished) and current_names != sorted(unfinished_names):
            raise calculate_task.retry()

    def save(self, dataset, formula, name, group_str=None):
        """Parse, save, and calculate a formula.

        Validate `formula` and `group_str` for the given `dataset`. If the
        formula and group are valid for the dataset, then save a new
        calculation for them under `name`. Finally, create a background task
        to compute the calculation.

        Calculations are initially saved in a **pending** state, after the
        calculation has finished processing it will be in a **ready** state.

        :param dataset: The DataSet to save.
        :param formula: The formula to save.
        :param name: The name of the formula.
        :param group_str: Columns to group on.
        :type group_str: String, list or strings, or None.

        :raises: `ParseError` if an invalid formula was supplied.
        """
        # ensure that the formula is parsable
        groups = self.split_groups(group_str) if group_str else []
        Parser.validate(dataset, formula, groups)
        aggregation = Parser.parse_aggregation(formula)

        if aggregation:
            # set group if aggregation and group unset
            group_str = group_str or ''

            # check that name is unique for aggregation
            aggregated_dataset = dataset.aggregated_dataset(groups)

            if aggregated_dataset:
                name = _check_name_and_make_unique(name, aggregated_dataset)

        else:
            # set group if aggregation and group unset
            name = _check_name_and_make_unique(name, dataset)

        record = {
            DATASET_ID: dataset.dataset_id,
            self.AGGREGATION: aggregation,
            self.FORMULA: formula,
            self.GROUP: group_str,
            self.NAME: name,
            self.STATE: self.STATE_PENDING,
        }
        super(self.__class__, self).save(record)

        return self

    def set_aggregation_id(self, _id):
        self.update({self.AGGREGATION_ID: _id})

    def __add_and_update_set(self, link_key, existing, new):
        new_list = list(set(existing + [new]))

        if new_list != existing:
            self.update({link_key: new_list}) class Observation(AbstractModel):

    __collectionname__ = 'observations'

    DELETED_AT = '-1'  # use a short code for key
    ENCODING = 'enc'
    ENCODING_DATASET_ID = '%s_%s' % (DATASET_ID, ENCODING)

    @classmethod
    def delete(cls, dataset, index):
        """Delete observation at index for dataset.

        :param dataset: The dataset to delete the observation from.
        :param index: The index of the observation to delete.
        """
        query = {INDEX: index, DATASET_ID: dataset.dataset_id}
        query = cls.encode(query, dataset=dataset)

        cls.__soft_delete(query)

    @classmethod
    def delete_all(cls, dataset, query=None):
        """Delete the observations for `dataset`.

        :param dataset: The dataset to delete observations for.
        :param query: An optional query to restrict deletion.
        """
        query = query or {}
        query.update({DATASET_ID: dataset.dataset_id})
        query = cls.encode(query, dataset=dataset)

        super(cls, cls()).delete(query)

    @classmethod
    def delete_columns(cls, dataset, columns):
        """Delete a column from the dataset."""
        encoding = cls.encoding(dataset)

        cls.unset({cls.ENCODING_DATASET_ID: dataset.dataset_id},
                  {"%s.%s" % (cls.ENCODING, c): 1 for c in columns})

        cls.unset(
            cls.encode({DATASET_ID: dataset.dataset_id}, encoding=encoding),
            cls.encode({c: 1 for c in columns}, encoding=encoding))

    @classmethod
    def delete_encoding(cls, dataset):
        query = {cls.ENCODING_DATASET_ID: dataset.dataset_id}

        super(cls, cls()).delete(query)

    @classmethod
    def encoding(cls, dataset, encoded_dframe=None):
        record = super(cls, cls).find_one({
            cls.ENCODING_DATASET_ID: dataset.dataset_id}).record

        if record is None and encoded_dframe is not None:
            encoding = cls.__make_encoding(encoded_dframe)
            cls.__store_encoding(dataset, encoding)

            return cls.encoding(dataset)

        return record[cls.ENCODING] if record else None

    @classmethod
    def encode(cls, dict_, dataset=None, encoding=None):
        if dataset:
            encoding = cls.encoding(dataset)

        return replace_keys(dict_, encoding) if encoding else dict_

    @classmethod
    def decoding(cls, dataset):
        return invert_dict(cls.encoding(dataset))

    @classmethod
    def find(cls, dataset, query_args=None, as_cursor=False,
             include_deleted=False):
        """Return observation rows matching parameters.

        :param dataset: Dataset to return rows for.
        :param include_deleted: If True, return delete records, default False.
        :param query_args: An optional QueryArgs to hold the query arguments.

        :raises: `JSONError` if the query could not be parsed.

        :returns: A list of dictionaries matching the passed in `query` and
            other parameters.
        """
        encoding = cls.encoding(dataset) or {}
        query_args = query_args or QueryArgs()

        query_args.query = parse_timestamp_query(query_args.query,
                                                 dataset.schema)
        query_args.encode(encoding, {DATASET_ID: dataset.dataset_id})

        if not include_deleted:
            query = query_args.query
            query[cls.DELETED_AT] = 0
            query_args.query = query

        # exclude deleted at column
        query_args.select = query_args.select or {cls.DELETED_AT: 0}

        distinct = query_args.distinct
        records = super(cls, cls).find(query_args, as_dict=True,
                                       as_cursor=(as_cursor or distinct))

        return records.distinct(encoding.get(distinct, distinct)) if distinct\
            else records

    @classmethod
    def update_from_dframe(cls, df, dataset):
        dataset.build_schema(df)
        encoded_dframe = encode(df.reset_index(), dataset, append_index=False)
        encoding = cls.encoding(dataset)

        cls.__batch_update(encoded_dframe, encoding)
        cls.__store_encoding(dataset, encoding)
        dataset.update_stats(df, update=True)

    @classmethod
    def find_one(cls, dataset, index, decode=True):
        """Return row by index.

        :param dataset: The dataset to find the row for.
        :param index: The index of the row to find.
        """
        query = {INDEX: index, DATASET_ID: dataset.dataset_id,
                 cls.DELETED_AT: 0}
        query = cls.encode(query, dataset=dataset)
        decoding = cls.decoding(dataset)
        record = super(cls, cls).find_one(query, as_dict=True)

        return cls(cls.encode(record, encoding=decoding) if decode else record)

    @classmethod
    def append(cls, dframe, dataset):
        """Append an additional dframe to an existing dataset.

        :params dframe: The DataFrame to append.
        :params dataset: The DataSet to add `dframe` to.
        """
        encoded_dframe = encode(dframe, dataset)
        encoding = cls.encoding(dataset, encoded_dframe)

        cls.__batch_save(encoded_dframe, encoding)
        dataset.clear_summary_stats()

    @classmethod
    def save(cls, dframe, dataset):
        """Save data in `dframe` with the `dataset`.

        Encode `dframe` for MongoDB, and add fields to identify it with the
        passed in `dataset`. All column names in `dframe` are converted to
        slugs using the dataset's schema.  The dataset is update to store the
        size of the stored data.

        :param dframe: The DataFrame to store.
        :param dataset: The dataset to store the dframe in.
        """
        # Build schema for the dataset after having read it from file.
        if not dataset.schema:
            dataset.build_schema(dframe)

        # Update stats, before inplace encoding.
        dataset.update_stats(dframe)

        encoded_dframe = encode(dframe, dataset)
        encoding = cls.encoding(dataset, encoded_dframe)

        cls.__batch_save(encoded_dframe, encoding)

    @classmethod
    def update(cls, dataset, index, record):
        """Update a dataset row by index.

        The record dictionary will update, not replace, the data in the row at
        index.

        :param dataset: The dataset to update a row for.
        :param dex: The index of the row to update.
        :param record: The dictionary to update the row with.
        """
        previous_record = cls.find_one(dataset, index).record
        previous_record.pop(MONGO_ID)
        record = combine_dicts(previous_record, record)
        record = update_calculations(record, dataset)

        record = cls.encode(record, dataset=dataset)

        cls.delete(dataset, index)

        super(cls, cls()).save(record)

    @classmethod
    def batch_read_dframe_from_cursor(cls, dataset, observations, distinct,
                                      limit):
        """Read a DataFrame from a MongoDB Cursor in batches."""
        dframes = []
        batch = 0
        decoding = cls.decoding(dataset)

        while True:
            start = batch * cls.DB_READ_BATCH_SIZE
            end = start + cls.DB_READ_BATCH_SIZE

            if limit > 0 and end > limit:
                end = limit

            # if there is a limit this may occur, and we are done
            if start >= end:
                break

            current_observations = [
                replace_keys(ob, decoding) for ob in observations[start:end]]

            # if the batches exhausted the data
            if not len(current_observations):
                break

            dframes.append(DataFrame(current_observations))

            if not distinct:
                observations.rewind()

            batch += 1

        return concat(dframes) if len(dframes) else DataFrame()

    @classmethod
    def __batch_save(cls, dframe, encoding):
        """Save records in batches to avoid document size maximum setting.

        :param dframe: A DataFrame to save in the current model.
        """
        def command(records, encoding):
            cls.collection.insert(records)

        batch_size = cls.DB_SAVE_BATCH_SIZE

        cls.__batch_command_wrapper(command, dframe, encoding, batch_size)

    @classmethod
    def __batch_update(cls, dframe, encoding):
        """Update records in batches to avoid document size maximum setting.

        DataFrame must have column with record (object) ids.

        :param dfarme: The DataFrame to update.
        """
        def command(records, encoding):
            # Encode the reserved key to access the row ID.
            mongo_id_key = encoding.get(MONGO_ID_ENCODED, MONGO_ID_ENCODED)

            # MongoDB has no batch updates.
            for record in records:
                spec = {MONGO_ID: record[mongo_id_key]}
                del record[mongo_id_key]
                doc = {'$set': record}
                cls.collection.update(spec, doc)

        cls.__batch_command_wrapper(command, dframe, encoding,
                                    cls.DB_SAVE_BATCH_SIZE)

    @classmethod
    def __batch_command_wrapper(cls, command, df, encoding, batch_size):
        try:
            cls.__batch_command(command, df, encoding, batch_size)
        except AutoReconnect:
            batch_size /= 2

            # If batch size drop is less than MIN_BATCH_SIZE, assume the
            # records are too large or there is another error and fail.
            if batch_size >= cls.MIN_BATCH_SIZE:
                cls.__batch_command_wrapper(command, df, encoding, batch_size)

    @classmethod
    def __batch_command(cls, command, dframe, encoding, batch_size):
        batches = int(ceil(float(len(dframe)) / batch_size))

        for batch in xrange(0, batches):
            start = batch * batch_size
            end = start + batch_size
            current_dframe = dframe[start:end]
            records = cls.__encode_records(current_dframe, encoding)
            command(records, encoding)

    @classmethod
    def __encode_records(cls, dframe, encoding):
        return [cls.__encode_record(row.to_dict(), encoding)
                for (_, row) in dframe.iterrows()]

    @classmethod
    def __encode_record(cls, row, encoding):
        encoded = replace_keys(row, encoding)
        encoded[cls.DELETED_AT] = 0

        return encoded

    @classmethod
    def __make_encoding(cls, dframe, start=0):
        # Ensure that DATASET_ID is first so that we can guarantee an index.
        columns = [DATASET_ID] + sorted(dframe.columns - [DATASET_ID])
        return {v: str(start + i) for (i, v) in enumerate(columns)}

    @classmethod
    def __soft_delete(cls, query):
        cls.collection.update(query,
                              {'$set': {cls.DELETED_AT: now().isoformat()}})

    @classmethod
    def __store_encoding(cls, dataset, encoding):
        """Store encoded columns with dataset.

        :param dataset: The dataset to store the encoding with.
        :param encoding: The encoding for dataset.
        """
        record = {cls.ENCODING_DATASET_ID: dataset.dataset_id,
                  cls.ENCODING: encoding}
        super(cls, cls()).delete({cls.ENCODING_DATASET_ID: dataset.dataset_id})
        super(cls, cls()).save(record)
import re
import uuid
from time import gmtime, strftime
from celery.task import task
from pandas import DataFrame, rolling_window
from bamboo.core.calculator import calculate_updates, dframe_from_update,\
    propagate
from bamboo.core.frame import BAMBOO_RESERVED_KEY_PREFIX,\
    DATASET_ID, INDEX, join_dataset, PARENT_DATASET_ID, remove_reserved_keys
from bamboo.core.summary import summarize
from bamboo.lib.async import call_async
from bamboo.lib.exceptions import ArgumentError
from bamboo.lib.mongo import df_mongo_decode
from bamboo.lib.readers import ImportableDataset
from bamboo.lib.query_args import QueryArgs
from bamboo.lib.schema_builder import Schema
from bamboo.lib.utils import combine_dicts, to_list
from bamboo.models.abstract_model import AbstractModel
from bamboo.models.calculation import Calculation
from bamboo.models.observation import Observation
class Dataset(AbstractModel, ImportableDataset):

    __collectionname__ = 'datasets'

    # caching keys
    STATS = '_stats'
    ALL = '_all'

    # metadata
    AGGREGATED_DATASETS = BAMBOO_RESERVED_KEY_PREFIX + 'linked_datasets'
    ATTRIBUTION = 'attribution'
    CREATED_AT = 'created_at'
    DESCRIPTION = 'description'
    ID = 'id'
    JOINED_DATASETS = 'joined_datasets'
    LABEL = 'label'
    LICENSE = 'license'
    NUM_COLUMNS = 'num_columns'
    NUM_ROWS = 'num_rows'
    MERGED_DATASETS = 'merged_datasets'
    PARENT_IDS = 'parent_ids'
    PENDING_UPDATES = 'pending_updates'
    SCHEMA = 'schema'
    UPDATED_AT = 'updated_at'

    def __init__(self, record=None):
        super(Dataset, self).__init__(record)
        self.__dframe = None

    @property
    def aggregated_datasets(self):
        return [(self.split_groups(group), self.find_one(_id)) for (
            group, _id) in self.aggregated_datasets_dict.items()]

    @property
    def aggregated_datasets_dict(self):
        return self.record.get(self.AGGREGATED_DATASETS, {})

    @property
    def attribution(self):
        return self.record.get(self.ATTRIBUTION)

    @property
    def columns(self):
        return self.schema.keys() if self.num_rows else []

    @property
    def dataset_id(self):
        return self.record[DATASET_ID]

    @property
    def description(self):
        return self.record.get(self.DESCRIPTION)

    @property
    def joined_datasets(self):
        # TODO: fetch all datasets in single DB call
        # (let Dataset.find take a list of IDs)
        return [
            (direction, self.find_one(other_dataset_id), on,
             self.find_one(joined_dataset_id))
            for direction, other_dataset_id, on, joined_dataset_id in
            self.joined_dataset_ids]

    @property
    def joined_dataset_ids(self):
        return [
            tuple(_list) for _list in self.record.get(self.JOINED_DATASETS, [])
        ]

    @property
    def label(self):
        return self.record.get(self.LABEL)

    @property
    def labels(self):
        return [column[self.LABEL] for column in self.schema.values()]

    @property
    def license(self):
        return self.record.get(self.LICENSE)

    @property
    def merged_datasets(self):
        return self.__linked_datasets(self.merged_dataset_ids)

    @property
    def merged_datasets_with_map(self):
        results = self.merged_dataset_info

        if len(results):
            mappings, ids = zip(*results)
            results = zip(mappings, self.__linked_datasets(ids))

        return results

    @property
    def merged_dataset_ids(self):
        results = self.merged_dataset_info
        return zip(*results)[-1] if results else results

    @property
    def merged_dataset_info(self):
        return self.record.get(self.MERGED_DATASETS, [])

    @property
    def num_columns(self):
        return self.record.get(self.NUM_COLUMNS, 0)

    @property
    def num_rows(self):
        return self.record.get(self.NUM_ROWS, 0)

    @property
    def on_columns_for_rhs_of_joins(self):
        return [on for direction, _, on, __ in
                self.joined_datasets if direction == 'left']

    @property
    def parent_ids(self):

        query_args = QueryArgs(select={PARENT_DATASET_ID: 1},