Skip to content
This repository was archived by the owner on Aug 13, 2026. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
45 commits
Select commit Hold shift + click to select a range
29d624a
WIP transport refactor. Things are broken, do not use.
Dec 27, 2012
486c5a7
Actually connect to PBC by connecting the socket.
Dec 27, 2012
29cd619
Add transport method to client.
Dec 27, 2012
3208134
Fix some typos and comments from @ultimatebuster.
Dec 27, 2012
f60700b
Move client -> client/__init__.
Dec 27, 2012
f188f8e
Move connection-claiming into a new module, add retry logic.
Dec 27, 2012
12f596c
The node is a private attribute on the transport.
Dec 27, 2012
618f75a
Move http.py in preparation for refactoring.
Dec 27, 2012
f924239
Start breaking out HTTP into multiple files.
Dec 27, 2012
47cc10d
Add resource URL path generator methods, not wired up.
Dec 27, 2012
34078d2
Move HTTP request methods into a new file and wire up URL generators.
Dec 28, 2012
21c1c38
Make RETRY_COUNT a property and use a for loop over a range instead o…
Dec 28, 2012
6db4661
Fix gobs of pyflakes, pylint, pep8 problems.
Dec 28, 2012
991df2b
Fixed a bunch of test failures and stupid typos/bugs.
Dec 28, 2012
75c2e54
Fix the iteration logic in PBC streaming.
Dec 30, 2012
f263f2f
Fix get_buckets logic.
Dec 30, 2012
9adb48b
Add a print format for RiakBucket.
Dec 30, 2012
3ebede2
Fix PBC list-keys.
Dec 30, 2012
87a00bd
Fix search bug: shouldn't lowercase AND/OR/NOT in query string
Dec 30, 2012
8443fa0
Fix bug in index_path.
Dec 30, 2012
d3bc5e7
Fix encoding differences for binary objects between HTTP/PBC, 2.6/2.7.
Dec 30, 2012
117f00c
Fix a few pyflakes and pep8 issues.
Dec 30, 2012
6c3d673
Remove capitalized methods from RiakHttpConnection.
Dec 31, 2012
88f691c
Spec out missing methods.
Dec 31, 2012
af2faca
Use a dict to match message code with protobuf message, simplifying d…
Dec 31, 2012
80476ee
Fix protobuf build dependency.
Dec 31, 2012
2adb47d
Add key streaming to HTTP.
Dec 31, 2012
44a25e5
Simplify HTTP key streaming by pushing empty-result culling into the …
Dec 31, 2012
3004664
Add unit test for key streaming.
Dec 31, 2012
87ba442
Add streaming MapReduce to HTTP.
Dec 31, 2012
997d6e9
Fix some 'is' problems.
Jan 10, 2013
4b86364
Rethink the automatic retries stuff.
Jan 29, 2013
054a590
Restore _transport(), which is used when getting the client id.
Jan 29, 2013
add0760
Ensure that streams drain the socket and cleanup when there is an exc…
Jan 29, 2013
efc0918
PEP8 cleanup.
Jan 29, 2013
41df3a7
Cleanup some pyflakes errors.
Jan 29, 2013
76edabe
Test aborting a key stream.
Jan 29, 2013
7060c30
Document all the things
Jan 30, 2013
a680612
One last fix.
Jan 30, 2013
0abef5f
Remove some more 'is' in favor of '=='.
Jan 31, 2013
fb4e2dd
Remove commented imports.
Jan 31, 2013
8806150
Stylistic fix.
Jan 31, 2013
93019a8
Make read() private on HTTP stream iterators.
Jan 31, 2013
fdb1aa4
PEP8 line-length fix.
Jan 31, 2013
535d0bc
Remove retries from streaming per peer-review discussion.
Jan 31, 2013
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion .travis.yml
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@ language: python
python:
- "2.6"
- "2.7"
install: ./setup.py develop
install:
- ./setup.py develop
- ./setup.py easy_install protobuf
script: ./setup.py test
before_script: sudo /usr/sbin/search-cmd install searchbucket
notifications:
Expand Down
11 changes: 4 additions & 7 deletions riak/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,19 +32,16 @@


class RiakError(Exception):
"""
Base class for exceptions generated in the Riak API.
"""
def __init__(self, value):
self.value = value

def __str__(self):
return repr(self.value)

from riak_object import RiakObject
from bucket import RiakBucket
from client import RiakClient
from mapreduce import RiakMapReduce, RiakMapReducePhase, RiakLinkPhase,\
RiakKeyFilter
from transports.pbc import RiakPbcTransport
from transports.http import RiakHttpTransport
from mapreduce import RiakKeyFilter

ONE = "one"
ALL = "all"
Expand Down
84 changes: 58 additions & 26 deletions riak/bucket.py
Original file line number Diff line number Diff line change
Expand Up @@ -222,56 +222,76 @@ def _set_r(self, val):
def _get_r(self):
return self.get_property('r')

