Skip to content

CKAN

CKAN plugin provides Package and Resource models and converters between CKAN and Data Package notations

Installation

Not extra dependencies are required

Usage

Converting a CKAN descriptor to the Data Package notation:

from dplib.plugins.ckan.models import CkanPackage

package = CkanPackage.from_path("data/plugins/ckan/package.json").to_dp()
print(package.to_text(format='json'))

{
  "resources": [
    {
      "name": "sample_linked",
      "path": "sample-linked.csv",
      "format": "csv",
      "mediatype": "text/csv",
      "ckan:id": "e687245d-7835-44b0-8ed3-0827de123895"
    },
    {
      "name": "sample",
      "path": "sample.csv",
      "format": "csv",
      "mediatype": "application/csv",
      "bytes": 6731,
      "ckan:id": "b53c9e72-6b59-4cda-8c0c-7d6a51dad12a"
    },
    {
      "name": "views",
      "path": "views.csv",
      "format": "csv",
      "bytes": 32773,
      "ckan:id": "9ce6650b-6ff0-4a52-9b10-09cfc29bbd7e"
    },
    {
      "name": "sample",
      "path": "sample.pdf",
      "format": "pdf",
      "bytes": 712352,
      "ckan:id": "8aa53505-3b7f-4b9c-9b54-cf674eadc3f1"
    },
    {
      "name": "sample",
      "path": "sample.txt",
      "format": "txt",
      "bytes": 85,
      "ckan:id": "0185907b-2812-437f-9c64-eae24771ef5f"
    },
    {
      "name": "sample",
      "path": "sample.geojson",
      "format": "geojson",
      "bytes": 255943,
      "ckan:id": "ecd4a62d-998b-46e4-8a64-cadac2125c64"
    },
    {
      "name": "sample",
      "path": "sample.kml",
      "format": "kml",
      "bytes": 474000,
      "ckan:id": "048333ab-9608-42dc-901b-a7dd9fca3dda"
    },
    {
      "name": "avoid_crowds_when_buying_materials_social_media_post",
      "path": "avoid-crowds-when-buying-materials-social-media-post.jpeg",
      "format": "jpeg",
      "mediatype": "image/png",
      "bytes": 444695,
      "ckan:id": "b6c22c1d-e789-490d-b935-989093bbb173"
    },
    {
      "name": "sample_wms",
      "path": "Sample WMS",
      "format": "wms",
      "ckan:id": "664e5e2c-bd7d-4972-a245-a747f7d61cc9"
    }
  ],
  "name": "sample-dataset-1",
  "title": "Sample Dataset",
  "description": "A CKAN Dataset is a collection of data resources (such as files), together with a description and other information (what is known as metadata), at a fixed URL. \r\n\r\n",
  "version": "1.0",
  "licenses": [
    {
      "name": "cc-by",
      "title": "Creative Commons Attribution",
      "url": "http://www.opendefinition.org/licenses/cc-by"
    }
  ],
  "contributors": [
    {
      "title": "Test Author",
      "email": "test@email.com",
      "role": "author"
    },
    {
      "title": "Test Maintainer",
      "email": "test@email.com",
      "role": "maintainer"
    }
  ],
  "keywords": [
    "csv",
    "economy",
    "geojson",
    "kml",
    "pdf",
    "sample",
    "txt",
    "wms"
  ],
  "created": "2021-04-09T11:39:37.657233",
  "ckan:id": "c322307a-b871-44fe-a602-32ee8437ff04"
}

Converting a Data Package to CKAN notation:

from dplib.models import Package
from dplib.plugins.ckan.models import CkanPackage

package = CkanPackage.from_dp(Package.from_path("data/package.json"))
print(package.to_text(format="json"))

Reference

dplib.plugins.ckan.models.CkanPackage

Bases: Model

CKAN Package model

