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
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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
|
# -*- encoding: utf-8 -*-
# couchdb.py
# Copyright (C) 2013 LEAP
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
Classes for working with CouchDB or BigCouch instances which store email alias
maps, user UUIDs, and GPG keyIDs.
"""
import logging
from functools import partial
try:
from paisley import client
except ImportError:
print "This software requires paisley. Please see the README file"
print "for instructions on getting required dependencies."
try:
from twisted.internet import defer
except ImportError:
print "This software requires Twisted. Please see the README file"
print "for instructions on getting required dependencies."
logger = logging.getLogger(__name__)
class ConnectedCouchDB(client.CouchDB):
"""
Connect to a CouchDB instance.
CouchDB document for testing is '_design', and the view is simply
a preconfigured set of mapped responses.
"""
def __init__(self, host, port=5984, dbName=None, username=None,
password=None, *args, **kwargs):
"""
Connect to a CouchDB instance.
@param host: A hostname string for the CouchDB server.
@type host: str
@param port: The port of the CouchDB server.
@type port: int
@param dbName: (optional) The default database to bind queries to.
@type dbName: str
@param username: (optional) The username for authorization.
@type username: str
@param str password: (optional) The password for authorization.
@type password: str
"""
client.CouchDB.__init__(self,
host,
port=port,
dbName=dbName,
username=username,
password=password,
*args, **kwargs)
self._cache = {}
if dbName is None:
databases = self.listDB()
databases.addCallback(self._print_databases)
def _print_databases(self, data):
"""
Callback for listDB that prints the available databases
@param data: response from the listDB command
@type data: array
"""
logger.msg("Available databases:")
for database in data:
logger.msg(" * %s" % (database,))
def createDB(self, dbName):
"""
Overrides ``paisley.client.CouchDB.createDB``.
"""
pass
def deleteDB(self, dbName):
"""
Overrides ``paisley.client.CouchDB.deleteDB``.
"""
pass
def queryByLoginOrAlias(self, alias):
"""
Check to see if a particular email or alias exists.
@param alias: A string representing the email or alias to check.
@type alias: str
@return: a deferred for this query
@rtype twisted.defer.Deferred
"""
assert isinstance(alias, str), "Email or alias queries must be string"
# TODO: Cache results
d = self.openView(docId="User",
viewId="by_login_or_alias/",
key=alias,
reduce=False,
include_docs=True)
d.addCallbacks(partial(self._get_uuid, alias), logger.error)
return d
def _get_uuid(self, alias, result):
"""
Parses the result of the by_login_or_alias query and gets the
uuid
@param alias: alias looked up
@type alias: string
@param result: result dictionary
@type result: dict
@return: The uuid for alias if available
@rtype: str
"""
for row in result["rows"]:
if row["key"] == alias:
uuid = row["id"]
try:
self._cache[uuid] = row["doc"]["public_key"]
except:
pass # no public key for this user
return uuid
return None
def getPubKey(self, uuid):
pubkey = None
try:
pubkey = self._cache[uuid]
except:
pass
return pubkey
if __name__ == "__main__":
from twisted.internet import reactor
cdb = ConnectedCouchDB("localhost",
port=6666,
dbName="users",
username="",
password="")
d = cdb.queryByLoginOrAlias("test1")
@d.addCallback
def right(result):
print "Should be an actual uuid:", result
print "Public Key:"
print cdb.getPubKey(result)
d2 = cdb.queryByLoginOrAlias("asdjaoisdjoiqwjeoi")
@d2.addCallback
def wrong(result):
print "Should be None:", result
reactor.callLater(5, reactor.stop)
reactor.run()
|
