Repository: frictionlessdata/frictionless-py

Issue:
Improve SPSS/BigQuery plugins
# Overview

At the moment, the SPSS import/export write in-general but type mappings especially on reading is not correct.

# Tasks

- [ ] fix mappings
- [ ] fix unicode support (test `data/table.csv`)

# References
- https://pythonhosted.org/savReaderWriter/index.html
- https://github.com/frictionlessdata/datapackage-pipelines-spss-driver

---

> https://github.com/frictionlessdata/frictionless-py/issues/410

With `tableschema-spss` we run into a few things that don't allow to use it as a first-class citizien in `tableschema/datapackage` integrations:

- more important that we should have special Table Schema descriptor containing `spss:format` properties. May be it's possible to infer this information somehow? For examples `tableschema-pandas` only creates an emtpy dataframe on `storage.create` and then on `storage.write` does real descriptor mapping using provided data (the same we should do here to get size of data for SPSS formats)
- less important that bucket names include `.sav` extension. I think we still better to hide storage backend details and use abstract bucket names like `bucket` instead of `bucket.sav` (adding/removing an extension on the mapping step). But that's not vital.

So for now we can't use it in standard integration scenario with `tableschema/datapackage` like:

```python
from tableschema import Table

table = Table('data.csv', schema='schema.csv')
table.save('data', storage='sql', engine=engine)
table.save('data', storage='bigquery', project=project, dataset=dataset)
table.save('data', storage='pandas')
# Will fail if schema doesn't have spss:format properties
# table.save('data', storage='spss', base_path='dir/path') 
```

Not sure we have something to do with it now because it works great for other stuff (SPSS-base pilot etc). So just leaving it here for a future.

---

Our storage plugins for `SQL/BigQuery/Pandas` are able to pass this testsuite (with some differences in reflected schema - here just an example). It's updated to `SPSS` storage and use simple `cast` function to cast resource data:


```python
# Resources

ARTICLES = {
    'schema': {
        'fields': [
            {'name': 'id', 'type': 'integer', 'constraints': {'required': True}},
            {'name': 'parent', 'type': 'integer'},
            {'name': 'name', 'type': 'string'},
            {'name': 'current', 'type': 'boolean'},
            {'name': 'rating', 'type': 'number'},
        ],
        # 'primaryKey': 'id',
        # 'foreignKeys': [
            # {'fields': 'parent', 'reference': {'resource': '', 'fields': 'id'}},
        # ],
    },
    'data': [
        ['1', '', 'Taxes', 'True', '9.5'],
        ['2', '1', '中国人', 'False', '7'],
    ],
}
COMMENTS = {
    'schema': {
        'fields': [
            {'name': 'entry_id', 'type': 'integer', 'constraints': {'required': True}},
            {'name': 'comment', 'type': 'string'},
            {'name': 'note', 'type': 'any'},
        ],
        # 'primaryKey': 'entry_id',
        # 'foreignKeys': [
            # {'fields': 'entry_id', 'reference': {'resource': 'articles', 'fields': 'id'}},
        # ],
    },
    'data': [
        ['1', 'good', 'note1'],
        ['2', 'bad', 'note2'],
    ],
}
TEMPORAL = {
    'schema': {
        'fields': [
            {'name': 'date', 'type': 'date'},
            {'name': 'date_year', 'type': 'date', 'format': '%Y'},
            {'name': 'datetime', 'type': 'datetime'},
            {'name': 'duration', 'type': 'duration'},
            {'name': 'time', 'type': 'time'},
            {'name': 'year', 'type': 'year'},
            {'name': 'yearmonth', 'type': 'yearmonth'},
        ],
    },
    'data': [
        ['2015-01-01', '2015', '2015-01-01T03:00:00Z', 'P1Y1M', '03:00:00', '2015', '2015-01'],
        ['2015-12-31', '2015', '2015-12-31T15:45:33Z', 'P2Y2M', '15:45:33', '2015', '2015-01'],
    ],
}
LOCATION = {
    'schema': {
        'fields': [
            {'name': 'location', 'type': 'geojson'},
            {'name': 'geopoint', 'type': 'geopoint'},
        ],
    },
    'data': [
        ['{"type": "Point","coordinates":[33.33,33.33]}', '30,75'],
        ['{"type": "Point","coordinates":[50.00,50.00]}', '90,45'],
    ],
}
COMPOUND = {
    'schema': {
        'fields': [
            {'name': 'stats', 'type': 'object'},
            {'name': 'persons', 'type': 'array'},
        ],
    },
    'data': [
        ['{"chars":560}', '["Mike", "John"]'],
        ['{"chars":970}', '["Paul", "Alex"]'],
    ],
}


# Tests

def test_storage(tmpdir):

    # Create storage
    storage = Storage(base_path=str(tmpdir))

    # Delete buckets
    storage.delete()

    # Create buckets
    storage.create(['articles', 'comments'], [ARTICLES['schema'], COMMENTS['schema']])
    storage.create('comments', COMMENTS['schema'], force=True)
    storage.create('temporal', TEMPORAL['schema'])
    storage.create('location', LOCATION['schema'])
    storage.create('compound', COMPOUND['schema'])

    # Write data
    storage.write('articles', ARTICLES['data'])
    storage.write('comments', COMMENTS['data'])
    storage.write('temporal', TEMPORAL['data'])
    storage.write('location', LOCATION['data'])
    storage.write('compound', COMPOUND['data'])

    # Create new storage to use reflection only
    storage = Storage(base_path=str(tmpdir))

    # Create existent bucket
    with pytest.raises(tableschema.exceptions.StorageError):
        storage.create('articles.sav', ARTICLES['schema'])

    # Assert buckets
    assert storage.buckets == ['articles', 'compound', 'location', 'temporal', 'comments']

    # Assert schemas
    assert storage.describe('articles') == ARTICLES['schema']
    assert storage.describe('comments') == {
        'fields': [
            {'name': 'entry_id', 'type': 'integer', 'constraints': {'required': True}},
            {'name': 'comment', 'type': 'string'},
            {'name': 'note', 'type': 'string'}, # type downgrade
        ],
    }
    assert storage.describe('temporal') == {
        'fields': [
            {'name': 'date', 'type': 'date'},
            {'name': 'date_year', 'type': 'date'}, # format removal
            {'name': 'datetime', 'type': 'datetime'},
            {'name': 'duration', 'type': 'string'}, # type fallback
            {'name': 'time', 'type': 'time'},
            {'name': 'year', 'type': 'integer'}, # type downgrade
            {'name': 'yearmonth', 'type': 'string'}, # type fallback
        ],
    }
    assert storage.describe('location') == {
        'fields': [
            {'name': 'location', 'type': 'object'}, # type downgrade
            {'name': 'geopoint', 'type': 'string'}, # type fallback
        ],
    }
    assert storage.describe('compound') == {
        'fields': [
            {'name': 'stats', 'type': 'object'},
            {'name': 'persons', 'type': 'string'}, # type fallback
        ],
    }

    # Assert data
    assert storage.read('articles') == cast(ARTICLES)['data']
    assert storage.read('comments') == cast(COMMENTS)['data']
    assert storage.read('temporal') == cast(TEMPORAL, skip=['duration', 'yearmonth'])['data']
    assert storage.read('location') == cast(LOCATION, skip=['geopoint'])['data']
    assert storage.read('compound') == cast(COMPOUND, skip=['array'])['data']

    # Assert data with forced schema
    storage.describe('compound.sav', COMPOUND['schema'])
    assert storage.read('compound.sav') == cast(COMPOUND)['data']

    # Delete non existent bucket
    with pytest.raises(tableschema.exceptions.StorageError):
        storage.delete('non_existent')

    # Delete buckets
    storage.delete()
```