r = property(_get_r, _set_r)
r = property(_get_r, _set_r, doc="""
The default 'read' quorum for this bucket (how many replicas must
reply for a successful read). This should be an integer less than
the 'n_val' property, or a string of 'one', 'quorum', 'all', or
'default'""")

def _set_pr(self, val):
return self.set_property('pr', val)

def _get_pr(self):
return self.get_property('pr')

pr = property(_get_pr, _set_pr)
pr = property(_get_pr, _set_pr, doc="""
The default 'primary read' quorum for this bucket (how many
primary replicas are required for a successful read). This should
be an integer less than the 'n_val' property, or a string of
'one', 'quorum', 'all', or 'default'""")

def _set_rw(self, val):
return self.set_property('rw', val)

def _get_rw(self):
return self.get_property('rw')

rw = property(_get_rw, _set_rw)
rw = property(_get_rw, _set_rw, doc="""
The default 'read' and 'write' quorum for this bucket (equivalent
to 'r' and 'w' but for deletes). This should be an integer less
than the 'n_val' property, or a string of 'one', 'quorum', 'all',
or 'default'""")

def _set_w(self, val):
return self.set_property('w', val)

def _get_w(self):
return self.get_property('w')

w = property(_get_w, _set_w)
w = property(_get_w, _set_w, doc="""
The default 'write' quorum for this bucket (how many replicas must
acknowledge receipt of a write). This should be an integer less
than the 'n_val' property, or a string of 'one', 'quorum', 'all',
or 'default'""")

def _set_dw(self, val):
return self.set_property('dw', val)

def _get_dw(self):
return self.get_property('dw')

dw = property(_get_dw, _set_dw)
dw = property(_get_dw, _set_dw, doc="""
The default 'durable write' quorum for this bucket (how many
replicas must commit the write). This should be an integer less
than the 'n_val' property, or a string of 'one', 'quorum', 'all',
or 'default'""")

def _set_pw(self, val):
return self.set_property('pw', val)

def _get_pw(self):
return self.get_property('pw')

pw = property(_get_pw, _set_pw)
pw = property(_get_pw, _set_pw, doc="""
The default 'primary write' quorum for this bucket (how many
primary replicas are required for a successful write). This should
be an integer less than the 'n_val' property, or a string of
'one', 'quorum', 'all', or 'default'""")

def set_property(self, key, value):
"""
Set a bucket property.

.. warning::

This should only be used if you know what you are doing.

:param key: Property to set.
:type key: string
:param value: Property value.
Expand All @@ -296,24 +316,18 @@ def set_properties(self, props):
"""
Set multiple bucket properties in one call.

.. warning::

This should only be used if you know what you are doing.

:param props: An associative array of key:value.
:type props: array
:param props: A dictionary of properties
:type props: dict
"""
t = self._client.get_transport()
t.set_bucket_props(self, props)
self._client.set_bucket_props(self, props)

def get_properties(self):
"""
Retrieve an associative array of all bucket properties.
Retrieve a dict of all bucket properties.

:rtype: array
:rtype: dict
"""
t = self._client.get_transport()
return t.get_bucket_props(self)
return self._client.get_bucket_props(self)

def get_keys(self):
"""
Expand All @@ -323,7 +337,19 @@ def get_keys(self):

At current, this is a very expensive operation. Use with caution.
"""
return self._client.get_transport().get_keys(self)
return self._client.get_keys(self)

def stream_keys(self):
"""
Streams all keys within the bucket through an iterator.

.. warning::

At current, this is a very expensive operation. Use with caution.

:rtype: iterator
"""
return self._client.stream_keys(self)

def new_binary_from_file(self, key, filename):
"""
Expand All @@ -332,6 +358,10 @@ def new_binary_from_file(self, key, filename):
"""
binary_data = open(filename, "rb").read()
mimetype, encoding = mimetypes.guess_type(filename)
if encoding:
binary_data = bytearray(binary_data, encoding)
else:
binary_data = bytearray(binary_data)
if not mimetype:
mimetype = 'application/octet-stream'
return self.new_binary(key, binary_data, mimetype)
Expand Down Expand Up @@ -371,11 +401,13 @@ def search(self, query, **params):
"""
Queries a search index over objects in this bucket/index.
"""
return self._client.solr().search(self.name, query, **params)
return self._client.solr.search(self.name, query, **params)

def get_index(self, index, startkey, endkey=None):
"""
Queries a secondary index over objects in this bucket, returning keys.
"""
return self._client._transport.get_index(self.name, index, startkey,
endkey)
return self._client.get_index(self.name, index, startkey, endkey)

def __str__(self):
return '<RiakBucket "{0}">'.format(self.name)
Loading