Repository navigation
Expand file tree
/
Copy pathreplicator.py
More file actions
173 lines (136 loc) · 5.39 KB
/
Copy pathreplicator.py
File metadata and controls
173 lines (136 loc) · 5.39 KB
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
"""
_replicator_
Implement a friendly wrapper around a Cloudant _replicator database.
"""
import uuid
from .database import CloudantDatabase
from .errors import CloudantException
class ReplicatorDatabase(CloudantDatabase):
"""
CloudantDatabase object with a fixed name and some additional
methods specific to a _replicator database.
"""
def __init__(self, cloudant, fetch_limit=100):
super(ReplicatorDatabase, self).__init__(
cloudant=cloudant,
database_name="_replicator",
fetch_limit=fetch_limit
)
def create_replication(self, source_db=None, target_db=None,
repl_id=None, **kwargs):
"""
_create_replication_
Create a new replication.
@param Database source_db: Database object to replicate from
@param Database target_db: Database object to replicate to
@param str repl_id: replication_id (I'll create one if you
don't specify.)
Optional overrides (I'll compose these for you, unless you
explicitly specify them):
@param str/dict source: string or dict representing the source
database, along with authentication info, if any.
@param str/dict target: string or dict representing the
target database, possibly including authentication info.
@param dict user_ctx: User to act as.
Additional params you might specify (I won't pass these along
unless specified):
@param boolean: create_target: specifies whether or not to
create the target, if it doesn't already exist.
@param boolean continuous: set to True for a continuous replication.
"""
data = dict(
_id=repl_id if repl_id else unicode(uuid.uuid4()),
**kwargs
)
if not data.get('source'):
if source_db is None:
raise CloudantException(
u"You must specify either a source_db Database "
u"object or a manually composed 'source' string/dict."
)
data['source'] = {
"url": source_db.database_url,
"headers": {
"Authorization": source_db.creds['basic_auth']
}
}
if not data.get('target'):
if target_db is None:
raise CloudantException(
u"You must specify either a target_db Database "
u"object or a manually composed 'target' string/dict."
)
data['target'] = {
"url": target_db.database_url,
"headers": {
"Authorization": target_db.creds['basic_auth']
}
}
if not data.get('user_ctx'):
data['user_ctx'] = self.creds['user_ctx']
return self.create_document(data, throw_on_exists=True)
def list_replications(self):
"""
_list_replications_
Returns a list of all replications.
"""
docs = self.all_docs(include_docs="true")['rows']
return [doc['doc'] for doc in docs]
def replication_state(self, repl_id):
"""
_replication_state_
Get the state of the current replication. Possible values are
"triggered", "completed", "error", and None (this last in the
case where the replication is not yet triggered in couch).
@param str replication_id: id of the replication to inspect.
"""
try:
repl_doc = self[repl_id]
except KeyError:
raise CloudantException(
"Replication {} not found".format(repl_id)
)
repl_doc.fetch()
return repl_doc.get('_replication_state')
def follow_replication(self, repl_id):
"""
_follow_replication_
Block and stream status of a given replication.
@param str repl_id: id of the replication to follow
"""
def update_state():
try:
repl_doc = self[repl_id]
repl_doc.fetch()
state = repl_doc.get('_replication_state')
except KeyError:
repl_doc = None
state = None
return repl_doc, state
while True:
# Make sure we fetch the state up front, just in case it moves
# too fast and we miss it in the changes feed.
repl_doc, state = update_state()
if repl_doc:
yield repl_doc
if state is not None and state in ['error', 'completed']:
raise StopIteration
# Now listen on changes feed for the state
for change in self.changes():
if change.get('id') == repl_id:
repl_doc, state = update_state()
if repl_doc is not None:
yield repl_doc
if state is not None and state in ['error', 'completed']:
raise StopIteration
def stop_replication(self, repl_id):
""" Stop a given replication.
@param str repl_id: doc id of the replication to stop.
"""
try:
repl_doc = self[repl_id]
except KeyError:
raise CloudantException(
u"Could not find replication with id {}".format(repl_id))
repl_doc.fetch()
repl_doc.delete()