Discussion:

Patch:
diff --git a/.travis.yml b/.travis.yml
index 9dab9059..0ca03778 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -20,7 +20,7 @@ before_install:
   - cp .env.example .env
   - psql -c 'CREATE DATABASE test;' -U postgres
   - mysql -e 'CREATE DATABASE test CHARACTER SET utf8 COLLATE utf8_general_ci;'
-  - openssl aes-256-cbc -K $encrypted_4885f94bfb6d_key -iv $encrypted_4885f94bfb6d_iv -in .google.json.enc -out .google.json -d
+  - openssl aes-256-cbc -K $encrypted_4885f94bfb6d_key -iv $encrypted_4885f94bfb6d_iv -in .google.json.enc -out .google.json -d || echo 'Encrypted files have been removed for security reasons.'
 
 install:
   # TODO: remove after
diff --git a/frictionless/parser.py b/frictionless/parser.py
index 7c4875d9..b567b3d9 100644
--- a/frictionless/parser.py
+++ b/frictionless/parser.py
@@ -136,11 +136,11 @@ class Parser:
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         """Write row stream into the resource
 
         Parameters:
-            gen<Row[]>: row stream
+            read_row_stream (gen<Row[]>): row stream factory
         """
         raise NotImplementedError
 
diff --git a/frictionless/parsers/csv.py b/frictionless/parsers/csv.py
index 62ee4252..5724ec2c 100644
--- a/frictionless/parsers/csv.py
+++ b/frictionless/parsers/csv.py
@@ -49,7 +49,7 @@ class CsvParser(Parser):
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         options = {}
         for name in vars(self.resource.dialect.to_python()):
             value = getattr(self.resource.dialect, name, None)
@@ -57,7 +57,7 @@ class CsvParser(Parser):
                 options[name] = value
         with tempfile.NamedTemporaryFile(delete=False) as file:
             writer = unicodecsv.writer(file, encoding=self.resource.encoding, **options)
-            for row in row_stream:
+            for row in read_row_stream():
                 schema = row.schema
                 if row.row_number == 1:
                     writer.writerow(schema.field_names)
diff --git a/frictionless/parsers/excel.py b/frictionless/parsers/excel.py
index c64bbdbf..c50c5aa6 100644
--- a/frictionless/parsers/excel.py
+++ b/frictionless/parsers/excel.py
@@ -117,7 +117,7 @@ class XlsxParser(Parser):
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         dialect = self.resource.dialect
         helpers.ensure_dir(self.resource.source)
         book = openpyxl.Workbook(write_only=True)
@@ -125,7 +125,7 @@ class XlsxParser(Parser):
         if isinstance(title, int):
             title = f"Sheet {dialect.sheet}"
         sheet = book.create_sheet(title)
-        for row in row_stream:
+        for row in read_row_stream():
             cells = []
             if row.row_number == 1:
                 sheet.append(row.schema.field_names)
@@ -223,7 +223,7 @@ class XlsParser(Parser):
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         dialect = self.resource.dialect
         helpers.ensure_dir(self.resource.source)
         book = xlwt.Workbook()
@@ -231,7 +231,7 @@ class XlsParser(Parser):
         if isinstance(title, int):
             title = f"Sheet {dialect.sheet}"
         sheet = book.add_sheet(title)
-        for row_index, row in enumerate(row_stream):
+        for row_index, row in enumerate(read_row_stream()):
             if row.row_number == 1:
                 for field_index, name in enumerate(row.schema.field_names):
                     sheet.write(0, field_index, name)
diff --git a/frictionless/parsers/inline.py b/frictionless/parsers/inline.py
index e72b2f2c..422d20bf 100644
--- a/frictionless/parsers/inline.py
+++ b/frictionless/parsers/inline.py
@@ -74,10 +74,10 @@ class InlineParser(Parser):
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         dialect = self.resource.dialect
         self.resource.data = []
-        for row in row_stream:
+        for row in read_row_stream():
             item = row.to_dict() if dialect.keyed else list(row.values())
             if not dialect.keyed and row.row_number == 1:
                 self.resource.data.append(row.schema.field_names)
diff --git a/frictionless/parsers/json.py b/frictionless/parsers/json.py
index 10ac6c49..74b57021 100644
--- a/frictionless/parsers/json.py
+++ b/frictionless/parsers/json.py
@@ -53,10 +53,10 @@ class JsonParser(Parser):
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         data = []
         dialect = self.resource.dialect
-        for row in row_stream:
+        for row in read_row_stream():
             cells = list(row.values())
             cells, notes = row.schema.write_data(cells, native_types=self.native_types)
             item = dict(zip(row.schema.field_names, cells)) if dialect.keyed else cells
@@ -103,11 +103,11 @@ class JsonlParser(Parser):
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         dialect = self.resource.dialect
         with tempfile.NamedTemporaryFile(delete=False) as file:
             writer = jsonlines.Writer(file)
-            for row in row_stream:
+            for row in read_row_stream():
                 schema = row.schema
                 cells = list(row.values())
                 cells, notes = schema.write_data(cells, native_types=self.native_types)
diff --git a/frictionless/plugins/bigquery.py b/frictionless/plugins/bigquery.py
index e3d12add..e0eea157 100644
--- a/frictionless/plugins/bigquery.py
+++ b/frictionless/plugins/bigquery.py
@@ -147,7 +147,7 @@ class BigqueryParser(Parser):
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         dialect = self.resource.dialect
         schema = self.resource.schema
         storage = BigqueryStorage(
@@ -155,7 +155,7 @@ class BigqueryParser(Parser):
             project=dialect.project,
             dataset=dialect.dataset,
         )
-        resource = Resource(name=dialect.table, data=row_stream, schema=schema)
+        resource = Resource(name=dialect.table, data=read_row_stream, schema=schema)
         storage.write_resource(resource)
 
 
@@ -227,7 +227,7 @@ class BigqueryStorage(Storage):
 
         # Create resource
         schema = self.__read_convert_schema(response["schema"])