Source code in dplib/plugins/ckan/models/package.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class CkanPackage(Model):
    """CKAN Package model"""

    resources: List[CkanResource] = []

    organization: Optional[CkanOrganization] = None
    tags: List[CkanTag] = []

    id: Optional[str] = None
    name: Optional[str] = None
    title: Optional[str] = None
    notes: Optional[str] = None
    version: Optional[str] = None
    notes: Optional[str] = None
    license_id: Optional[str] = None
    license_title: Optional[str] = None
    license_url: Optional[str] = None
    author: Optional[str] = None
    author_email: Optional[str] = None
    maintainer: Optional[str] = None
    maintainer_email: Optional[str] = None
    metadata_created: Optional[str] = None
    metadata_modified: Optional[str] = None

    # Converters

    def to_dp(self) -> Package:
        """Convert to Data Package

        Returns:
           Data Package
        """
        package = Package()

        # Name
        if self.name:
            package.name = self.name

        # Title
        if self.title:
            package.title = self.title

        # Description
        if self.notes:
            package.description = self.notes

        # Version
        if self.version:
            package.version = self.version

        # Created
        if self.metadata_created:
            package.created = add_timezone_infromation(self.metadata_created)

        # License
        if self.license_id:
            license = License(name=self.license_id)
            if self.license_title:
                license.title = self.license_title
            if self.license_url:
                license.path = self.license_url
            package.licenses.append(license)

        # Contributors
        if self.author:
            contributor = Contributor(title=self.author, roles=["author"])
            if self.author_email:
                contributor.email = self.author_email
            package.contributors.append(contributor)
        if self.maintainer:
            contributor = Contributor(title=self.maintainer, roles=["maintainer"])
            if self.maintainer_email:
                contributor.email = self.maintainer_email
            package.contributors.append(contributor)

        # Resources
        for item in self.resources:
            resource = item.to_dp()
            package.resources.append(resource)

        # Keywords
        for tag in self.tags:
            package.keywords.append(tag.name)

        # Custom
        if self.id:
            package.custom["ckan:id"] = self.id

        return package

    @classmethod
    def from_dp(cls, package: Package) -> CkanPackage:
        """Create a CKAN Package from Data Package

        Parameters:
            package: Data Package

        Returns:
            CKAN Package
        """
        ckan = CkanPackage()

        # Name
        if package.name:
            ckan.name = package.name

        # Title
        if package.title:
            ckan.title = package.title

        # Description
        if package.description:
            ckan.notes = package.description

        # Version
        if package.version:
            ckan.version = package.version

        # License
        if package.licenses:
            license = package.licenses[0]
            ckan.license_id = license.name
            if license.title:
                ckan.license_title = license.title
            if license.path:
                ckan.license_url = license.path

        # Contributors
        for contributor in package.contributors:
            if contributor.roles:
                if "author" in contributor.roles:
                    ckan.author = contributor.title
                    ckan.author_email = contributor.email
                elif "maintainer" in contributor.roles:
                    ckan.maintainer = contributor.title
                    ckan.maintainer_email = contributor.email

        # Resources
        for resource in package.resources:
            item = CkanResource.from_dp(resource)
            if item:
                ckan.resources.append(item)

        # Keywords
        for keyword in package.keywords:
            tag = CkanTag(name=keyword)
            ckan.tags.append(tag)

        return ckan

author: Optional[str] = None class-attribute instance-attribute

author_email: Optional[str] = None class-attribute instance-attribute

id: Optional[str] = None class-attribute instance-attribute

license_id: Optional[str] = None class-attribute instance-attribute

license_title: Optional[str] = None class-attribute instance-attribute

license_url: Optional[str] = None class-attribute instance-attribute

maintainer: Optional[str] = None class-attribute instance-attribute

maintainer_email: Optional[str] = None class-attribute instance-attribute

metadata_created: Optional[str] = None class-attribute instance-attribute

metadata_modified: Optional[str] = None class-attribute instance-attribute

name: Optional[str] = None class-attribute instance-attribute

notes: Optional[str] = None class-attribute instance-attribute

organization: Optional[CkanOrganization] = None class-attribute instance-attribute

resources: List[CkanResource] = [] class-attribute instance-attribute

tags: List[CkanTag] = [] class-attribute instance-attribute

title: Optional[str] = None class-attribute instance-attribute

version: Optional[str] = None class-attribute instance-attribute

from_dp(package) classmethod

Create a CKAN Package from Data Package

Parameters:

Name Type Description Default
package Package

Data Package

 required

Returns:

Type Description
CkanPackage

CKAN Package
Source code in dplib/plugins/ckan/models/package.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
@classmethod
def from_dp(cls, package: Package) -> CkanPackage:
    """Create a CKAN Package from Data Package

    Parameters:
        package: Data Package

    Returns:
        CKAN Package
    """
    ckan = CkanPackage()

    # Name
    if package.name:
        ckan.name = package.name

    # Title
    if package.title:
        ckan.title = package.title

    # Description
    if package.description:
        ckan.notes = package.description

    # Version
    if package.version:
        ckan.version = package.version

    # License
    if package.licenses:
        license = package.licenses[0]
        ckan.license_id = license.name
        if license.title:
            ckan.license_title = license.title
        if license.path:
            ckan.license_url = license.path

    # Contributors
    for contributor in package.contributors:
        if contributor.roles:
            if "author" in contributor.roles:
                ckan.author = contributor.title
                ckan.author_email = contributor.email
            elif "maintainer" in contributor.roles:
                ckan.maintainer = contributor.title
                ckan.maintainer_email = contributor.email

    # Resources
    for resource in package.resources:
        item = CkanResource.from_dp(resource)
        if item:
            ckan.resources.append(item)

    # Keywords
    for keyword in package.keywords:
        tag = CkanTag(name=keyword)
        ckan.tags.append(tag)

    return ckan

to_dp()

Convert to Data Package

Returns:

Type Description
Package

