-
Notifications
You must be signed in to change notification settings - Fork 2k
/
domain_object.py
117 lines (89 loc) · 2.88 KB
/
domain_object.py
1
2
3
4
5
6
7
8
9
10
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
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
# encoding: utf-8
import datetime
from collections import OrderedDict
import sqlalchemy as sa
from sqlalchemy import orm
from ckan.model import meta, core
__all__ = ['DomainObject', 'DomainObjectOperation']
class Enum(set):
'''Simple enumeration
e.g. Animal = Enum("dog", "cat", "horse")
joey = Animal.DOG
'''
def __init__(self, *names):
super(Enum, self).__init__(names)
def __getattr__(self, name):
if name in self:
return name
raise AttributeError
DomainObjectOperation = Enum('new', 'changed', 'deleted')
class DomainObject(object):
text_search_fields = []
Session = meta.Session
def __init__(self, **kwargs):
for k,v in kwargs.items():
setattr(self, k, v)
@classmethod
def count(cls):
return cls.Session.query(cls).count()
@classmethod
def by_name(cls, name, autoflush=True):
obj = meta.Session.query(cls).autoflush(autoflush)\
.filter_by(name=name).first()
return obj
@classmethod
def text_search(cls, query, term):
register = cls
make_like = lambda x,y: x.ilike('%' + y + '%')
q = None
for field in cls.text_search_fields:
attr = getattr(register, field)
q = sa.or_(q, make_like(attr, term))
return query.filter(q)
@classmethod
def active(cls):
return meta.Session.query(cls).filter_by(state=core.State.ACTIVE)
def save(self):
self.add()
self.commit()
def add(self):
self.Session.add(self)
def commit_remove(self):
self.commit()
self.remove()
def commit(self):
self.Session.commit()
def remove(self):
self.Session.remove()
def delete(self):
# stateful objects have this method overridden - see
# core.StatefulObjectMixin
self.Session.delete(self)
def purge(self):
self.Session().autoflush = False
self.Session.delete(self)
def as_dict(self):
_dict = OrderedDict()
table = orm.class_mapper(self.__class__).mapped_table
for col in table.c:
val = getattr(self, col.name)
if isinstance(val, datetime.date):
val = str(val)
if isinstance(val, datetime.datetime):
val = val.isoformat()
_dict[col.name] = val
return _dict
def __str__(self):
return self.__unicode__().encode('utf8')
def __unicode__(self):
repr = u'<%s' % self.__class__.__name__
table = orm.class_mapper(self.__class__).mapped_table
for col in table.c:
try:
repr += u' %s=%s' % (col.name, getattr(self, col.name))
except Exception as inst:
repr += u' %s=%s' % (col.name, inst)
repr += '>'
return repr
def __repr__(self):
return self.__unicode__().encode('utf-8')