-        data = partial(self.__read_data_stream, name, schema)
+        data = partial(self.__read_convert_data, name, schema)
         resource = Resource(name=name, schema=schema, data=data)
 
         return resource
@@ -257,28 +257,7 @@ class BigqueryStorage(Storage):
 
         return schema
 
-    def __read_convert_type(self, bq_type):
-
-        # Mapping
-        mapping = {
-            "BOOLEAN": "boolean",
-            "DATE": "date",
-            "DATETIME": "datetime",
-            "INTEGER": "integer",
-            "FLOAT": "number",
-            "STRING": "string",
-            "TIME": "time",
-        }
-
-        # Return type
-        if bq_type in mapping:
-            return mapping[bq_type]
-
-        # Not supported
-        note = "Type %s is not supported" % type
-        raise exceptions.FrictionlessException(errors.StorageError(note=note))
-
-    def __read_data_stream(self, name, schema):
+    def __read_convert_data(self, name, schema):
         bq_name = self.__write_convert_name(name)
 
         # Get response
@@ -306,6 +285,26 @@ class BigqueryStorage(Storage):
         yield schema.field_names
         yield from data
 
+    def __read_convert_type(self, bq_type=None):
+
+        # Mapping
+        mapping = {
+            "BOOLEAN": "boolean",
+            "DATE": "date",
+            "DATETIME": "datetime",
+            "INTEGER": "integer",
+            "FLOAT": "number",
+            "STRING": "string",
+            "TIME": "time",
+        }
+
+        # Return type
+        if bq_type:
+            return mapping.get(bq_type, "string")
+
+        # Return mapping
+        return mapping
+
     # Write
 
     def write_resource(self, resource, *, force=False):
@@ -344,23 +343,21 @@ class BigqueryStorage(Storage):
             ).execute()
 
             # Write data
-            self.__write_row_stream(resource)
+            self.__write_convert_data(resource)
 
     def __write_convert_name(self, name):
         return _slugify_name(self.__prefix + name)
 
     def __write_convert_schema(self, schema):
+        bq_schema = {"fields": []}
 
         # Fields
-        bq_fields = []
         for field in schema.fields:
             bq_type = self.__write_convert_type(field.type)
-            if not bq_type:
-                bq_type = "STRING"
             mode = "NULLABLE"
             if field.required:
                 mode = "REQUIRED"
