Repository: frictionlessdata/frictionless-py

Issue:
Migrate CKAN plugin
# Overview

For the CKAN plugin, we need to write a `plugins.ckan.CkanStorage class` which already exists in the repository but we need to update to the latest `Storage` interface requirements:

```
class Storage:
    __init__(self, **options):
    __iter__(self):
   read_resource(self, name, **options)
   read_package(self, **options)
   write_resource(self, resource, *, force=False, **options)
   write_package(self, package, *, force=False, **options)
   delete_resource(self, name, *, ignore=False, **options)
   delete_package(self, names, *, ignore=False, **options)
```

We already have implemented:
- `plugins.sql.SqlStorage` (more complex as it supports foreign keys)
- `plugins.pandas.PandasStorage`

So it should be relatively easy to update `CkanStorage` based on the reference interface and 2 existent implementations.

We also need:
- update `ckanPlugin.create_storage`
- add `resource.from_ckan/to_ckan`
- add `package.from_ckan/to_ckan`

For testing we use a reference package that we should be able to export and then import back validating results, for example, for pandas:

```python
def test_storage_types():

    # Export/Import
    source = Package("data/storage/types.json")
    storage = source.to_pandas()
    target = Package.from_pandas(dataframes=storage.dataframes)

    # Assert metadata
    assert target.get_resource("main").schema == {
        "fields": [
            {"name": "any", "type": "string"},  # type fallback
            {"name": "array", "type": "array"},
            {"name": "boolean", "type": "boolean"},
            {"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": "integer", "type": "integer"},
            {"name": "number", "type": "number"},
            {"name": "object", "type": "object"},
            {"name": "string", "type": "string"},
            {"name": "time", "type": "time"},
            {"name": "year", "type": "integer"},  # type downgrade
            {"name": "yearmonth", "type": "array"},  # type downgrade
        ],
    }

    # Assert data
    assert target.get_resource("main").read_rows() == [
        {
            "any": "note1",
            "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],
            "integer": 1,
            "number": 7,
            "object": {"chars": 560},
            "string": "good",
            "time": datetime.time(3, 0),
            "year": 2015,
            "yearmonth": [2015, 1],
        },
    ]

    # Cleanup storage
    storage.delete_package(target.resource_names)


def test_storage_integrity():

    # Export/Import
    source = Package("data/storage/integrity.json")
    storage = source.to_pandas()
    target = Package.from_pandas(dataframes=storage.dataframes)

    # Assert metadata (main)
    assert target.get_resource("main").schema == {
        "fields": [
            # added required
            {"name": "id", "type": "integer", "constraints": {"required": True}},
            {"name": "parent", "type": "number"},  # type downgrade
            {"name": "description", "type": "string"},
        ],
        "primaryKey": ["id"],
        # foreign keys removal
    }

    # Assert metadata (link)
    assert target.get_resource("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"},
        ],
        "primaryKey": ["main_id", "some_id"],
        # foreign keys removal
    }

    # Assert data (main)
    assert target.get_resource("main").read_rows() == [
        {"id": 1, "parent": None, "description": "english"},
        {"id": 2, "parent": 1, "description": "中国人"},
    ]

    # Assert data (link)
    assert target.get_resource("link").read_rows() == [
        {"main_id": 1, "some_id": 1, "description": "note1"},
        {"main_id": 2, "some_id": 2, "description": "note2"},
    ]

    # Cleanup storage
    storage.delete_package(target.resource_names)


def test_storage_constraints():

    # Export/Import
    source = Package("data/storage/constraints.json")
    storage = source.to_pandas()
    target = Package.from_pandas(dataframes=storage.dataframes)

    # Assert metadata
    assert target.get_resource("main").schema == {
        "fields": [
            {"name": "required", "type": "string"},  # constraint removal
            {"name": "minLength", "type": "string"},  # constraint removal
            {"name": "maxLength", "type": "string"},  # constraint removal
            {"name": "pattern", "type": "string"},  # constraint removal
            {"name": "enum", "type": "string"},  # constraint removal
            {"name": "minimum", "type": "integer"},  # constraint removal
            {"name": "maximum", "type": "integer"},  # constraint removal
        ],
    }

    # Assert data
    assert target.get_resource("main").read_rows() == [
        {
            "required": "passing",
            "minLength": "passing",
            "maxLength": "passing",
            "pattern": "passing",
            "enum": "passing",
            "minimum": 5,
            "maximum": 5,
        },
    ]

    # Cleanup storage
    storage.delete_package(target.resource_names)


# NOTE: can we add consratins 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:
        storage.read_resource("bad")
    error = excinfo.value.error
    assert error.code == "storage-error"
    assert error.note.count("does not exist")


def test_storage_write_resource_existent_error():
    resource = Resource(path="data/table.csv")
    storage = resource.to_pandas()
    with pytest.raises(exceptions.FrictionlessException) as excinfo:
        storage.write_resource(resource)
    error = excinfo.value.error
    assert error.code == "storage-error"
    assert error.note.count("already exists")
    # Cleanup storage
    storage.delete_package(list(storage))


def test_storage_delete_resource_not_existent_error():
    storage = PandasStorage()
    with pytest.raises(exceptions.FrictionlessException) as excinfo:
        storage.delete_resource("bad")
    error = excinfo.value.error
    assert error.code == "storage-error"
    assert error.note.count("does not exist")
```

Every plugin has its own limitations so in the code snippet above we note type fallbacks/downgrades and metadata removals.

# References

- how it works in the prev software generation - https://frictionlessdata.io/how-to/how-to-use-data-packages-with-ckan/ (note TODO in step 2)
- https://github.com/frictionlessdata/frictionless-py/issues/413 (an issue created after the mentioned TODO)
- https://github.com/frictionlessdata/datapackage-pipelines-ckan-driver
- see "Contributing" and other tech documents in the root of the repo

# Tasks

- [ ] migrating the plugin
- [ ] fixing the naming problem from the original implementation
- [ ] write a guide

Discussion:

Patch:
diff --git a/data/ckan_mock_responses/ckan_error.json b/data/ckan_mock_responses/ckan_error.json
new file mode 100644
index 00000000..0699473b
--- /dev/null
+++ b/data/ckan_mock_responses/ckan_error.json
@@ -0,0 +1,8 @@
+{
+  "help": "https://demo.ckan.org/api/3/action/help_show?name=package_show", 
+  "success": false, 
+  "error": {
+    "message": "Not found", 
+    "__type": "Not Found Error"
+  }
+}
diff --git a/data/ckan_mock_responses/datastore_search_describe.json b/data/ckan_mock_responses/datastore_search_describe.json
new file mode 100644
index 00000000..df292923
--- /dev/null
+++ b/data/ckan_mock_responses/datastore_search_describe.json
@@ -0,0 +1,67 @@
+{
+  "help": "https://demo.ckan.org/api/3/action/help_show?name=datastore_search",
+  "success": true,
+  "result": {
+    "resource_id": "79843e49-7974-411c-8eb5-fb2d1111d707",
+    "fields": [
+      {
+        "type": "int4",
+        "id": "_id"
+      },
+      {
+        "type": "int4",
+        "id": "id"
+      },
+      {
+        "type": "int4",
+        "id": "parent"
+      },
+      {
+        "type": "text",
+        "id": "name"
+      },
+      {
+        "type": "bool",
+        "id": "current"
+      },
+      {
+        "type": "float8",
+        "id": "rating"
+      },
+      {
+        "type": "int4",
+        "id": "created_year"
+      },
+      {
+        "type": "date",
+        "id": "created_date"
+      },
+      {
+        "type": "time",
+        "id": "created_time"
+      },
+      {
+        "type": "timestamp",
+        "id": "created_datetime"
+      },
+      {
+        "type": "json",
+        "id": "stats"
+      },
+      {
+        "type": "_text",
+        "id": "persons"
+      },
+      {
+        "type": "json",
+        "id": "location"
+      }
+    ],
+    "records": [],
+    "_links": {
+      "start": "/api/3/action/datastore_search?limit=0&resource_id=79843e49-7974-411c-8eb5-fb2d1111d707",
+      "next": "/api/3/action/datastore_search?offset=0&limit=0&resource_id=79843e49-7974-411c-8eb5-fb2d1111d707"
+    },
+    "limit": 0
+  }
+}
diff --git a/data/ckan_mock_responses/datastore_search_rows.json b/data/ckan_mock_responses/datastore_search_rows.json
new file mode 100644
index 00000000..65d35fb8
--- /dev/null
+++ b/data/ckan_mock_responses/datastore_search_rows.json
@@ -0,0 +1,137 @@
+{
+  "help": "https://demo.ckan.org/api/3/action/help_show?name=datastore_search",
+  "success": true,
+  "result": {
+    "resource_id": "79843e49-7974-411c-8eb5-fb2d1111d707",
+    "fields": [
+      {
+        "type": "int4",
+        "id": "_id"
+      },
+      {
+        "type": "int4",
+        "id": "id"
+      },
+      {
+        "type": "int4",
+        "id": "parent"
+      },
+      {
+        "type": "text",
+        "id": "name"
+      },
+      {
+        "type": "bool",
+        "id": "current"
+      },
+      {
+        "type": "float8",
+        "id": "rating"
+      },
+      {
+        "type": "int4",
+        "id": "created_year"
+      },
+      {
+        "type": "date",
+        "id": "created_date"
+      },
+      {
+        "type": "time",
+        "id": "created_time"
+      },
+      {
+        "type": "timestamp",
+        "id": "created_datetime"
+      },
+      {
+        "type": "json",
+        "id": "stats"
+      },
+      {
+        "type": "_text",
+        "id": "persons"
+      },
+      {
+        "type": "json",
+        "id": "location"
+      }
+    ],
+    "records": [
+      {
+        "rating": 9.5,
+        "stats": {
+          "chars": 560,
+          "height": 54.8
+        },
+        "name": "Taxes",
+        "parent": null,
+        "created_datetime": "2015-01-01T03:00:00",
+        "created_year": 2015,
+        "current": true,
+        "persons": [
+          "mike",
+          "alice"
+        ],
+        "location": {
+          "type": "Point",
+          "coordinates": [
+            50.0,
+            50.0
+          ]
+        },
+        "created_date": "2015-01-01",
+        "created_time": "03:00:00",
+        "_id": 1,
+        "id": 1
+      },
+      {
+        "rating": 7.0,
+        "stats": {
+          "chars": 970,
+          "height": 40
+        },
+        "name": "中国人",
+        "parent": 1,
+        "created_datetime": "2015-12-31T15:45:33",
+        "created_year": 2015,
+        "current": false,
+        "persons": [
+          "chen",
+          "alice"
+        ],
+        "location": {
+          "type": "Point",
+          "coordinates": [
+            33.33,
+            33.33
+          ]
+        },
+        "created_date": "2015-12-31",
+        "created_time": "15:45:33",
+        "_id": 2,
+        "id": 2
+      },
+      {
+        "rating": null,
+        "stats": null,
+        "name": "",
+        "parent": 1,
+        "created_datetime": null,
+        "created_year": null,
+        "current": false,
+        "persons": null,
+        "location": null,
+        "created_date": null,
+        "created_time": null,
+        "_id": 3,
+        "id": 3
+      }
+    ],
+    "_links": {
+      "start": "/api/3/action/datastore_search?resource_id=79843e49-7974-411c-8eb5-fb2d1111d707",
+      "next": "/api/3/action/datastore_search?offset=100&resource_id=79843e49-7974-411c-8eb5-fb2d1111d707"
+    },
+    "total": 3
+  }
+}
diff --git a/data/ckan_mock_responses/datastore_search_rows_empty.json b/data/ckan_mock_responses/datastore_search_rows_empty.json
new file mode 100644
index 00000000..c8389ee6
--- /dev/null
+++ b/data/ckan_mock_responses/datastore_search_rows_empty.json
@@ -0,0 +1,67 @@
+{
+  "help": "https://demo.ckan.org/api/3/action/help_show?name=datastore_search",
+  "success": true,
+  "result": {
+    "resource_id": "79843e49-7974-411c-8eb5-fb2d1111d707",
+    "fields": [
+      {
+        "type": "int4",
+        "id": "_id"
+      },
+      {
+        "type": "int4",
+        "id": "id"
+      },
+      {
+        "type": "int4",
+        "id": "parent"
+      },
+      {
+        "type": "text",
+        "id": "name"
+      },
+      {
+        "type": "bool",
+        "id": "current"
+      },
+      {
+        "type": "float8",
+        "id": "rating"
+      },
+      {
+        "type": "int4",
+        "id": "created_year"
+      },
+      {
+        "type": "date",
+        "id": "created_date"
+      },
+      {
+        "type": "time",
+        "id": "created_time"
+      },
+      {
+        "type": "timestamp",
+        "id": "created_datetime"
+      },
+      {
+        "type": "json",
+        "id": "stats"
+      },
+      {
+        "type": "_text",
+        "id": "persons"
+      },
+      {
+        "type": "json",
+        "id": "location"
+      }
+    ],
+    "records": [],
+    "_links": {
+      "start": "/api/3/action/datastore_search?resource_id=79843e49-7974-411c-8eb5-fb2d1111d707",
+      "next": "/api/3/action/datastore_search?offset=200&resource_id=79843e49-7974-411c-8eb5-fb2d1111d707"
+    },
+    "offset": 100
+  }
+}
diff --git a/data/ckan_mock_responses/datastore_search_table_metadata_01.json b/data/ckan_mock_responses/datastore_search_table_metadata_01.json
new file mode 100644
index 00000000..85f9a7e1
--- /dev/null
+++ b/data/ckan_mock_responses/datastore_search_table_metadata_01.json
@@ -0,0 +1,50 @@
+{
+  "help": "https://demo.ckan.org/api/3/action/help_show?name=datastore_search",
+  "success": true,
+  "result": {
+    "resource_id": "_table_metadata",
+    "fields": [
+      {
+        "type": "text",
+        "id": "_id"
+      },
+      {
+        "type": "name",
+        "id": "name"
+      },
+      {
+        "type": "oid",
+        "id": "oid"
+      },
+      {
+        "type": "name",
+        "id": "alias_of"
+      }
+    ],
+    "records": [
+      {
+        "_id": "127e87deb7b16215",
+        "alias_of": null,
+        "name": "bd79c992-40f0-454a-a0ff-887f84a792fb",
+        "oid": 14788505
+      },
+      {
+        "_id": "d16cf8659bb85f02",
+        "alias_of": null,
+        "name": "79843e49-7974-411c-8eb5-fb2d1111d707",
+        "oid": 14787229
+      }
+    ],
+    "_links": {
+      "start": "/api/3/action/datastore_search?filters=%7B%22name%22%3A+%5B%22bd79c992-40f0-454a-a0ff-887f84a792fb%22%2C+%2279843e49-7974-411c-8eb5-fb2d1111d707%22%5D%7D&resource_id=_table_metadata",
+      "next": "/api/3/action/datastore_search?offset=100&filters=%7B%22name%22%3A+%5B%22bd79c992-40f0-454a-a0ff-887f84a792fb%22%2C+%2279843e49-7974-411c-8eb5-fb2d1111d707%22%5D%7D&resource_id=_table_metadata"
+    },
+    "filters": {
+      "name": [
+        "bd79c992-40f0-454a-a0ff-887f84a792fb",
+        "79843e49-7974-411c-8eb5-fb2d1111d707"
+      ]
+    },
+    "total": 2
+  }
+}
diff --git a/data/ckan_mock_responses/datastore_search_table_metadata_02.json b/data/ckan_mock_responses/datastore_search_table_metadata_02.json
new file mode 100644
index 00000000..d3f5ff67
--- /dev/null
+++ b/data/ckan_mock_responses/datastore_search_table_metadata_02.json
@@ -0,0 +1,37 @@
+{
+  "help": "https://demo.ckan.org/api/3/action/help_show?name=datastore_search",
+  "success": true,
+  "result": {
+    "resource_id": "_table_metadata",
+    "fields": [
+      {
+        "type": "text",
+        "id": "_id"
+      },
+      {
+        "type": "name",
+        "id": "name"
+      },
+      {
+        "type": "oid",
+        "id": "oid"
+      },
+      {
+        "type": "name",
+        "id": "alias_of"
+      }
+    ],
+    "records": [],
+    "_links": {
+      "start": "/api/3/action/datastore_search?filters=%7B%22name%22%3A+%5B%22bd79c992-40f0-454a-a0ff-887f84a792fb%22%2C+%2279843e49-7974-411c-8eb5-fb2d1111d707%22%5D%7D&resource_id=_table_metadata",
+      "next": "/api/3/action/datastore_search?offset=200&filters=%7B%22name%22%3A+%5B%22bd79c992-40f0-454a-a0ff-887f84a792fb%22%2C+%2279843e49-7974-411c-8eb5-fb2d1111d707%22%5D%7D&resource_id=_table_metadata"
+    },
+    "filters": {
+      "name": [
+        "bd79c992-40f0-454a-a0ff-887f84a792fb",
+        "79843e49-7974-411c-8eb5-fb2d1111d707"
+      ]
+    },
+    "offset": 100
+  }
+}
diff --git a/data/ckan_mock_responses/doesnt_matter.json b/data/ckan_mock_responses/doesnt_matter.json
new file mode 100644
index 00000000..e8dba443
--- /dev/null
+++ b/data/ckan_mock_responses/doesnt_matter.json
@@ -0,0 +1,4 @@
+{
+  "success": true,
+  "comment": "Use this mock when it doesn't matter what is in the response. If we're just looking for a 200 OK status and success: true this mock will do."
+}
diff --git a/data/ckan_mock_responses/package_show.json b/data/ckan_mock_responses/package_show.json
new file mode 100644
index 00000000..cd5bcffc
--- /dev/null
+++ b/data/ckan_mock_responses/package_show.json
@@ -0,0 +1,103 @@
+{
+  "help": "https://demo.ckan.org/api/3/action/help_show?name=package_show",
+  "success": true,
+  "result": {
+    "license_title": "",
+    "maintainer": "",
+    "relationships_as_object": [],
+    "private": false,
+    "maintainer_email": "",
+    "num_tags": 0,
+    "id": "7766839b-face-4336-8e1a-3c51c5e7634d",
+    "metadata_created": "2017-10-13T14:00:49.983285",
+    "metadata_modified": "2017-10-18T14:44:12.587015",
+    "author": "",
+    "author_email": "",
+    "state": "active",
+    "version": "",
+    "creator_user_id": "9cdbdaf8-b7e9-468b-bd72-4b23a07b9066",
+    "type": "dataset",
+    "resources": [
+      {
+        "mimetype": null,
+        "cache_url": null,
+        "hash": "",
+        "description": "",
+        "name": "co2-mm-mlo_csv_not_streamed",
+        "format": "CSV",
+        "url": "https://pkgstore.datahub.io/core/co2-ppm:co2-mm-mlo_csv/data/co2-mm-mlo_csv.csv",
+        "datastore_active": true,
+        "cache_last_updated": null,
+        "package_id": "7766839b-face-4336-8e1a-3c51c5e7634d",
+        "created": "2017-10-18T14:44:10.636629",
+        "state": "active",
+        "mimetype_inner": null,
+        "last_modified": null,
+        "position": 0,
+        "revision_id": "99b5f272-aaca-4f69-9b74-b64432b27cde",
+        "url_type": null,
+        "id": "bd79c992-40f0-454a-a0ff-887f84a792fb",
+        "resource_type": null,
+        "size": null
+      },
+      {
+        "mimetype": null,
+        "cache_url": null,
+        "hash": "d83fd70eeef26cbc33904cf84664d99c",
+        "description": "",
+        "name": "co2-mm-mlo_csv_streamed",
+        "encoding": "utf-8",
+        "url": "https://demo.ckan.org/dataset/7766839b-face-4336-8e1a-3c51c5e7634d/resource/79843e49-7974-411c-8eb5-fb2d1111d707/download/co2-mm-mlo_csv_streamed.csv",
+        "datastore_active": true,
+        "format": "CSV",
+        "cache_last_updated": null,
+        "package_id": "7766839b-face-4336-8e1a-3c51c5e7634d",
+        "created": "2017-10-18T14:44:12.692356",
+        "state": "active",
+        "mimetype_inner": null,
+        "last_modified": null,
+        "position": 1,
+        "revision_id": "99b5f272-aaca-4f69-9b74-b64432b27cde",
+        "url_type": "upload",
+        "id": "79843e49-7974-411c-8eb5-fb2d1111d707",
+        "resource_type": null,
+        "size": null
+      }
+    ],
+    "num_resources": 2,
+    "tags": [],
+    "groups": [],
+    "license_id": "",
+    "relationships_as_subject": [],
+    "organization": {
+      "description": "",
+      "created": "2017-10-13T14:00:40.959338",
+      "title": "My Test Org",
+      "name": "my-test-org",
+      "is_organization": true,
+      "state": "active",
+      "image_url": "",
+      "revision_id": "bf91def8-2b83-41a8-98dd-6a27707fdba4",
+      "type": "organization",
+      "id": "4c0827c9-8eb0-4ec0-a95c-28491bceeb6f",
+      "approval_status": "approved"
+    },
+    "name": "test-dataset-010203",
+    "isopen": false,
+    "url": "",
+    "notes": "",
+    "owner_org": "4c0827c9-8eb0-4ec0-a95c-28491bceeb6f",
+    "extras": [
+      {
+        "key": "hash",
+        "value": "ae07f1ac9c7960e614f74d4160d1bf80"
+      },
+      {
+        "key": "profile",
+        "value": "data-package"
+      }
+    ],
+    "title": "Test Dataset",
+    "revision_id": "70564e27-9576-4c15-9dc2-6d57791f35da"
+  }
+}
diff --git a/frictionless/package.py b/frictionless/package.py
index 3432f628..2c8592da 100644
--- a/frictionless/package.py
+++ b/frictionless/package.py
@@ -284,6 +284,24 @@ class Package(Metadata):
         """
         return storage.read_package()
 
+    @staticmethod
+    def from_ckan(*, base_url, dataset_id, api_key=None):
+        """Import package from CKAN
+
+        Parameters:
+            base_url (str): (required) URL for CKAN instance (e.g: https://demo.ckan.org/ )
+            dataset_id (str): (required) ID or slug of dataset to fetch
+            api_key (str): (optional) Your CKAN API key
+        """
+        return Package.from_storage(
+            system.create_storage(
+                "ckan_datastore",
+                base_url=base_url,
+                dataset_id=dataset_id,
+                api_key=api_key,
+            )
+        )
+
     @staticmethod
     def from_sql(*, engine, prefix="", namespace=None):
         """Import package from SQL
@@ -396,6 +414,25 @@ class Package(Metadata):
         storage.write_package(self.to_copy(), force=force)
         return storage
 
+    def to_ckan(self, *, base_url, dataset_id=None, api_key=None, force=False):
+        """Export package to CKAN
+
+        Parameters:
+            base_url (str): (required) URL for CKAN instance (e.g: https://demo.ckan.org/ )
+            dataset_id (str): (optional) ID or slug of dataset this resource belongs to
+            api_key (str): (optional) Your CKAN API key
+            force (bool): (optional) overwrite existing data
+        """
+        return self.to_storage(
+            system.create_storage(
+                "ckan_datastore",
+                base_url=base_url,
+                dataset_id=dataset_id,
+                api_key=api_key,
+            ),
+            force=force,
+        )
+
     def to_sql(self, *, engine, prefix="", namespace=None, force=False):
         """Export package to SQL
 
diff --git a/frictionless/plugins/ckan.py b/frictionless/plugins/ckan.py
index f9db72e1..a0307965 100644
--- a/frictionless/plugins/ckan.py
+++ b/frictionless/plugins/ckan.py
@@ -1,14 +1,17 @@
-import os
 import json
 import logging
+import os
+from functools import partial
+
 import requests
+
+from .. import errors, exceptions
 from ..field import Field
-from ..schema import Schema
+from ..package import Package
 from ..plugin import Plugin
-from ..storage import Storage
 from ..resource import Resource
-from .. import exceptions
-from .. import errors
+from ..schema import Schema
+from ..storage import Storage
 
 log = logging.getLogger(__name__)
 
@@ -31,7 +34,8 @@ class CkanPlugin(Plugin):
         pass
 
     def create_storage(self, name, **options):
-        pass
+        if name == "ckan_datastore":
+            return CkanStorage(**options)
 
 
 # TODO: implement Dialect
@@ -59,10 +63,12 @@ class CkanStorage(Storage):
         self.__bucket_cache = None
         self.__tables = {}
 
+    def __iter__(self):
+        return iter(self.__bucket_cache)
+
     # Read
 
-    @property
-    def read_table_names(self):
+    def _read_table_names(self):
         if self.__bucket_cache:
             return self.__bucket_cache
         params = {"resource_id": "_table_metadata"}
@@ -83,20 +89,21 @@ class CkanStorage(Storage):
             if count == self.__max_pages:
                 log.warn("Max name count exceeded. {} names returned.".format(len(names)))
                 break
-        self.__name_cache = names
+        self.__bucket_cache = names
         return names
 
-    def read_table(self, name):
-        table = self.__table.get(name)
+    def _read_table(self, name):
+        table = self.__tables.get(name)
         if table is None:
             datastore_search_url = "{}/datastore_search".format(self.__base_endpoint)
             params = {"limit": 0, "resource_id": name}
             response = self.__make_ckan_request(datastore_search_url, params=params)
             ckan_table = response["result"]["fields"]
-            table = self.read_table_convert_table(name, ckan_table)
+            table = self._read_table_convert_table(name, ckan_table)
+        self.__tables[name] = table
         return table
 
-    def read_table_convert_table(self, name, ckan_table):
+    def _read_table_convert_table(self, name, ckan_table):
         schema = Schema()
 
         # Fields
@@ -106,17 +113,20 @@ class CkanStorage(Storage):
                 continue
             datastore_type = f["type"]
             datastore_id = f["id"]
-            ts_type, ts_format = self.datastore_field_type_to_schema_type(datastore_type)
+            ts_type, ts_format = self._read_table_convert_field_type(datastore_type)
             field = Field(name=datastore_id, type=ts_type)
             if ts_format is not None:
                 field.format = ts_format
             schema.fields.append(field)
 
         # Table
-        table = Resource(name=name, schema=schema)
+        data = partial(self._read_table_row_stream, name)
+        table = Resource(
+            name=name, schema=schema, data=data(), dialect={"keys": schema.field_names}
+        )
         return table
 
-    def read_table_convert_field_type(self, dstore_type):
+    def _read_table_convert_field_type(self, dstore_type):
         dstore_type = dstore_type.rstrip("0123456789")
         if dstore_type.startswith("_"):
             dstore_type = "array"
@@ -150,70 +160,53 @@ class CkanStorage(Storage):
             )
             return ("string", None)
 
-    def read_table_row_stream(self, name):
-        table = self.read_table(name)
+    def _read_table_row_stream(self, name):
+        table = self._read_table(name)
         datastore_search_url = "{}/datastore_search".format(self.__base_endpoint)
         params = {"resource_id": table.name}
         response = self.__make_ckan_request(datastore_search_url, params=params)
         while response["result"]["records"]:
             for row in response["result"]["records"]:
-                #  row = self.__mapper.restore_row(row, schema=table.schema)
                 yield row
             next_url = self.__base_url + response["result"]["_links"]["next"]
             response = self.__make_ckan_request(next_url)
 
-    # Write
+    def read_resource(self, name, **options):
+        return self._read_table(name)
+
+    def read_package(self, **options):
+        self._read_table_names()
+        package = Package()
+        for name in self:
+            resource = self.read_resource(name)
+            package.resources.append(resource)
+        return package
 
-    def write_table(self, *tables, force=False):
+    # Write
 
+    def _write_table(self, table, force=False):
         # Check for existence
-        for table in tables:
-            if table.name in self.read_table_names():
-                if not force:
-                    note = f'Table "{table.name}" already exists'
-                    raise exceptions.FrictionlessException(errors.StorageError(note=note))
-                self.write_table_remove(table.name)
+        if table.name in self._read_table_names():
+            if not force:
+                note = f'Table "{table.name}" already exists'
+                raise exceptions.FrictionlessException(errors.StorageError(note=note))
+            self._write_table_remove(table.name)
 
         # Define tables
-        for table in tables:
-            self.__tables[table.name] = table
-            datastore_dict = self.write_table_convert_table(table)
-            datastore_url = "{}/datastore_create".format(self.__base_endpoint)
-            self.__make_ckan_request(datastore_url, method="POST", json=datastore_dict)
+        self.__tables[table.name] = table
+        datastore_dict = self._write_table_convert_table(table)
+        datastore_url = "{}/datastore_create".format(self.__base_endpoint)
+        self.__make_ckan_request(datastore_url, method="POST", json=datastore_dict)
 
         # Invalidate cache
         self.__bucket_cache = None
 
-    def write_table_remove(self, *names, ignore=False):
-
-        # Remove tables
-        for name in names:
-
-            # Check existent
-            if name not in self.read_table_names():
-                if not ignore:
-                    note = f'Table "{name}" already exists'
-                    raise exceptions.FrictionlessException(errors.StorageError(note=note))
-                continue
-
-            # Remove from table
-            if name in self.__tables:
-                del self.__tables[name]
-
-            # Remove from ckan
-            datastore_delete_url = "{}/datastore_delete".format(self.__base_endpoint)
-            params = {"resource_id": name, "force": True}
-            self.__make_ckan_request(datastore_delete_url, method="POST", json=params)
-
-        # Invalidate cache
-        self.__bucket_cache = None
-
-    def write_table_convert_table(self, table):
+    def _write_table_convert_table(self, table):
         schema = table.schema
         datastore_dict = {"fields": [], "resource_id": table.name, "force": True}
         for field in schema.fields:
             datastore_field = {"id": field.name}
-            datastore_type = self.descritor_type_to_datastore_type(field.type)
+            datastore_type = self._write_table_convert_field_type(field.type)
             if datastore_type:
                 datastore_field["type"] = datastore_type
             datastore_dict["fields"].append(datastore_field)
@@ -221,7 +214,7 @@ class CkanStorage(Storage):
             datastore_dict["primary_key"] = schema.primary_key
         return datastore_dict
 
-    def write_table_convert_field_type(self, type):
+    def _write_table_convert_field_type(self, type):
         DESCRIPTOR_TYPE_MAPPING = {
             "number": "float",
             "string": "text",
@@ -230,32 +223,61 @@ class CkanStorage(Storage):
             "object": "json",
             "array": "text[]",
             "geojson": "json",
-            "date": "date",
+            "date": "text",
             "time": "time",
             "year": "int",
             "datetime": "timestamp",
         }
-        try:
-            return DESCRIPTOR_TYPE_MAPPING[type]
-        except KeyError:
-            log.warn("Unsupported descriptor type '{}'.".format(type))
-            return None
+        return DESCRIPTOR_TYPE_MAPPING.get(type, "text")
 
-    def write_table_row_stream(self, name, row_stream, method="upsert"):
-        table = self.read_table(name)
+    def _write_table_row_stream(self, name, row_stream):
+        table = self._read_table(name)
         datastore_upsert_url = "{}/datastore_upsert".format(self.__base_endpoint)
-        records = []
-        for r in row_stream:
-            #  r = self.__mapper.convert_row(r, table.eschema)
-            records.append(r)
+        records = [r.to_dict(json=True) for r in row_stream]
         params = {
             "resource_id": table.name,
-            "method": method,
+            "method": "insert",
             "force": True,
             "records": records,
         }
         self.__make_ckan_request(datastore_upsert_url, method="POST", json=params)
 
+    def write_resource(self, resource, *, force=False):
+        self._write_table(resource, force=force)
+        self._write_table_row_stream(resource.name, resource.read_row_stream())
+
+    def write_package(self, package, *, force=False):
+        for resource in package.resources:
+            self.write_resource(resource, force=force)
+
+    # Delete
+
+    def _write_table_remove(self, name, ignore=False):
+        # Check existent
+        if name not in self._read_table_names():
+            if not ignore:
+                note = f'Table "{name}" does not exist'
+                raise exceptions.FrictionlessException(errors.StorageError(note=note))
+
+        # Remove from ckan
+        datastore_delete_url = "{}/datastore_delete".format(self.__base_endpoint)
+        params = {"resource_id": name, "force": True}
+        self.__make_ckan_request(datastore_delete_url, method="POST", json=params)
+
+        # Remove from table
+        if name in self.__tables:
+            del self.__tables[name]
+
+        # Invalidate cache
+        self.__bucket_cache = None
+
+    def delete_resource(self, name, *, ignore=False):
+        self._write_table_remove(name, ignore=ignore)
+
+    def delete_package(self, names, *, ignore=False):
+        for name in names:
+            self._write_table_remove(name, ignore=ignore)
+
     # Private
 
     def __get_resource_ids_for_dataset(self, dataset_id):
diff --git a/frictionless/resource.py b/frictionless/resource.py
index 611e8872..fd8ce7a9 100644
--- a/frictionless/resource.py
+++ b/frictionless/resource.py
@@ -635,6 +635,24 @@ class Resource(Metadata):
         """
         return storage.read_resource(name)
 
+    @staticmethod
+    def from_ckan(*, base_url, resource_id, api_key=None):
+        """Import resource from CKAN
+
+        Parameters:
+            base_url (str): (required) URL for CKAN instance (e.g: https://demo.ckan.org/ )
+            resource_id (str): (required) ID of resource to fetch
+            api_key (str): (optional) Your CKAN API key
+        """
+        return Resource.from_storage(
+            system.create_storage(
+                "ckan_datastore",
+                base_url=base_url,
+                api_key=api_key,
+            ),
+            name=resource_id,
+        )
+
     @staticmethod
     def from_sql(*, name, engine, prefix="", namespace=None):
         """Import resource from SQL table
@@ -817,6 +835,25 @@ class Resource(Metadata):
         storage.write_resource(self.to_copy(), force=force)
         return storage
 
+    def to_ckan(self, *, base_url, dataset_id=None, api_key=None, force=False):
+        """Export resource to CKAN
+
+        Parameters:
+            base_url (str): (required) URL for CKAN instance (e.g: https://demo.ckan.org/ )
+            dataset_id (str): (optional) ID or slug of dataset this resource belongs to
+            api_key (str): (optional) Your CKAN API key
+            force (bool): (optional) overwrite existing data
+        """
+        return self.to_storage(
+            system.create_storage(
+                "ckan_datastore",
+                base_url=base_url,
+                dataset_id=dataset_id,
+                api_key=api_key,
+            ),
+            force=force,
+        )
+
     def to_sql(self, *, engine, prefix="", namespace=None, force=False):
         """Export resource to SQL table
 
diff --git a/frictionless/row.py b/frictionless/row.py
index fa54ce24..fe900e8b 100644
--- a/frictionless/row.py
+++ b/frictionless/row.py
@@ -1,5 +1,6 @@
 from itertools import zip_longest
 from collections import OrderedDict
+from decimal import Decimal
 from .helpers import cached_property
 from .parsers import JsonParser
 from . import errors
@@ -222,6 +223,8 @@ class Row(OrderedDict):
                     cell = self[field.name]
                     if field.type not in JsonParser.native_types:
                         cell, notes = field.write_cell(cell, ignore_missing=True)
+                    if isinstance(cell, Decimal):
+                        cell = float(cell)
                     result[field.name] = cell
             return result
         return dict(self)
@@ -241,6 +244,8 @@ class Row(OrderedDict):
                     cell = self[field.name]
                     if field.type not in JsonParser.native_types:
                         cell, notes = field.write_cell(cell, ignore_missing=True)
+                    if isinstance(cell, Decimal):
+                        cell = float(cell)
                     result.append(cell)
             return result
         return list(self.values())
diff --git a/setup.py b/setup.py
index 06c36d01..f1ca59d0 100644
--- a/setup.py
+++ b/setup.py
@@ -31,7 +31,7 @@ TESTS_REQUIRE = [
     "docstring-parser",
     "python-dotenv",
     "oauth2client",
-    "responses",
+    "responses==0.12.0",
     "coveralls",
     "nbconvert",
     "ipython",


Test patch:
diff --git a/tests/plugins/test_ckan.py b/tests/plugins/test_ckan.py
new file mode 100644
index 00000000..f4e517c8
--- /dev/null
+++ b/tests/plugins/test_ckan.py
@@ -0,0 +1,585 @@
+import datetime
+import json
+import urllib
+from decimal import Decimal
+
+import pytest
+import responses
+from frictionless import Package, Resource, exceptions
+from frictionless.plugins.ckan import CkanStorage
+
+"""
+Tests for CKAN plugin
+
+Note: in this module we decorate all test functions with `@responses.activate` even
+if we don't expect the test to make any HTTP calls. This way, if a test we are not
+expecting to communicate over HTTP tries to make a HTTP request we will throw
+`Connection refused by Responses - the call doesn't match any registered mock`
+instead of silently making the request.
+"""
+
+
+def _mock_json_call(url, json_file, *, method=responses.GET, status=200, **kwargs):
+    body = json.load(open(json_file, encoding="utf-8"))
+    responses.add(method, url, json=body, status=status, **kwargs)
+
+
+class TestRead:
+    def _assert_resource_valid(self, resource):
+
+        # metadata
+        assert resource.schema == {
+            "fields": [
+                {"name": "id", "type": "integer"},
+                {"name": "parent", "type": "integer"},
+                {"name": "name", "type": "string"},
+                {"name": "current", "type": "boolean"},
+                {"name": "rating", "type": "number"},
+                {"name": "created_year", "type": "integer"},
+                {"name": "created_date", "type": "date", "format": "any"},
+                {"name": "created_time", "type": "time", "format": "any"},
+                {"name": "created_datetime", "type": "datetime", "format": "any"},
+                {"name": "stats", "type": "object"},
+                {"name": "persons", "type": "array"},
+                {"name": "location", "type": "object"},
+            ]
+        }
+
+        # data
+        rows = resource.read_rows()
+        assert len(rows) == 3
+        assert rows[0].to_dict() == {
+            "id": 1,
+            "parent": None,
+            "name": "Taxes",
+            "current": True,
+            "rating": 9.5,
+            "created_year": 2015,
+            "created_date": datetime.date(2015, 1, 1),
+            "created_time": datetime.time(3, 0),
+            "created_datetime": datetime.datetime(2015, 1, 1, 3, 0),
+            "stats": {"chars": 560, "height": 54.8},
+            "persons": ["mike", "alice"],
+            "location": {"type": "Point", "coordinates": [50.0, 50.0]},
+        }
+        assert rows[1].to_dict() == {
+            "id": 2,
+            "parent": 1,
+            "name": "中国人",
+            "current": False,
+            "rating": Decimal("7.0"),
+            "created_year": 2015,
+            "created_date": datetime.date(2015, 12, 31),
+            "created_time": datetime.time(15, 45, 33),
+            "created_datetime": datetime.datetime(2015, 12, 31, 15, 45, 33),
+            "stats": {"chars": 970, "height": 40},
+            "persons": ["chen", "alice"],
+            "location": {"type": "Point", "coordinates": [33.33, 33.33]},
+        }
+        assert rows[2].to_dict() == {
+            "id": 3,
+            "parent": 1,
+            "name": None,
+            "current": False,
+            "rating": None,
+            "created_year": None,
+            "created_date": None,
+            "created_time": None,
+            "created_datetime": None,
+            "stats": None,
+            "persons": None,
+            "location": None,
+        }
+
+    @responses.activate
+    def test_read_package_valid(self):
+        resources_ids = [
+            "bd79c992-40f0-454a-a0ff-887f84a792fb",
+            "79843e49-7974-411c-8eb5-fb2d1111d707",
+        ]
+        filter_qs = urllib.parse.urlencode(
+            {"filters": json.dumps({"name": resources_ids})}
+        )
+        dataset_id = "my-dataset-id"
+        base_url = "https://demo.ckan.org"
+
+        _mock_json_call(
+            f"{base_url}/api/3/action/package_show?id={dataset_id}",
+            "data/ckan_mock_responses/package_show.json",
+        )
+
+        # First response gets the _table_metadata results
+        _mock_json_call(
+            f"{base_url}/api/3/action/datastore_search?resource_id=_table_metadata&{filter_qs}",
+            "data/ckan_mock_responses/datastore_search_table_metadata_01.json",
+        )
+        # Second response is empty, indicating the results are exhausted
+        _mock_json_call(
+            f"{base_url}/api/3/action/datastore_search?offset=100&{filter_qs}&resource_id=_table_metadata",
+            "data/ckan_mock_responses/datastore_search_table_metadata_02.json",
+        )
+
+        for resource_id in resources_ids:
+            _mock_json_call(
+                f"{base_url}/api/3/action/datastore_search?limit=0&resource_id={resource_id}",
+                "data/ckan_mock_responses/datastore_search_describe.json",
+            )
+
+            # First response gets the data rows
+            _mock_json_call(
+                f"{base_url}/api/3/action/datastore_search?resource_id={resource_id}",
+                "data/ckan_mock_responses/datastore_search_rows.json",
+            )
+            # Second response is empty, indicating the results are exhausted
+            _mock_json_call(
+                f"{base_url}/api/3/action/datastore_search?resource_id={resource_id}&offset=100",
+                "data/ckan_mock_responses/datastore_search_rows_empty.json",
+            )
+
+        package = Package.from_ckan(
+            base_url=base_url,
+            dataset_id=dataset_id,
+            api_key="env:CKAN_API_KEY",
+        )
+
+        resource = package.get_resource(resources_ids[1])
+        resource.onerror = "raise"  # be strict about errors under test
+
+        self._assert_resource_valid(resource)
+
+    @responses.activate
+    def test_read_resource_valid(self):
+        resource_id = "79843e49-7974-411c-8eb5-fb2d1111d707"
+        base_url = "https://demo.ckan.org"
+
+        _mock_json_call(
+            f"{base_url}/api/3/action/datastore_search?limit=0&resource_id={resource_id}",
+            "data/ckan_mock_responses/datastore_search_describe.json",
+        )
+
+        # First response gets the data rows
+        _mock_json_call(
+            f"{base_url}/api/3/action/datastore_search?resource_id={resource_id}",
+            "data/ckan_mock_responses/datastore_search_rows.json",
+        )
+        # Second response is empty, indicating the results are exhausted
+        _mock_json_call(
+            f"{base_url}/api/3/action/datastore_search?resource_id={resource_id}&offset=100",
+            "data/ckan_mock_responses/datastore_search_rows_empty.json",
+        )
+
+        resource = Resource.from_ckan(
+            base_url=base_url,
+            resource_id=resource_id,
+            api_key="env:CKAN_API_KEY",
+        )
+        resource.onerror = "raise"  # be strict about errors under test
+
+        self._assert_resource_valid(resource)
+
+
+@responses.activate
+def test_read_package_bad_package_id():
+    dataset_id = "bad-dataset-id"
+    base_url = "https://demo.ckan.org"
+
+    _mock_json_call(
+        f"{base_url}/api/3/action/package_show?id={dataset_id}",
+        "data/ckan_mock_responses/ckan_error.json",
+        status=404,
+    )
+
+    with pytest.raises(exceptions.FrictionlessException) as excinfo:
+        Package.from_ckan(
+            base_url=base_url,
+            dataset_id=dataset_id,
+            api_key="env:CKAN_API_KEY",
+        )
+    error = excinfo.value.error
+    assert error.code == "storage-error"
+    assert error.note.count("Not found")
+
+
+@responses.activate
+def test_read_resource_bad_resource_id():
+    resource_id = "bad-resource-id"
+    base_url = "https://demo.ckan.org"
+
+    _mock_json_call(
+        f"{base_url}/api/3/action/datastore_search?limit=0&resource_id={resource_id}",
+        "data/ckan_mock_responses/ckan_error.json",
+        status=404,
+    )
+
+    with pytest.raises(exceptions.FrictionlessException) as excinfo:
+        Resource.from_ckan(
+            base_url=base_url,
+            resource_id=resource_id,
+            api_key="env:CKAN_API_KEY",
+        )
+    error = excinfo.value.error
+    assert error.code == "storage-error"
+    assert error.note.count("Not found")
+
+
+class TestDelete:
+    def setup_method(self):
+        self.resource_id = "79843e49-7974-411c-8eb5-fb2d1111d707"
+        self.base_url = "https://demo.ckan.org"
+        self.storage = CkanStorage(base_url=self.base_url)
+        self.storage._CkanStorage__bucket_cache = [self.resource_id]
+        _mock_json_call(
+            f"{self.base_url}/api/3/action/datastore_delete",
+            "data/ckan_mock_responses/doesnt_matter.json",
+            method=responses.POST,
+        )
+
+    @responses.activate
+    def test_delete_resource_exists(self):
+        self.storage.delete_package([self.resource_id])
+        assert responses.assert_call_count(
+            f"{self.base_url}/api/3/action/datastore_delete", 1
+        )
+
+    @responses.activate
+    def test_delete_package_exists(self):
+        with pytest.raises(exceptions.FrictionlessException) as excinfo:
+            self.storage.delete_package(["bad-id"])
+        error = excinfo.value.error
+        assert error.code == "storage-error"
+        assert error.note.count('Table "bad-id" does not exist')
+
+    @responses.activate
+    def test_delete_resource_does_not_exist(self):
+        self.storage.delete_resource(self.resource_id)
+        assert responses.assert_call_count(
+            f"{self.base_url}/api/3/action/datastore_delete", 1
+        )
+
+    @responses.activate
+    def test_delete_package_does_not_exist(self):
+        with pytest.raises(exceptions.FrictionlessException) as excinfo:
+            self.storage.delete_resource("bad-id")
+        error = excinfo.value.error
+        assert error.code == "storage-error"
+        assert error.note.count('Table "bad-id" does not exist')
+
+
+class TestWriteDoesNotExist:
+    def setup_method(self):
+        self.existing_resources_ids = [
+            "bd79c992-40f0-454a-a0ff-887f84a792fb",
+            "79843e49-7974-411c-8eb5-fb2d1111d707",
+        ]
+        self.target_resource_id = self.existing_resources_ids[1]
+        filter_qs = urllib.parse.urlencode(
+            {"filters": json.dumps({"name": self.existing_resources_ids})}
+        )
+        self.dataset_id = "my-dataset-id"
+        self.base_url = "https://demo.ckan.org"
+
+        _mock_json_call(
+            f"{self.base_url}/api/3/action/package_show?id={self.dataset_id}",
+            "data/ckan_mock_responses/package_show.json",
+        )
+
+        _mock_json_call(
+            f"{self.base_url}/api/3/action/datastore_search?resource_id=_table_metadata&{filter_qs}",
+            "data/ckan_mock_responses/datastore_search_table_metadata_02.json",
+        )
+
+        _mock_json_call(
+            f"{self.base_url}/api/3/action/datastore_create",
+            "data/ckan_mock_responses/doesnt_matter.json",
+            method=responses.POST,
+        )
+
+        _mock_json_call(
+            f"{self.base_url}/api/3/action/datastore_upsert",
+            "data/ckan_mock_responses/doesnt_matter.json",
+            method=responses.POST,
+        )
+
+    @responses.activate
+    def test_write_resource(self):
+        package = Package("data/storage/types.json")
+        package.resources[0].name = self.target_resource_id
+
+        package.resources[0].to_ckan(
+            base_url=self.base_url, dataset_id=self.dataset_id, api_key="env:CKAN_API_KEY"
+        )
+        assert responses.assert_call_count(
+            f"{self.base_url}/api/3/action/datastore_upsert", 1
+        )
+
+    @responses.activate
+    def test_write_package(self):
+        package = Package("data/storage/types.json")
+        package.resources[0].name = self.target_resource_id
+
+        package.resources[0].to_ckan(
+            base_url=self.base_url, dataset_id=self.dataset_id, api_key="env:CKAN_API_KEY"
+        )
+        assert responses.assert_call_count(
+            f"{self.base_url}/api/3/action/datastore_upsert", 1
+        )
+
+    @responses.activate
+    def test_write_types(self):
+        package = Package("data/storage/types.json")
+        package.resources[0].name = self.target_resource_id
+
+        package.resources[0].to_ckan(
+            base_url=self.base_url, dataset_id=self.dataset_id, api_key="env:CKAN_API_KEY"
+        )
+
+        # metdadata
+        assert json.loads(responses.calls[2].request.body) == {
+            "fields": [
+                {"id": "any", "type": "text"},  # type fallback
+                {"id": "array", "type": "text[]"},
+                {"id": "boolean", "type": "bool"},
+                {"id": "date", "type": "text"},  # type downgrade
+                {"id": "date_year", "type": "text"},  # type downgrade
+                {"id": "datetime", "type": "timestamp"},
+                {"id": "duration", "type": "text"},  # type fallback
+                {"id": "geojson", "type": "json"},
+                {"id": "geopoint", "type": "text"},  # type fallback
+                {"id": "integer", "type": "int"},
+                {"id": "number", "type": "float"},
+                {"id": "object", "type": "json"},
+                {"id": "string", "type": "text"},
+                {"id": "time", "type": "time"},
+                {"id": "year", "type": "int"},
+                {"id": "yearmonth", "type": "text"},  # type fallback
+            ],
+            "resource_id": self.target_resource_id,
+            "force": True,
+            "primary_key": [],
+        }
+
+        # data
+        assert json.loads(responses.calls[3].request.body) == {
+            "resource_id": self.target_resource_id,
+            "method": "insert",
+            "force": True,
+            "records": [
+                {
+                    "any": "中国人",
+                    "array": ["Mike", "John"],
+                    "boolean": True,
+                    "date": "2015-01-01",
+                    "date_year": "2015",
+                    "datetime": "2015-01-01T03:00:00",
+                    "duration": "P1Y1M",
+                    "geojson": {"type": "Point", "coordinates": [33, 33.33]},
+                    "geopoint": "30,70",
+                    "integer": 1,
+                    "number": 7.0,
+                    "object": {"chars": 560},
+                    "string": "english",
+                    "time": "03:00:00",
+                    "year": 2015,
+                    "yearmonth": "2015-01",
+                }
+            ],
+        }
+
+    @responses.activate
+    def test_write_integrity(self):
+        package = Package("data/storage/integrity.json")
+        package.resources[0].name = self.existing_resources_ids[0]
+        package.resources[1].name = self.existing_resources_ids[1]
+        package.resources[1].schema["foreignKeys"][0]["reference"][
+            "resource"
+        ] = package.resources[0].name
+
+        package.to_ckan(
+            base_url=self.base_url, dataset_id=self.dataset_id, api_key="env:CKAN_API_KEY"
+        )
+
+        # metadata
+        assert json.loads(responses.calls[2].request.body) == {
+            "fields": [
+                {"id": "id", "type": "int"},  # constraint removal
+                {"id": "parent", "type": "int"},
+                {"id": "description", "type": "text"},
+            ],
+            "resource_id": self.existing_resources_ids[0],
+            "force": True,
+            "primary_key": ["id"],
+            # foreign keys removal
+        }
+        assert json.loads(responses.calls[6].request.body) == {
+            "fields": [
+                {"id": "main_id", "type": "int"},  # constraint removal
+                {"id": "some_id", "type": "int"},  # constraint removal
+                {"id": "description", "type": "text"},  # constraint removal
+            ],
+            "resource_id": self.existing_resources_ids[1],
+            "force": True,
+            "primary_key": ["main_id", "some_id"],
+            # foreign keys removal
+        }
+
+        # data
+        assert json.loads(responses.calls[3].request.body) == {
+            "resource_id": self.existing_resources_ids[0],
+            "method": "insert",
+            "force": True,
+            "records": [
+                {"id": 1, "parent": None, "description": "english"},
+                {"id": 2, "parent": 1, "description": "中国人"},
+            ],
+        }
+        assert json.loads(responses.calls[7].request.body) == {
+            "resource_id": self.existing_resources_ids[1],
+            "method": "insert",
+            "force": True,
+            "records": [
+                {"main_id": 1, "some_id": 1, "description": "note1"},
+                {"main_id": 2, "some_id": 2, "description": "note2"},
+            ],
+        }
+
+    @responses.activate
+    def test_write_constraints(self):
+        package = Package("data/storage/constraints.json")
+        package.resources[0].name = self.target_resource_id
+
+        package.to_ckan(
+            base_url=self.base_url, dataset_id=self.dataset_id, api_key="env:CKAN_API_KEY"
+        )
+
+        # metadata
+        assert json.loads(responses.calls[2].request.body) == {
+            "fields": [
+                {"id": "required", "type": "text"},  # constraint removal
+                {"id": "minLength", "type": "text"},  # constraint removal
+                {"id": "maxLength", "type": "text"},  # constraint removal
+                {"id": "pattern", "type": "text"},  # constraint removal
+                {"id": "enum", "type": "text"},  # constraint removal
+                {"id": "minimum", "type": "int"},  # constraint removal
+                {"id": "maximum", "type": "int"},  # constraint removal
+            ],
+            "resource_id": self.target_resource_id,
+            "force": True,
+            "primary_key": [],
+        }
+
+        # data
+        assert json.loads(responses.calls[3].request.body) == {
+            "resource_id": self.target_resource_id,
+            "method": "insert",
+            "force": True,
+            "records": [
+                {
+                    "required": "passing",
+                    "minLength": "passing",
+                    "maxLength": "passing",
+                    "pattern": "passing",
+                    "enum": "passing",
+                    "minimum": 5,
+                    "maximum": 5,
+                }
+            ],
+        }
+
+
+class TestWriteExists:
+    def setup_method(self):
+        existing_resources_ids = [
+            "bd79c992-40f0-454a-a0ff-887f84a792fb",
+            "79843e49-7974-411c-8eb5-fb2d1111d707",
+        ]
+        self.target_resource_id = existing_resources_ids[1]
+        filter_qs = urllib.parse.urlencode(
+            {"filters": json.dumps({"name": existing_resources_ids})}
+        )
+        self.dataset_id = "my-dataset-id"
+        self.base_url = "https://demo.ckan.org"
+
+        _mock_json_call(
+            f"{self.base_url}/api/3/action/package_show?id={self.dataset_id}",
+            "data/ckan_mock_responses/package_show.json",
+        )
+
+        # First response gets the _table_metadata results
+        _mock_json_call(
+            f"{self.base_url}/api/3/action/datastore_search?resource_id=_table_metadata&{filter_qs}",
+            "data/ckan_mock_responses/datastore_search_table_metadata_01.json",
+        )
+        # Second response is empty, indicating the results are exhausted
+        _mock_json_call(
+            f"{self.base_url}/api/3/action/datastore_search?offset=100&{filter_qs}&resource_id=_table_metadata",
+            "data/ckan_mock_responses/datastore_search_table_metadata_02.json",
+        )
+
+        _mock_json_call(
+            f"{self.base_url}/api/3/action/datastore_delete",
+            "data/ckan_mock_responses/doesnt_matter.json",
+            method=responses.POST,
+        )
+
+        _mock_json_call(
+            f"{self.base_url}/api/3/action/datastore_create",
+            "data/ckan_mock_responses/doesnt_matter.json",
+            method=responses.POST,
+        )
+
+        _mock_json_call(
+            f"{self.base_url}/api/3/action/datastore_upsert",
+            "data/ckan_mock_responses/doesnt_matter.json",
+            method=responses.POST,
+        )
+
+        self.package = Package("data/storage/types.json")
+        self.package.resources[0].name = self.target_resource_id
+
+    @responses.activate
+    def test_write_resource_no_force(self):
+        with pytest.raises(exceptions.FrictionlessException) as excinfo:
+            self.package.to_ckan(
+                base_url=self.base_url,
+                dataset_id=self.dataset_id,
+                api_key="env:CKAN_API_KEY",
+            )
+        error = excinfo.value.error
+        assert error.code == "storage-error"
+        assert error.note.count(f'Table "{self.target_resource_id}" already exists')
+
+    @responses.activate
+    def test_write_resource_with_force(self):
+        self.package.to_ckan(
+            base_url=self.base_url,
+            dataset_id=self.dataset_id,
+            api_key="env:CKAN_API_KEY",
+            force=True,
+        )
+        assert responses.assert_call_count(
+            f"{self.base_url}/api/3/action/datastore_upsert", 1
+        )
+
+    @responses.activate
+    def test_write_package_no_force(self):
+        with pytest.raises(exceptions.FrictionlessException) as excinfo:
+            self.package.to_ckan(
+                base_url=self.base_url,
+                dataset_id=self.dataset_id,
+                api_key="env:CKAN_API_KEY",
+            )
+        error = excinfo.value.error
+        assert error.code == "storage-error"
+        assert error.note.count(f'Table "{self.target_resource_id}" already exists')
+
+    @responses.activate
+    def test_write_package_with_force(self):
+        self.package.to_ckan(
+            base_url=self.base_url,
+            dataset_id=self.dataset_id,
+            api_key="env:CKAN_API_KEY",
+            force=True,
+        )
+        assert responses.assert_call_count(
+            f"{self.base_url}/api/3/action/datastore_upsert", 1
+        )
diff --git a/tests/test_field.py b/tests/test_field.py
index 341b45cd..e8fc669d 100644
--- a/tests/test_field.py
+++ b/tests/test_field.py
@@ -268,7 +268,7 @@ def test_field_read_cell_multiple_constraints():
 
 
 def test_field_to_copy():
-    source = Field(type="integet")
+    source = Field(type="integer")
     target = source.to_copy()
     assert source is not target
     assert source == target
diff --git a/tests/test_row.py b/tests/test_row.py
index 2466bf80..f7b6700e 100644
--- a/tests/test_row.py
+++ b/tests/test_row.py
@@ -1,3 +1,5 @@
+import json
+from decimal import Decimal
 from frictionless import Row, Field, Schema, extract
 
 
@@ -40,6 +42,20 @@ def test_to_list_with_json_null_values_issue_519():
     ]
 
 
+def test_decimal_to_json():
+    row = Row(
+        [Decimal("53.940135311587831")],
+        schema=Schema({"fields": [{"name": "dec1", "type": "number"}]}),
+        field_positions=[1],
+        row_position=1,
+        row_number=1,
+    )
+    # all we really want to 'assert' here is that these methods run without throwing
+    # TypeError: Object of type 'Decimal' is not JSON serializable
+    assert isinstance(json.dumps(row.to_dict(json=True)), str)
+    assert isinstance(json.dumps(row.to_list(json=True)), str)
+
+
 # Helpers