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
|
# -*- 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/>.
'''
couchdb.py
==========
Classes for working with CouchDB or BigCouch instances which store email alias
maps, user UUIDs, and GPG keyIDs.
'''
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."
from functools import partial
from leap.mx.util import log
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)
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
"""
log.msg("Available databases:")
for database in data:
log.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"
d = self.openView(docId="User",
viewId="by_login_or_alias/",
key=alias,
reduce=False)
d.addCallbacks(partial(self._get_uuid, alias), log.err)
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:
return row["id"]
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
d2 = cdb.queryByLoginOrAlias("asdjaoisdjoiqwjeoi")
@d2.addCallback
def wrong(result):
print "Should be None:", result
reactor.callLater(5, reactor.stop)
reactor.run()
|