-            bq_fields.append(
+            bq_schema["fields"].append(
                 {
                     "name": _slugify_name(field.name),
                     "type": bq_type,
@@ -368,66 +365,31 @@ class BigqueryStorage(Storage):
                 }
             )
 
-        # Schema
-        bq_schema = {
-            "fields": bq_fields,
-        }
-
         return bq_schema
 
-    def __write_convert_type(self, type):
-        mapping = self.__write_convert_types()
-
-        # Supported type
-        if type in mapping:
-            return mapping[type]
-
-        # Not supported
-        note = "Type %s is not supported" % type
-        raise exceptions.FrictionlessException(errors.StorageError(note=note))
-
-    def __write_convert_types(self):
-        return {
-            "any": "STRING",
-            "array": None,
-            "boolean": "BOOLEAN",
-            "date": "DATE",
-            "datetime": "DATETIME",
-            "duration": None,
-            "geojson": None,
-            "geopoint": None,
-            "integer": "INTEGER",
-            "number": "FLOAT",
-            "object": None,
-            "string": "STRING",
-            "time": "TIME",
-            "year": "INTEGER",
-            "yearmonth": None,
-        }
-
-    def __write_row_stream(self, resource):
-        mapping = self.__write_convert_types()
+    def __write_convert_data(self, resource):
+        mapping = self.__write_convert_type()
 
         # Fallback fields
         fallback_fields = []
-        mapping = self.__write_convert_types()
+        mapping = self.__write_convert_type()
         for field in resource.schema.fields:
-            if mapping[field.type] is None:
+            if not mapping.get(field.type):
                 fallback_fields.append(field)
 
         # Write data
         buffer = []
-        for row in resource.read_rows():
+        for row in resource.read_row_stream():
             for field in fallback_fields:
                 row[field.name], notes = field.write_cell(row[field.name])
             buffer.append(row.to_list())
             if len(buffer) > BUFFER_SIZE:
-                self.__write_row_stream_buffer(resource.name, buffer)
+                self.__write_convert_data_start_job(resource.name, buffer)
                 buffer = []
         if len(buffer) > 0:
-            self.__write_row_stream_buffer(resource.name, buffer)
+            self.__write_convert_data_start_job(resource.name, buffer)
 
-    def __write_row_stream_buffer(self, name, buffer):
+    def __write_convert_data_start_job(self, name, buffer):
         http = helpers.import_from_plugin("apiclient.http", plugin="bigquery")
         bq_name = self.__write_convert_name(name)
 
@@ -465,14 +427,14 @@ class BigqueryStorage(Storage):
 
         # Wait the job
         try:
-            self.__write_wait_job_is_done(response)
+            self.__write_convert_data_finish_job(response)
         except Exception as exception:
             if "not found: job" in str(exception).lower():
                 note = "BigQuery plugin supports only the US location of datasets"
                 raise exceptions.FrictionlessException(errors.StorageError(note=note))
             raise
 
-    def __write_wait_job_is_done(self, response):
+    def __write_convert_data_finish_job(self, response):
 
         # Get job instance
         job = self.__service.jobs().get(
@@ -490,6 +452,28 @@ class BigqueryStorage(Storage):
                 break
             time.sleep(1)
 
+    def __write_convert_type(self, type=None):
+
+        # Mapping
+        mapping = {
+            "any": "STRING",
+            "boolean": "BOOLEAN",
+            "date": "DATE",
+            "datetime": "DATETIME",
+            "integer": "INTEGER",
+            "number": "FLOAT",
+            "string": "STRING",
+            "time": "TIME",
+            "year": "INTEGER",
+        }
+
+        # Return type
+        if type:
+            return mapping.get(type, "STRING")
+
+        # Return mapping
+        return mapping
+
     # Delete
 
     def delete_resource(self, name, *, ignore=False):
diff --git a/frictionless/plugins/gsheet.py b/frictionless/plugins/gsheet.py
index 05e80708..68a2184c 100644
--- a/frictionless/plugins/gsheet.py
+++ b/frictionless/plugins/gsheet.py
@@ -83,7 +83,6 @@ class GsheetParser(Parser):
 
     # Write
 
-    # NOTE: if we migrate to the native driver we can enable it
-    def write(self, row_stream, *, schema):
+    def write(self, read_row_stream):
         error = errors.Error(note="Writing to Google Sheets is not supported")
         raise exceptions.FrictionlessException(error)
diff --git a/frictionless/plugins/html.py b/frictionless/plugins/html.py
index 3312fc94..c5e23511 100644
--- a/frictionless/plugins/html.py
+++ b/frictionless/plugins/html.py
@@ -141,9 +141,9 @@ class HtmlParser(Parser):
 
     # NOTE: rebase on proper pyquery
     # NOTE: take dialect into account
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         html = "<html><body><table>\n"
-        for row in row_stream:
+        for row in read_row_stream():
             if row.row_number == 1:
                 html += "<tr>"
                 for name in row.schema.field_names:
diff --git a/frictionless/plugins/ods.py b/frictionless/plugins/ods.py
index f789f9e4..e2c62bd8 100644
--- a/frictionless/plugins/ods.py
+++ b/frictionless/plugins/ods.py
@@ -165,7 +165,7 @@ class OdsParser(Parser):
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         dialect = self.resource.dialect
         helpers.ensure_dir(self.resource.source)
         ezodf = helpers.import_from_plugin("ezodf", plugin="ods")
@@ -173,7 +173,7 @@ class OdsParser(Parser):
         title = f"Sheet {dialect.sheet}"
         book.sheets += ezodf.Sheet(title)
         sheet = book.sheets[title]
-        for row_index, row in enumerate(row_stream):
+        for row_index, row in enumerate(read_row_stream()):
             if row.row_number == 1:
                 for field_index, name in enumerate(row.schema.field_names):
                     sheet[(0, field_index)].set_value(name)
diff --git a/frictionless/plugins/pandas.py b/frictionless/plugins/pandas.py
index 86559655..2d64d758 100644
--- a/frictionless/plugins/pandas.py
+++ b/frictionless/plugins/pandas.py
@@ -88,10 +88,10 @@ class PandasParser(Parser):
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         schema = self.resource.schema
         storage = PandasStorage()
-        resource = Resource(name=self.resource.name, data=row_stream, schema=schema)
+        resource = Resource(name=self.resource.name, data=read_row_stream, schema=schema)
         storage.write_resource(resource)
         self.resource.data = storage.dataframe
 
@@ -136,7 +136,7 @@ class PandasStorage(Storage):
             note = f'Resource "{name}" does not exist'
             raise exceptions.FrictionlessException(errors.StorageError(note=note))
         schema = self.__read_convert_schema(dataframe)
-        data = partial(self.__read_data_stream, name, schema)
+        data = partial(self.__read_convert_data, name, schema)
         resource = Resource(name=name, schema=schema, data=data)
         return resource
 
@@ -170,6 +170,25 @@ class PandasStorage(Storage):
         # Return schema
         return schema
 
+    def __read_convert_data(self, name, schema):
+        np = helpers.import_from_plugin("numpy", plugin="pandas")
+        dataframe = self.__read_pandas_dataframe(name)
+        yield schema.field_names
+        for pk, item in dataframe.iterrows():
+            cells = []
+            for field in schema.fields:
+                if field.name in schema.primary_key:
+                    pk = pk if isinstance(pk, tuple) else [pk]
+                    value = pk[schema.primary_key.index(field.name)]
+                else:
+                    value = item[field.name]
+                if field.type == "number" and np.isnan(value):
+                    value = None
+                elif field.type == "datetime":
+                    value = value.to_pydatetime()
+                cells.append(value)
+            yield cells
+
     def __read_convert_type(self, dtype, sample=None):
         pdc = helpers.import_from_plugin("pandas.core.dtypes.api", plugin="pandas")
 
@@ -201,25 +220,6 @@ class PandasStorage(Storage):
         # Default
         return "string"
 
-    def __read_data_stream(self, name, schema):
-        np = helpers.import_from_plugin("numpy", plugin="pandas")
-        dataframe = self.__read_pandas_dataframe(name)
-        yield schema.field_names
-        for pk, item in dataframe.iterrows():
-            cells = []
-            for field in schema.fields:
-                if field.name in schema.primary_key:
-                    pk = pk if isinstance(pk, tuple) else [pk]
-                    value = pk[schema.primary_key.index(field.name)]
-                else:
-                    value = item[field.name]
-                if field.type == "number" and np.isnan(value):
-                    value = None
-                elif field.type == "datetime":
-                    value = value.to_pydatetime()
-                cells.append(value)
-            yield cells
-
     def __read_pandas_dataframe(self, name):
         return self.__dataframes.get(name)
 
@@ -254,7 +254,7 @@ class PandasStorage(Storage):
         data_rows = []
         index_rows = []
         fixed_types = {}
-        for row in resource.read_rows():
+        for row in resource.read_row_stream():
             data_values = []
             index_values = []
             for field in resource.schema.fields:
@@ -305,35 +305,26 @@ class PandasStorage(Storage):
 
         return dataframe
 
-    def __write_convert_type(self, type):
+    def __write_convert_type(self, type=None):
         np = helpers.import_from_plugin("numpy", plugin="pandas")
 
         # Mapping
         mapping = {
-            "any": np.dtype("O"),
             "array": np.dtype(list),
             "boolean": np.dtype(bool),
-            "date": np.dtype("O"),
             "datetime": np.dtype("datetime64[ns]"),
-            "duration": np.dtype("O"),
-            "geojson": np.dtype("O"),
-            "geopoint": np.dtype("O"),
             "integer": np.dtype(int),
             "number": np.dtype(float),
             "object": np.dtype(dict),
-            "string": np.dtype("O"),
-            "time": np.dtype("O"),
             "year": np.dtype(int),
-            "yearmonth": np.dtype("O"),
         }
 
         # Return type
-        if type in mapping:
-            return mapping[type]
+        if type:
+            return mapping.get(type, np.dtype("O"))
 
-        # Not supported type
-        note = f'Field type "{type}" is not supported'
-        raise exceptions.FrictionlessException(errors.StorageError(note=note))
+        # Return mapping
+        return mapping
 
     # Delete
 
diff --git a/frictionless/plugins/spss.py b/frictionless/plugins/spss.py
index abd5bba8..9983986a 100644
--- a/frictionless/plugins/spss.py
+++ b/frictionless/plugins/spss.py
@@ -3,8 +3,10 @@ import os
 import warnings
 from functools import partial
 from ..resource import Resource
+from ..dialects import Dialect
 from ..package import Package
 from ..storage import Storage
+from ..parser import Parser
 from ..plugin import Plugin
 from ..schema import Schema
 from ..field import Field
@@ -25,14 +27,74 @@ class SpssPlugin(Plugin):
 
     """
 
+    def create_dialect(self, resource, *, descriptor):
+        if resource.format in ["sav", "zsav"]:
+            return SpssDialect(descriptor)
+
+    def create_parser(self, resource):
+        if resource.format in ["sav", "zsav"]:
+            return SpssParser(resource)
+
     def create_storage(self, name, **options):
         if name == "spss":
-            warnings.warn("SPSS support is in a draft state", UserWarning)
             return SpssStorage(**options)
 
 
-# TODO: implement Dialect
-# TODO: implement Parser
+# Dialect
+
+
+class SpssDialect(Dialect):
+    """Spss dialect representation
+
+    API      | Usage
+    -------- | --------
+    Public   | `from frictionless.plugins.spss import SpssDialect`
+
+    Parameters:
+        descriptor? (str|dict): descriptor
+
+    Raises:
+        FrictionlessException: raise any error that occurs during the process
+
+    """
+
+    pass
+
+
+# Parser
+
+
+class SpssParser(Parser):
+    """Spss parser implementation.
+
+    API      | Usage
+    -------- | --------
+    Public   | `from frictionless.plugins.spss import SpssParser`
+
+    """
+
+    loading = False
+
+    # Read
+
+    def read_data_stream_create(self):
+        name = os.path.basename(self.resource.path)
+        basepath = os.path.dirname(self.resource.path)
+        storage = SpssStorage(basepath=basepath)
+        resource = storage.read_resource(name)
+        self.resource.schema = resource.schema
+        yield resource.schema.field_names
+        yield from resource.read_data_stream()
+
+    # Write
+
+    def write(self, read_row_stream):
+        name = os.path.basename(self.resource.path)
+        basepath = os.path.dirname(self.resource.path)
+        schema = self.resource.schema
+        storage = SpssStorage(basepath=basepath)
+        resource = Resource(name=name, data=read_row_stream, schema=schema)
+        storage.write_resource(resource)
 
 
 # Storage
@@ -52,12 +114,18 @@ class SpssStorage(Storage):
     """
 
     def __init__(self, *, basepath=None):
+
+        # Set attributes
         basepath = basepath or os.getcwd()
         if not os.path.isdir(basepath):
             note = f'Path "{basepath}" is not a directory, or doesn\'t exist'
             raise exceptions.FrictionlessException(errors.StorageError(note=note))
         self.__basepath = basepath
 
+        # Silent warnings
+        sav = helpers.import_from_plugin("savReaderWriter", plugin="spss")
+        warnings.filterwarnings("ignore", category=sav.SPSSIOWarning)
+
     def __iter__(self):
         names = []
         for path in os.listdir(self.__basepath):
@@ -74,10 +142,10 @@ class SpssStorage(Storage):
         if not os.path.isfile(path):
             note = f'Resource "{name}" does not exist'
             raise exceptions.FrictionlessException(errors.StorageError(note=note))
-        with sav.SavHeaderReader(path, ioUtf8=True) as header:
-            spss_schema = header.all()
+        with sav.SavHeaderReader(path, ioUtf8=True) as reader:
+            spss_schema = reader.all()
             schema = self.__read_convert_schema(spss_schema)
-            data = partial(self.__read_data_stream, name, schema)
+            data = partial(self.__read_convert_data, name, schema)
             resource = Resource(name=name, schema=schema, data=data)
             return resource
 
@@ -94,28 +162,34 @@ class SpssStorage(Storage):
 
     def __read_convert_schema(self, spss_schema):
         schema = Schema()
-
-        # Formats
-        date_formats = {
-            "time": TIME_FORMAT,
-            "date": DATE_FORMAT,
-            "datetime": DATETIME_FORMAT,
-        }
-
-        # Fields
         for name in spss_schema.varNames:
             type = self.__read_convert_type(spss_schema.formats[name])
             field = Field(name=name, type=type)
             title = spss_schema.varLabels[name]
             if title:
                 field.title = title
-            if type in date_formats.keys():
-                field.format = date_formats[type]
             schema.fields.append(field)
-
         return schema
 
-    def __read_convert_type(self, spss_type):
+    def __read_convert_data(self, name, schema):
+        sav = helpers.import_from_plugin("savReaderWriter", plugin="spss")
+        path = self.__write_convert_name(name)
+        yield schema.field_names
+        with sav.SavReader(path, ioUtf8=True) as reader:
+            for item in reader:
+                cells = []
+                for index, field in enumerate(schema.fields):
+                    value = item[index]
+                    if value is not None:
+                        if field.type == "integer":
+                            value = int(float(value))
+                        elif field.type in ["datetime", "date", "time"]:
+                            format = FORMAT_READ[field.type]
+                            value = reader.spss2strDate(value, format, None)
+                    cells.append(value)
+                yield cells
+
+    def __read_convert_type(self, spss_type=None):
 
         # Mapping
         mapping = [
@@ -134,33 +208,14 @@ class SpssStorage(Storage):
         ]
 
         # Return type
-        for type, pattern in mapping:
-            if pattern.match(spss_type):
-                return type
+        if spss_type:
+            for type, pattern in mapping:
+                if pattern.match(spss_type):
+                    return type
+            return "string"
 
-        # Default
-        return "string"
-
-    def __read_data_stream(self, name, schema):
-        sav = helpers.import_from_plugin("savReaderWriter", plugin="spss")
-        path = self.__write_convert_name(name)
-        yield schema.field_names
-        with sav.SavReader(path, ioUtf8=True, rawMode=False) as reader:
-            for item in reader:
-                cells = []
-                for index, field in enumerate(schema.fields):
-                    value = item[index]
-                    # Fix decimals that should be integers
-                    if field.type == "integer" and value is not None:
-                        value = int(float(value))
-                    # We need to decode bytes to strings
-                    if isinstance(value, bytes):
-                        value = value.decode(reader.fileEncoding)
-                    # Time values need a decimal, add one if missing.
-                    if field.type == "time" and not re.search(r"\.\d*", value):
-                        value = "{}.0".format(value)
-                    cells.append(value)
-                yield cells
+        # Return mapping
+        return mapping
 
     # Write
 
@@ -183,7 +238,7 @@ class SpssStorage(Storage):
         for resource in package.resources:
             if not resource.schema:
                 resource.infer(only_sample=True)
-            self.__write_row_stream(resource)
+            self.__write_convert_data(resource)
 
     def __write_convert_name(self, name):
         path = os.path.normpath(os.path.join(self.__basepath, f"{name}.sav"))
@@ -192,55 +247,75 @@ class SpssStorage(Storage):
             raise exceptions.FrictionlessException(errors.StorageError(note=note))
         return path
 
-    def __write_convert_schema(self, schema):
-        spss_schema = {"varNames": [], "varTypes": {}}
-        for field in schema.fields:
+    def __write_convert_schema(self, resource):
+        spss_schema = {"varNames": [], "varLabels": {}, "varTypes": {}, "formats": {}}
+        mapping = self.__write_convert_type()
+
+        # Add fields
+        sizes = {}
+        for field in resource.schema.fields:
             spss_schema["varNames"].append(field.name)
-            spss_schema["varTypes"][field.name] = self.__write_convert_type(field.type)
-        return spss_schema
+            if field.title:
+                spss_schema["varLabels"][field.name] = field.title
+            spss_type = mapping.get(field.type)
+            if spss_type:
+                spss_schema["varTypes"][field.name] = spss_type[0]
+                spss_schema["formats"][field.name] = spss_type[1]
+            else:
+                sizes[field.name] = 0
+
+        # Set string sizes
+        for row in resource.read_row_stream():
+            for name in sizes.keys():
+                cell = row[name]
+                field = resource.schema.get_field(name)
+                cell, notes = field.write_cell(cell)
+                size = len(cell.encode("utf-8"))
+                if size > sizes[name]:
+                    sizes[name] = size
+        for name, size in sizes.items():
+            spss_schema["varTypes"][name] = size
 
-    def __write_convert_type(self, type):
-        if type in ["integer", "number", "date", "datetime", "time", "year"]:
-            return 0
-        return 5
+        return spss_schema
 
-    # NOTE: move partially to write_package
-    def __write_row_stream(self, resource):
+    def __write_convert_data(self, resource):
+        mapping = self.__write_convert_type()
         sav = helpers.import_from_plugin("savReaderWriter", plugin="spss")
-
-        # Prepare SPSS meta
         path = self.__write_convert_name(resource.name)
-        spss_schema = self.__write_convert_schema(resource.schema)
-        fallback_types = [
-            "boolean",
-            "object",
-            "geojson",
-            "geopoint",
-            "array",
-            "duration",
-            "yearmonth",
-        ]
-
-        # Write row stream
+        spss_schema = self.__write_convert_schema(resource)
         with sav.SavWriter(path, ioUtf8=True, **spss_schema) as writer:
-            for row in resource.read_rows():
+            for row in resource.read_row_stream():
                 result = []
                 for field in resource.schema.fields:
                     cell = row[field.name]
-                    if field.type == "date":
-                        cell = cell.strftime(DATE_FORMAT).encode()
-                        cell = writer.spssDateTime(cell, DATE_FORMAT)
-                    elif field.type == "datetime":
-                        cell = cell.strftime(DATETIME_FORMAT).encode()
-                        cell = writer.spssDateTime(cell, DATETIME_FORMAT)
-                    elif field.type == "time":
-                        cell = cell.strftime(TIME_FORMAT).encode()
-                        cell = writer.spssDateTime(cell, TIME_FORMAT)
-                    elif field.type in fallback_types:
+                    if field.type in ["datetime", "date", "time"]:
+                        format = FORMAT_WRITE[field.type]
+                        cell = writer.spssDateTime(cell.strftime(format).encode(), format)
+                    elif field.type not in mapping:
                         cell, notes = field.write_cell(cell)
+                        cell = cell.encode("utf-8")
                     result.append(cell)
                 writer.writerow(result)
 
+    def __write_convert_type(self, type=None):
+
+        # Mapping
+        mapping = {
+            "integer": [0, "F10"],
+            "number": [0, "F10.2"],
+            "datetime": [0, "DATETIME20"],
+            "date": [0, "DATE10"],
+            "time": [0, "TIME8"],
+            "year": [0, "F10"],
+        }
+
+        # Return type
+        if type:
+            return mapping.get(type)
+
+        # Return mapping
+        return mapping
+
     # Delete
 
     def delete_resource(self, name, *, ignore=False):
@@ -263,6 +338,14 @@ class SpssStorage(Storage):
 
 # Internal
 
-DATE_FORMAT = "%Y-%m-%d"
-DATETIME_FORMAT = "%Y-%m-%d %H:%M:%S"
-TIME_FORMAT = "%H:%M:%S.%f"
+FORMAT_READ = {
+    "date": "%Y-%m-%d",
+    "datetime": "%Y-%m-%d %H:%M:%S",
+    "time": "%H:%M:%S.%f",
+}
+
+FORMAT_WRITE = {
+    "date": "%Y-%m-%d",
+    "datetime": "%Y-%m-%d %H:%M:%S",
+    "time": "%H:%M:%S.%f",
+}
diff --git a/frictionless/plugins/sql.py b/frictionless/plugins/sql.py
index cf6b7e2c..5b701082 100644
--- a/frictionless/plugins/sql.py
+++ b/frictionless/plugins/sql.py
@@ -130,13 +130,13 @@ class SqlParser(Parser):
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         sa = helpers.import_from_plugin("sqlalchemy", plugin="sql")
         engine = sa.create_engine(self.resource.source)
         dialect = self.resource.dialect
         schema = self.resource.schema
         storage = SqlStorage(engine=engine)
-        resource = Resource(name=dialect.table, data=row_stream, schema=schema)
+        resource = Resource(name=dialect.table, data=read_row_stream, schema=schema)
         storage.write_resource(resource)
 
 
@@ -194,7 +194,7 @@ class SqlStorage(Storage):
             note = f'Resource "{name}" does not exist'
             raise exceptions.FrictionlessException(errors.StorageError(note=note))
         schema = self.__read_convert_schema(sql_table)
-        data = partial(self.__read_data_stream, name, order_by=order_by)
+        data = partial(self.__read_convert_data, name, order_by=order_by)
         resource = Resource(name=name, schema=schema, data=data)
         return resource
 
@@ -246,25 +246,28 @@ class SqlStorage(Storage):
         # Return schema
         return schema
 
-    def __read_convert_type(self, sql_type):
-        mapping = self.__read_convert_types()
-
-        # Supported type
-        for key, value in mapping.items():
-            if isinstance(sql_type, key):
-                return value
-
-        # Not supported type
-        note = f'Field type "{sql_type}" is not supported'
-        raise exceptions.FrictionlessException(errors.StorageError(note=note))
+    def __read_convert_data(self, name, *, order_by=None):
+        sa = helpers.import_from_plugin("sqlalchemy", plugin="sql")
+        sql_table = self.__read_sql_table(name)
+        with self.__connection.begin():
+            # Streaming could be not working for some backends:
+            # http://docs.sqlalchemy.org/en/latest/core/connections.html
+            select = sql_table.select().execution_options(stream_results=True)
+            if order_by:
+                select = select.order_by(sa.sql.text(order_by))
+            result = select.execute()
+            yield result.keys()
+            for item in result:
+                cells = list(item)
+                yield cells
 
-    def __read_convert_types(self):
+    def __read_convert_type(self, sql_type=None):
         sa = helpers.import_from_plugin("sqlalchemy", plugin="sql")
         sapg = helpers.import_from_plugin("sqlalchemy.dialects.postgresql", plugin="sql")
         sams = helpers.import_from_plugin("sqlalchemy.dialects.mysql", plugin="sql")
 
         # Return mapping
-        return {
+        mapping = {
             sapg.ARRAY: "array",
             sams.BIT: "string",
             sa.Boolean: "boolean",
@@ -283,20 +286,15 @@ class SqlStorage(Storage):
             sapg.UUID: "string",
         }
 
-    def __read_data_stream(self, name, *, order_by=None):
-        sa = helpers.import_from_plugin("sqlalchemy", plugin="sql")
-        sql_table = self.__read_sql_table(name)
-        with self.__connection.begin():
-            # Streaming could be not working for some backends:
-            # http://docs.sqlalchemy.org/en/latest/core/connections.html
-            select = sql_table.select().execution_options(stream_results=True)
-            if order_by:
-                select = select.order_by(sa.sql.text(order_by))
-            result = select.execute()
-            yield result.keys()
-            for item in result:
-                cells = list(item)
-                yield cells
+        # Return type
+        if sql_type:
+            for key, value in mapping.items():
+                if isinstance(sql_type, key):
+                    return value
+            return "string"
+
+        # Return mapping
+        return mapping
 
     def __read_sql_table(self, name):
         sql_name = self.__write_convert_name(name)
@@ -337,7 +335,7 @@ class SqlStorage(Storage):
 
             # Write data
             for resource in package.resources:
-                self.__write_row_stream(resource)
+                self.__write_convert_data(resource)
 
     def __write_convert_name(self, name):
         return self.__prefix + name
@@ -357,8 +355,6 @@ class SqlStorage(Storage):
             checks = []
             nullable = not field.required
             column_type = self.__write_convert_type(field.type)
-            if not column_type:
-                column_type = sa.Text
             unique = field.constraints.get("unique", False)
             for const, value in field.constraints.items():
                 if const == "minLength":
@@ -405,38 +401,44 @@ class SqlStorage(Storage):
         sql_table = sa.Table(sql_name, self.__metadata, *(columns + constraints))
         return sql_table
 
-    def __write_convert_type(self, type):
-        mapping = self.__write_convert_types()
+    def __write_convert_data(self, resource):
 
-        # Supported type
-        if type in mapping:
-            return mapping[type]
+        # Fallback fields
+        fallback_fields = []
+        mapping = self.__write_convert_type()
+        for field in resource.schema.fields:
+            if not mapping.get(field.type):
+                fallback_fields.append(field)
 
-        # Not supported type
-        note = f'Field type "{type}" is not supported'
-        raise exceptions.FrictionlessException(errors.StorageError(note=note))
+        # Write data
+        buffer = []
+        buffer_size = 1000
+        sql_table = self.__read_sql_table(resource.name)
+        for row in resource.read_row_stream():
+            for field in fallback_fields:
+                row[field.name], notes = field.write_cell(row[field.name])
+            buffer.append(row)
+            if len(buffer) > buffer_size:
+                self.__connection.execute(sql_table.insert().values(buffer))
+                buffer = []
+        if len(buffer):
+            self.__connection.execute(sql_table.insert().values(buffer))
 
-    def __write_convert_types(self):
+    def __write_convert_type(self, type=None):
         sa = helpers.import_from_plugin("sqlalchemy", plugin="sql")
         sapg = helpers.import_from_plugin("sqlalchemy.dialects.postgresql", plugin="sql")
 
         # Default dialect
         mapping = {
             "any": sa.Text,
-            "array": None,
             "boolean": sa.Boolean,
             "date": sa.Date,
             "datetime": sa.DateTime,
-            "duration": None,
-            "geojson": None,
-            "geopoint": None,
             "integer": sa.Integer,
             "number": sa.Float,
-            "object": None,
             "string": sa.Text,
             "time": sa.Time,
             "year": sa.Integer,
-            "yearmonth": None,
         }
 
         # Postgresql dialect
@@ -450,30 +452,12 @@ class SqlStorage(Storage):
                 }
             )
 
-        return mapping
+        # Return type
+        if type:
+            return mapping.get(type, sa.Text)
 
-    def __write_row_stream(self, resource):
-
-        # Fallback fields
-        fallback_fields = []
-        mapping = self.__write_convert_types()
-        for field in resource.schema.fields:
-            if mapping[field.type] is None:
-                fallback_fields.append(field)
-
-        # Write data
-        buffer = []
-        buffer_size = 1000
-        sql_table = self.__read_sql_table(resource.name)
-        for row in resource.read_row_stream():
-            for field in fallback_fields:
-                row[field.name], notes = field.write_cell(row[field.name])
-            buffer.append(row)
-            if len(buffer) > buffer_size:
-                self.__connection.execute(sql_table.insert().values(buffer))
-                buffer = []
-        if len(buffer):
-            self.__connection.execute(sql_table.insert().values(buffer))
+        # Return mapping
+        return mapping
 
     # Delete
 
diff --git a/frictionless/plugins/tsv.py b/frictionless/plugins/tsv.py
index a5fc25d6..6d93ac1d 100644
--- a/frictionless/plugins/tsv.py
+++ b/frictionless/plugins/tsv.py
@@ -72,11 +72,11 @@ class TsvParser(Parser):
 
     # Write
 
-    def write(self, row_stream):
+    def write(self, read_row_stream):
         tsv = helpers.import_from_plugin("tsv", plugin="tsv")
         with tempfile.NamedTemporaryFile("wt", delete=False) as file:
             writer = tsv.writer(file)
-            for row in row_stream:
+            for row in read_row_stream():
                 schema = row.schema
                 if row.row_number == 1:
                     writer.writerow(schema.field_names)
diff --git a/frictionless/table.py b/frictionless/table.py
index bb6bf04a..13bdb07f 100644
--- a/frictionless/table.py
+++ b/frictionless/table.py
@@ -772,7 +772,7 @@ class Table:
     # Write
 
     # NOTE: implement proper usage of loaders (e.g. write to s3)
-    # TODO: allow None target and return result for inline/pandas/etc?
+    # NOTE: can we rebase on source/target resources instead of read_row_stream?
     def write(
         self,
         target=None,
@@ -809,11 +809,12 @@ class Table:
         )
 
         # Write file
-        row_stream = self.__write_row_stream_create()
+        read_row_stream = self.__write_row_stream_create
         parser = system.create_parser(resource)
-        parser.write(row_stream)
+        parser.write(read_row_stream)
         return resource.source
 
     def __write_row_stream_create(self):
-        self.__read_data_stream_raise_closed()
-        yield from self.row_stream
+        if self.__row_position:
+            self.open()
+        return self.row_stream


Test patch:
diff --git a/tests/plugins/test_bigquery.py b/tests/plugins/test_bigquery.py
index 27eb4b14..dbb69ef0 100644
--- a/tests/plugins/test_bigquery.py
+++ b/tests/plugins/test_bigquery.py
@@ -22,6 +22,7 @@ pytestmark = pytest.mark.skipif(
 # Parser
 
 
+@pytest.mark.ci
 def test_table_bigquery(options):
     prefix = options.pop("prefix")
     service = options.pop("service")
@@ -34,7 +35,10 @@ def test_table_bigquery(options):
     # Read
     with Table(service, dialect=dialect) as table:
         assert table.header == ["id", "name"]
-        assert table.read_data() == [["1", "english"], ["2", "中国人"]]
+        assert table.read_rows() == [
+            {"id": 1, "name": "english"},
+            {"id": 2, "name": "中国人"},
+        ]
 
 
 # Storage
diff --git a/tests/plugins/test_pandas.py b/tests/plugins/test_pandas.py
index db6dba00..9e035848 100644
--- a/tests/plugins/test_pandas.py
+++ b/tests/plugins/test_pandas.py
@@ -174,24 +174,6 @@ def test_storage_constraints():
     storage.delete_package(target.resource_names)
 
 
-# NOTE: can we add constraints support to Pandas?
-@pytest.mark.skip
-@pytest.mark.parametrize(
-    "field_name, cell",
-    [
-        ("required", ""),
-        ("minLength", "bad"),
-        ("maxLength", "badbadbad"),
-        ("pattern", "bad"),
-        ("enum", "bad"),
-        ("minimum", 3),
-        ("maximum", 9),
-    ],
-)
-def test_storage_constraints_not_valid_error(field_name, cell):
-    pass
-
-
 def test_storage_read_resource_not_existent_error():
     storage = PandasStorage()
     with pytest.raises(exceptions.FrictionlessException) as excinfo:
diff --git a/tests/plugins/test_spss.py b/tests/plugins/test_spss.py
index c1777e82..b04c2a8a 100644
--- a/tests/plugins/test_spss.py
+++ b/tests/plugins/test_spss.py
@@ -1,15 +1,31 @@
 import pytest
-import isodate
 import datetime
-from frictionless import Package, Resource
+from frictionless import Package, Resource, Table
 from frictionless.plugins.spss import SpssStorage, exceptions
 
 
+# Parser
+
+
+def test_table_spss(tmpdir):
+    target = str(tmpdir.join("table.sav"))
+
+    # Write
+    with Table("data/table.csv") as table:
+        table.write(target)
+
+    # Read
+    with Table(target) as table:
+        assert table.header == ["id", "name"]
+        assert table.read_rows() == [
+            {"id": 1, "name": "english"},
+            {"id": 2, "name": "中国人"},
+        ]
+
+
 # Storage
 
 
-# TODO: fix SPSS type mappings
-@pytest.mark.skip
 def test_storage_types(tmpdir):
 
     # Export/Import
@@ -21,21 +37,21 @@ def test_storage_types(tmpdir):
     assert target.get_resource("types").schema == {
         "fields": [
             {"name": "any", "type": "string"},  # type fallback
-            {"name": "array", "type": "array"},
-            {"name": "boolean", "type": "boolean"},
+            {"name": "array", "type": "string"},  # type fallback
+            {"name": "boolean", "type": "string"},  # type fallback
             {"name": "date", "type": "date"},
             {"name": "date_year", "type": "date"},  # format removal
             {"name": "datetime", "type": "datetime"},
-            {"name": "duration", "type": "duration"},
-            {"name": "geojson", "type": "object"},
-            {"name": "geopoint", "type": "array"},
+            {"name": "duration", "type": "string"},  # type fallback
+            {"name": "geojson", "type": "string"},  # type fallback
+            {"name": "geopoint", "type": "string"},  # type fallback
             {"name": "integer", "type": "integer"},
             {"name": "number", "type": "number"},
-            {"name": "object", "type": "object"},
+            {"name": "object", "type": "string"},  # type fallback
             {"name": "string", "type": "string"},
             {"name": "time", "type": "time"},
             {"name": "year", "type": "integer"},  # type downgrade
-            {"name": "yearmonth", "type": "array"},  # type downgrade
+            {"name": "yearmonth", "type": "string"},  # type fallback
         ],
     }
 
@@ -43,21 +59,21 @@ def test_storage_types(tmpdir):
     assert target.get_resource("types").read_rows() == [
         {
             "any": "中国人",
-            "array": ["Mike", "John"],
-            "boolean": True,
+            "array": '["Mike", "John"]',
+            "boolean": "true",
             "date": datetime.date(2015, 1, 1),
             "date_year": datetime.date(2015, 1, 1),
             "datetime": datetime.datetime(2015, 1, 1, 3, 0),
-            "duration": isodate.parse_duration("P1Y1M"),
-            "geojson": {"type": "Point", "coordinates": [33, 33.33]},
-            "geopoint": [30, 70],
+            "duration": "P1Y1M",
+            "geojson": '{"type": "Point", "coordinates": [33, 33.33]}',
+            "geopoint": "30,70",
             "integer": 1,
-            "number": 7,
-            "object": {"chars": 560},
+            "number": 7.0,
+            "object": '{"chars": 560}',
             "string": "english",
             "time": datetime.time(3, 0),
             "year": 2015,
-            "yearmonth": [2015, 1],
+            "yearmonth": "2015-01",
         },
     ]
 
@@ -65,8 +81,6 @@ def test_storage_types(tmpdir):
     storage.delete_package(target.resource_names)
 
 
-# TODO: fix SPSS type mappings
-@pytest.mark.skip
 def test_storage_integrity(tmpdir):
 
     # Export/Import
@@ -78,25 +92,22 @@ def test_storage_integrity(tmpdir):
     assert target.get_resource("integrity_main").schema == {
         "fields": [
             # added required
-            {"name": "id", "type": "integer", "constraints": {"required": True}},
-            {"name": "parent", "type": "number"},  # type downgrade
+            {"name": "id", "type": "integer"},
+            {"name": "parent", "type": "integer"},
             {"name": "description", "type": "string"},
         ],
-        "primaryKey": ["id"],
+        # primary key removal
         # foreign keys removal
     }
 
     # Assert metadata (link)
     assert target.get_resource("integrity_link").schema == {
         "fields": [
-            # added required
-            {"name": "main_id", "type": "integer", "constraints": {"required": True}},
-            # added required; removed unique
-            {"name": "some_id", "type": "integer", "constraints": {"required": True}},
-            # removed unique
-            {"name": "description", "type": "string"},
+            {"name": "main_id", "type": "integer"},
+            {"name": "some_id", "type": "integer"},  # constraint removal
+            {"name": "description", "type": "string"},  # constraint removal
         ],
-        "primaryKey": ["main_id", "some_id"],
+        # primary key removal
         # foreign keys removal
     }
 
@@ -116,8 +127,6 @@ def test_storage_integrity(tmpdir):
     storage.delete_package(target.resource_names)
 
 
-# TODO: fix SPSS type mappings
-@pytest.mark.skip
 def test_storage_constraints(tmpdir):
 
     # Export/Import
@@ -155,24 +164,6 @@ def test_storage_constraints(tmpdir):
     storage.delete_package(target.resource_names)
 
 
-# NOTE: can we add consratins support to SPSS?
-@pytest.mark.skip
-@pytest.mark.parametrize(
-    "field_name, cell",
-    [
-        ("required", ""),
-        ("minLength", "bad"),
-        ("maxLength", "badbadbad"),
-        ("pattern", "bad"),
-        ("enum", "bad"),
-        ("minimum", 3),
-        ("maximum", 9),
-    ],
-)
-def test_storage_constraints_not_valid_error(field_name, cell):
-    pass
-
-
 def test_storage_read_resource_not_existent_error():
     storage = SpssStorage()
     with pytest.raises(exceptions.FrictionlessException) as excinfo: