Working validation by schema on storage

backdrop/core/data_set.py

@@ -1,8 +1,8 @@
 from collections import namedtuple
 from flask import logging
-from .records import add_auto_ids, parse_timestamp, validate_record, \
+from .records import add_auto_ids, parse_timestamp, validate_record,\
     add_period_keys
-from .validation import data_set_is_valid
+from .validation import data_set_is_valid, validate_record_schema
 from .nested_merge import nested_merge
 from .errors import InvalidSortError
 from backdrop.core.response import PeriodGroupedData, PeriodData, \
@@ -47,6 +47,9 @@ def empty(self):
     def store(self, records):
         log.info('received {} records'.format(len(records)))
 
+        # Validate schema
+        for item in records:
+            validate_record_schema(item, self.config.schema)
         # add auto-id keys
         records = add_auto_ids(records, self.config.auto_ids)
         # parse _timestamp
@@ -126,7 +129,7 @@ def _limit_grouped_results(results, limit):
     "_DataSetConfig",
     "name data_group data_type raw_queries_allowed bearer_token upload_format "
     "upload_filters auto_ids queryable realtime capped_size max_age_expected "
-    "published")
+    "published schema")
 
 
 class DataSetConfig(_DataSetConfig):
@@ -135,7 +138,7 @@ def __new__(cls, name, data_group, data_type, raw_queries_allowed=False,
                 bearer_token=None, upload_format="csv", upload_filters=None,
                 auto_ids=None, queryable=True, realtime=False,
                 capped_size=5040, max_age_expected=2678400,
-                published=True):
+                published=True, schema={}):
         if not data_set_is_valid(name):
             raise ValueError("DataSet name is not valid: '{}'".format(name))
 
@@ -150,7 +153,7 @@ def __new__(cls, name, data_group, data_type, raw_queries_allowed=False,
                                                  upload_filters, auto_ids,
                                                  queryable, realtime,
                                                  capped_size, max_age_expected,
-                                                 published)
+                                                 published, schema)
 
     @property
     def max_age(self):