8
by William Grant
A little bit of server. |
1 |
# Copyright (c) 2012 Canonical Ltd
|
2 |
#
|
|
3 |
# This program is free software: you can redistribute it and/or modify
|
|
4 |
# it under the terms of the GNU Affero General Public License as published by
|
|
5 |
# the Free Software Foundation, either version 3 of the License, or
|
|
6 |
# (at your option) any later version.
|
|
7 |
#
|
|
8 |
# This program is distributed in the hope that it will be useful,
|
|
9 |
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
10 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
11 |
# GNU Affero General Public License for more details.
|
|
12 |
#
|
|
13 |
# You should have received a copy of the GNU Affero General Public
|
|
14 |
# License along with this program. If not, see
|
|
15 |
# <http://www.gnu.org/licenses/>.
|
|
16 |
||
17 |
import datetime |
|
9
by William Grant
Extract some message metadata. Log betterer. |
18 |
import dateutil.tz |
19 |
import email.parser |
|
20 |
from email.utils import parsedate_tz |
|
21 |
import logging |
|
31
by William Grant
Use real UTC timestamps in the archive_message TimeUUIDs. pycassa's end up including the local timezone offset :/ |
22 |
import time |
8
by William Grant
A little bit of server. |
23 |
import uuid |
24 |
||
25 |
import pycassa |
|
26 |
from pycassa.system_manager import ( |
|
27 |
LEXICAL_UUID_TYPE, |
|
28 |
SystemManager, |
|
29 |
TIME_UUID_TYPE, |
|
30 |
)
|
|
32
by William Grant
Date filtering. |
31 |
from pycassa.util import convert_time_to_uuid |
8
by William Grant
A little bit of server. |
32 |
|
21
by William Grant
Merge grackle.server into grackle. Alter Makefile to run all the tests. |
33 |
from grackle.cassandra import workaround_1779 |
8
by William Grant
A little bit of server. |
34 |
|
35 |
||
19
by William Grant
grackle-create-instance can now create the keyspace too. |
36 |
def create_schema(host, keyspace, clobber=False, create_keyspace=False): |
8
by William Grant
A little bit of server. |
37 |
mgr = SystemManager(host) |
38 |
||
19
by William Grant
grackle-create-instance can now create the keyspace too. |
39 |
if create_keyspace: |
40 |
mgr.create_keyspace(keyspace, replication_factor=1) |
|
41 |
||
8
by William Grant
A little bit of server. |
42 |
if clobber: |
43 |
for cf in mgr.get_keyspace_column_families(keyspace): |
|
44 |
mgr.drop_column_family(keyspace, cf) |
|
45 |
||
46 |
try: |
|
47 |
workaround_1779( |
|
48 |
mgr.create_column_family, keyspace, 'message', |
|
49 |
key_validation_class=LEXICAL_UUID_TYPE) |
|
50 |
workaround_1779( |
|
51 |
mgr.create_column_family, keyspace, 'archive_message', |
|
52 |
comparator_type=TIME_UUID_TYPE, |
|
53 |
default_validation_class=LEXICAL_UUID_TYPE) |
|
54 |
pass
|
|
55 |
finally: |
|
56 |
mgr.close() |
|
57 |
||
58 |
||
17
by William Grant
Turn _parse_message into a non-member function. |
59 |
def _parse_message(message): |
60 |
"""Get a date and dict of an RFC822 message."""
|
|
61 |
parsed = email.parser.Parser().parsestr(message) |
|
62 |
message_dict = {} |
|
63 |
||
64 |
for key in ('from', 'to', 'subject', 'message-id'): |
|
65 |
value = parsed.get(key, None) |
|
66 |
if value is not None: |
|
67 |
message_dict[key] = value |
|
68 |
||
69 |
date = parsed.get('date') |
|
70 |
if date is not None: |
|
71 |
try: |
|
72 |
pdate = parsedate_tz(date) |
|
73 |
date = datetime.datetime( |
|
74 |
*pdate[:6], |
|
75 |
tzinfo=dateutil.tz.tzoffset('', pdate[9])) |
|
76 |
except ValueError: |
|
77 |
pass
|
|
78 |
message_dict['date'] = date.isoformat() if date is not None else None |
|
79 |
||
80 |
return date, message_dict |
|
81 |
||
82 |
||
31
by William Grant
Use real UTC timestamps in the archive_message TimeUUIDs. pycassa's end up including the local timezone offset :/ |
83 |
def _utc_datetime(dt): |
84 |
return dt.astimezone(dateutil.tz.tzutc()) |
|
85 |
||
86 |
||
87 |
def _utc_timestamp(dt): |
|
88 |
return time.mktime(_utc_datetime(dt).timetuple()) - time.timezone |
|
89 |
||
90 |
||
32
by William Grant
Date filtering. |
91 |
def _utc_timeuuid(dt, lowest_val=True): |
92 |
return convert_time_to_uuid(_utc_timestamp(dt), lowest_val) |
|
93 |
||
94 |
||
95 |
def _cmp_timeuuid(a, b): |
|
96 |
if a.time != b.time: |
|
97 |
return cmp(a.time, b.time) |
|
98 |
return cmp(a, b) |
|
99 |
||
100 |
||
101 |
def _bound_timeuuid(a, b, max=False): |
|
102 |
if a == '' or _cmp_timeuuid(b, a) == (1 if max else -1): |
|
103 |
return b |
|
104 |
return a |
|
105 |
||
106 |
||
34
by William Grant
Extract _make_bounds. |
107 |
def _make_bounds(memo, range_start, range_finish, backward): |
108 |
start = finish = '' |
|
109 |
if memo != '': |
|
110 |
memo = uuid.UUID(memo) |
|
111 |
if backward: |
|
112 |
finish = memo |
|
113 |
else: |
|
114 |
start = memo |
|
115 |
if range_start is not None: |
|
116 |
start = _bound_timeuuid( |
|
117 |
start, _utc_timeuuid(range_start), max=True) |
|
118 |
if range_finish is not None: |
|
119 |
finish = _bound_timeuuid( |
|
120 |
finish, _utc_timeuuid(range_finish, lowest_val=False)) |
|
121 |
return memo, start, finish |
|
122 |
||
123 |
||
8
by William Grant
A little bit of server. |
124 |
class CassandraConnection(object): |
125 |
||
126 |
def __init__(self, keyspace, host): |
|
127 |
self._keyspace = keyspace |
|
128 |
self._host = host |
|
129 |
self._connection = self._connect() |
|
130 |
self.messages = self._column_family('message') |
|
131 |
self.archive_messages = self._column_family('archive_message') |
|
132 |
||
133 |
def _connect(self): |
|
134 |
return pycassa.connect(self._keyspace, self._host) |
|
135 |
||
136 |
def _column_family(self, name): |
|
137 |
return pycassa.ColumnFamily(self._connection, name) |
|
138 |
||
14
by William Grant
Refactor. |
139 |
def add_message(self, archive_uuid, message): |
140 |
message_uuid = uuid.uuid4() |
|
17
by William Grant
Turn _parse_message into a non-member function. |
141 |
message_date, message_dict = _parse_message(message) |
16
by William Grant
message content isn't always included. |
142 |
message_dict['content'] = message |
15
by William Grant
date_created is no longer part of the original message dict. |
143 |
message_dict['date_created'] = ( |
144 |
datetime.datetime.utcnow().isoformat() + 'Z') |
|
14
by William Grant
Refactor. |
145 |
self.messages.insert(message_uuid, message_dict) |
8
by William Grant
A little bit of server. |
146 |
self.archive_messages.insert( |
9
by William Grant
Extract some message metadata. Log betterer. |
147 |
archive_uuid, |
31
by William Grant
Use real UTC timestamps in the archive_message TimeUUIDs. pycassa's end up including the local timezone offset :/ |
148 |
{_utc_timestamp(message_date): message_uuid}) |
9
by William Grant
Extract some message metadata. Log betterer. |
149 |
logging.debug( |
14
by William Grant
Refactor. |
150 |
'Imported %s into %s' |
151 |
% (message_dict.get('message-id', None), archive_uuid)) |
|
8
by William Grant
A little bit of server. |
152 |
return message_uuid |
153 |
||
9
by William Grant
Extract some message metadata. Log betterer. |
154 |
def _format_message(self, message): |
155 |
return { |
|
14
by William Grant
Refactor. |
156 |
'date': message.get('date'), |
157 |
'from': message.get('from'), |
|
158 |
'subject': message.get('subject'), |
|
23
by William Grant
Include message-id in the result. |
159 |
'message-id': message.get('message-id'), |
9
by William Grant
Extract some message metadata. Log betterer. |
160 |
}
|
161 |
||
29
by William Grant
Do backward correctly. |
162 |
def _trim(self, sequence, end): |
30
by William Grant
Comment and document. |
163 |
"""Return the sequence with one of the ends trimmed.
|
164 |
||
165 |
:param end: if true, remove the last element. otherwise remove
|
|
166 |
the first.
|
|
167 |
"""
|
|
29
by William Grant
Do backward correctly. |
168 |
if end: |
169 |
return sequence[:-1] |
|
170 |
else: |
|
171 |
return sequence[1:] |
|
172 |
||
32
by William Grant
Date filtering. |
173 |
def get_messages(self, archive_uuid, order, count, memo, backward=False, |
174 |
start_date=None, finish_date=None): |
|
10
by William Grant
Let the HTTP client decide count and order (to an extent). |
175 |
if order in ("date", "-date"): |
176 |
reversed = order[0] == '-' |
|
177 |
else: |
|
178 |
raise AssertionError("Unsupported order.") |
|
34
by William Grant
Extract _make_bounds. |
179 |
|
180 |
memo, start, finish = _make_bounds( |
|
181 |
memo, start_date, finish_date, backward) |
|
30
by William Grant
Comment and document. |
182 |
|
183 |
# Get up to n+1 messages from the memo: the last item of the
|
|
184 |
# previous batch (because that's where the memo starts) + this
|
|
185 |
# batch.
|
|
11
by William Grant
Support batching. |
186 |
pairs = self.archive_messages.get( |
29
by William Grant
Do backward correctly. |
187 |
archive_uuid, column_count=count + 1, column_start=start, |
188 |
column_finish=finish, column_reversed=reversed).items() |
|
30
by William Grant
Comment and document. |
189 |
|
190 |
if len(pairs) and memo and pairs[0][0] <= memo: |
|
191 |
# The memo (from the previous batch) was included in the result.
|
|
192 |
# Trim it.
|
|
29
by William Grant
Do backward correctly. |
193 |
pairs = self._trim(pairs, False ^ backward) |
27
by William Grant
Fix batching to almost work backwards too. |
194 |
elif len(pairs) > count: |
30
by William Grant
Comment and document. |
195 |
# There was no memo in the result, so the n+1th element is
|
196 |
# unnecessary. Kill it.
|
|
29
by William Grant
Do backward correctly. |
197 |
pairs = self._trim(pairs, True ^ backward) |
27
by William Grant
Fix batching to almost work backwards too. |
198 |
|
199 |
if len(pairs) == 0: |
|
200 |
return (None, [], None) |
|
201 |
||
202 |
assert 0 < len(pairs) <= count |
|
203 |
||
30
by William Grant
Comment and document. |
204 |
# We've narrowed down the message references. Fetch the messages.
|
11
by William Grant
Support batching. |
205 |
ids = [v for k, v in pairs] |
206 |
messages = self.messages.multiget( |
|
23
by William Grant
Include message-id in the result. |
207 |
ids, columns=['date', 'from', 'subject', 'message-id']) |
27
by William Grant
Fix batching to almost work backwards too. |
208 |
|
11
by William Grant
Support batching. |
209 |
return ( |
27
by William Grant
Fix batching to almost work backwards too. |
210 |
str(pairs[0][0]), |
211 |
[self._format_message(messages[id]) for id in ids], |
|
212 |
str(pairs[-1][0]), |
|
11
by William Grant
Support batching. |
213 |
)
|