Skip to content

base

BaseQuery

Bases: object

A base class for easy CRUD (create, read, update, delete) access to attribution models.

Source code in ckanext/attribution/model/crud/_base.py 
 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
 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
class BaseQuery(object):
    """
    A base class for easy CRUD (create, read, update, delete) access to attribution
    models.
    """

    __metaclass__ = ABCMeta

    #: :type: The associated database model type.
    m = DomainObject

    #: :sqlalchemy.Table: The associated database table instance
    t = Table()

    @classmethod
    def _columns(cls, **kwargs):
        return {c.name: kwargs.get(c.name) for c in cls.t.c if c.name in kwargs}

    @classmethod
    def validate(cls, data_dict):
        """
        Ensure the data_dict provided contains the correct parameters for creating or
        updating a record, and fix issues where possible by deleting extra fields.

        :param data_dict: a complete dictionary of parameters that will be passed to :func:`create`
                          or :func:`update`
        :type data_dict: dict
        :returns: updated data_dict if valid, raises error if not
        """

        def _empty_string_to_null(item):
            if isinstance(item, list):
                return [_empty_string_to_null(i) for i in item]
            if isinstance(item, dict):
                return {k: _empty_string_to_null(v) for k, v in item.items()}
            if item == '':
                return None
            return item

        data_dict = cls._columns(**_empty_string_to_null(data_dict))

        if 'id' in data_dict:
            existing_record = cls.read(data_dict.get('id'))
            if existing_record is None:
                return data_dict
            for c in cls.t.c:
                if c.name not in data_dict:
                    data_dict[c.name] = getattr(existing_record, c.name)

        return data_dict

    @classmethod
    def create(cls, **kwargs):
        """
        Create a new record of type :class:`~m`.
        """
        item_dict = cls._columns(**kwargs)
        new_item = cls.m(**item_dict)
        Session.add(new_item)
        Session.commit()
        return new_item

    @classmethod
    def read(cls, item_id):
        """
        Retrieve a record of type :class:`~m` by its ID.

        :param item_id: the ID of the record.
        :type item_id: str
        """
        retrieved_item = Session.query(cls.m).get(item_id)
        if retrieved_item is None:
            raise toolkit.ObjectNotFound('{0} was not found.'.format(item_id))
        return retrieved_item

    @classmethod
    def exists(cls, item_id):
        """
        Check if a record with the given ID exists.

        :param item_id: the ID of the potential record
        :returns: bool
        """
        return Session.query(cls.m).get(item_id) is not None

    @classmethod
    def search(cls, query):
        """
        Retrieve all records matching the search criteria.

        :param query: a sqlalchemy filter query
        """
        return Session.query(cls.m).filter(query).all()

    @classmethod
    def all(cls):
        """
        Return all records.
        """
        return Session.query(cls.m).all()

    @classmethod
    def update(cls, item_id, **kwargs):
        try:
            del kwargs['id']
        except KeyError:
            pass
        retrieved_item = Session.query(cls.m).filter(cls.m.id == item_id)
        if retrieved_item.count() < 1:
            raise toolkit.ObjectNotFound('{0} was not found.'.format(item_id))
        retrieved_item.update(cls._columns(**kwargs))
        Session.commit()
        return Session.query(cls.m).get(item_id)

    @classmethod
    def delete(cls, item_id):
        to_delete = Session.query(cls.m).get(item_id)
        if to_delete is not None:
            Session.delete(to_delete)
            Session.commit()

all() classmethod

Return all records.

Source code in ckanext/attribution/model/crud/_base.py 
108
109
110
111
112
113
@classmethod
def all(cls):
    """
    Return all records.
    """
    return Session.query(cls.m).all()

create(**kwargs) classmethod

Create a new record of type :class:~m.

Source code in ckanext/attribution/model/crud/_base.py 
65
66
67
68
69
70
71
72
73
74
@classmethod
def create(cls, **kwargs):
    """
    Create a new record of type :class:`~m`.
    """
    item_dict = cls._columns(**kwargs)
    new_item = cls.m(**item_dict)
    Session.add(new_item)
    Session.commit()
    return new_item

exists(item_id) classmethod

Check if a record with the given ID exists.

Parameters:

Name Type Description Default
item_id

the ID of the potential record

 required

Returns:

Type Description

bool
Source code in ckanext/attribution/model/crud/_base.py 
89
90
91
92
93
94
95
96
97
@classmethod
def exists(cls, item_id):
    """
    Check if a record with the given ID exists.

    :param item_id: the ID of the potential record
    :returns: bool
    """
    return Session.query(cls.m).get(item_id) is not None

read(item_id) classmethod

Retrieve a record of type :class:~m by its ID.

Parameters:

Name Type Description Default
item_id str

the ID of the record.

 required
Source code in ckanext/attribution/model/crud/_base.py 
76
77
78
79
80
81
82
83
84
85
86
87
@classmethod
def read(cls, item_id):
    """
    Retrieve a record of type :class:`~m` by its ID.

    :param item_id: the ID of the record.
    :type item_id: str
    """
    retrieved_item = Session.query(cls.m).get(item_id)
    if retrieved_item is None:
        raise toolkit.ObjectNotFound('{0} was not found.'.format(item_id))
    return retrieved_item

search(query) classmethod

Retrieve all records matching the search criteria.

Parameters:

Name Type Description Default
query

a sqlalchemy filter query

 required
Source code in ckanext/attribution/model/crud/_base.py 
 99
100
101
102
103
104
105
106
@classmethod
def search(cls, query):
    """
    Retrieve all records matching the search criteria.

    :param query: a sqlalchemy filter query
    """
    return Session.query(cls.m).filter(query).all()

validate(data_dict) classmethod

Ensure the data_dict provided contains the correct parameters for creating or updating a record, and fix issues where possible by deleting extra fields.

Parameters:

Name Type Description Default
data_dict dict

a complete dictionary of parameters that will be passed to :func:create or :func:update

 required

Returns:

Type Description

updated data_dict if valid, raises error if not
Source code in ckanext/attribution/model/crud/_base.py 
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
@classmethod
def validate(cls, data_dict):
    """
    Ensure the data_dict provided contains the correct parameters for creating or
    updating a record, and fix issues where possible by deleting extra fields.

    :param data_dict: a complete dictionary of parameters that will be passed to :func:`create`
                      or :func:`update`
    :type data_dict: dict
    :returns: updated data_dict if valid, raises error if not
    """

    def _empty_string_to_null(item):
        if isinstance(item, list):
            return [_empty_string_to_null(i) for i in item]
        if isinstance(item, dict):
            return {k: _empty_string_to_null(v) for k, v in item.items()}
        if item == '':
            return None
        return item

    data_dict = cls._columns(**_empty_string_to_null(data_dict))

    if 'id' in data_dict:
        existing_record = cls.read(data_dict.get('id'))
        if existing_record is None:
            return data_dict
        for c in cls.t.c:
            if c.name not in data_dict:
                data_dict[c.name] = getattr(existing_record, c.name)

    return data_dict