Data Package
Source code in dplib/plugins/ckan/models/package.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def to_dp(self) -> Package:
    """Convert to Data Package

    Returns:
       Data Package
    """
    package = Package()

    # Name
    if self.name:
        package.name = self.name

    # Title
    if self.title:
        package.title = self.title

    # Description
    if self.notes:
        package.description = self.notes

    # Version
    if self.version:
        package.version = self.version

    # Created
    if self.metadata_created:
        package.created = add_timezone_infromation(self.metadata_created)

    # License
    if self.license_id:
        license = License(name=self.license_id)
        if self.license_title:
            license.title = self.license_title
        if self.license_url:
            license.path = self.license_url
        package.licenses.append(license)

    # Contributors
    if self.author:
        contributor = Contributor(title=self.author, roles=["author"])
        if self.author_email:
            contributor.email = self.author_email
        package.contributors.append(contributor)
    if self.maintainer:
        contributor = Contributor(title=self.maintainer, roles=["maintainer"])
        if self.maintainer_email:
            contributor.email = self.maintainer_email
        package.contributors.append(contributor)

    # Resources
    for item in self.resources:
        resource = item.to_dp()
        package.resources.append(resource)

    # Keywords
    for tag in self.tags:
        package.keywords.append(tag.name)

    # Custom
    if self.id:
        package.custom["ckan:id"] = self.id

    return package

dplib.plugins.ckan.models.CkanResource

Bases: Model

CKAN Resource model

Source code in dplib/plugins/ckan/models/resource.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class CkanResource(Model):
    """CKAN Resource model"""

    url: str
    name: str
    created: Optional[str] = None
    description: Optional[str] = None
    format: Optional[str] = None  # NOTE: uppercased
    hash: Optional[str] = None
    id: Optional[str] = None
    last_modified: Optional[str] = None
    metadata_modified: Optional[str] = None
    mimetype: Optional[str] = None
    size: Optional[int] = None

    # Converters

    def to_dp(self) -> Resource:
        """Convert to Data Package resource

        Returns:
           Data Resource
        """
        # Path/Name
        resource = Resource(path=self.url, name=slugify_name(self.name))

        # Description
        if self.description:
            resource.description = self.description

        # Format
        if self.format:
            resource.format = self.format.lower()

        # Mediatype
        if self.mimetype:
            resource.mediatype = self.mimetype

        # Size
        if self.size:
            resource.bytes = self.size

        # Custom
        if self.id:
            resource.custom["ckan:id"] = self.id

        return resource

    @classmethod
    def from_dp(cls, resource: Resource) -> Optional[CkanResource]:
        """Create CKAN Resource from Data Resource

        Parameters:
            resource: Data Resource

        Returns:
            CKAN Resource
        """
        if not resource.path or not isinstance(resource.path, str):
            return

        # Path/Name
        ckan = CkanResource(url=resource.path, name=os.path.basename(resource.path))

        # Description
        if resource.description:
            ckan.description = resource.description

        # Format
        if resource.format:
            ckan.format = resource.format.upper()

        # Mediatype
        if resource.mediatype:
            ckan.mimetype = resource.mediatype

        # Bytes
        if resource.bytes:
            ckan.size = resource.bytes

        return ckan

created: Optional[str] = None class-attribute instance-attribute

description: Optional[str] = None class-attribute instance-attribute

format: Optional[str] = None class-attribute instance-attribute

hash: Optional[str] = None class-attribute instance-attribute

id: Optional[str] = None class-attribute instance-attribute

last_modified: Optional[str] = None class-attribute instance-attribute

metadata_modified: Optional[str] = None class-attribute instance-attribute

mimetype: Optional[str] = None class-attribute instance-attribute

name: str instance-attribute

size: Optional[int] = None class-attribute instance-attribute

url: str instance-attribute

from_dp(resource) classmethod

Create CKAN Resource from Data Resource

Parameters:

Name Type Description Default
resource Resource

Data Resource

 required

Returns:

Type Description
Optional[CkanResource]

CKAN Resource
Source code in dplib/plugins/ckan/models/resource.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@classmethod
def from_dp(cls, resource: Resource) -> Optional[CkanResource]:
    """Create CKAN Resource from Data Resource

    Parameters:
        resource: Data Resource

    Returns:
        CKAN Resource
    """
    if not resource.path or not isinstance(resource.path, str):
        return

    # Path/Name
    ckan = CkanResource(url=resource.path, name=os.path.basename(resource.path))

    # Description
    if resource.description:
        ckan.description = resource.description

    # Format
    if resource.format:
        ckan.format = resource.format.upper()

    # Mediatype
    if resource.mediatype:
        ckan.mimetype = resource.mediatype

    # Bytes
    if resource.bytes:
        ckan.size = resource.bytes

    return ckan

to_dp()

Convert to Data Package resource

Returns:

Type Description
Resource

Data Resource
Source code in dplib/plugins/ckan/models/resource.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def to_dp(self) -> Resource:
    """Convert to Data Package resource

    Returns:
       Data Resource
    """
    # Path/Name
    resource = Resource(path=self.url, name=slugify_name(self.name))

    # Description
    if self.description:
        resource.description = self.description

    # Format
    if self.format:
        resource.format = self.format.lower()

    # Mediatype
    if self.mimetype:
        resource.mediatype = self.mimetype

    # Size
    if self.size:
        resource.bytes = self.size

    # Custom
    if self.id:
        resource.custom["ckan:id"] = self.id

    return resource