blob: bd11856bc83887255fb9e2c8d780d7942addc27a [file] [log] [blame]
#!/usr/bin/env python
#
# Camlistore uploader client for Python.
#
# Copyright 2010 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Client library for Camlistore."""
__author__ = 'Brett Slatkin (bslatkin@gmail.com)'
import base64
import cStringIO
import hashlib
import httplib
import logging
import mimetools
import urllib
import urlparse
import simplejson
__all__ = ['Error', 'ServerError', 'PayloadError', 'BUFFER_SIZE', 'CamliOp']
BUFFER_SIZE = 512 * 1024
class Error(Exception):
"""Base class for exceptions in this module."""
class ServerError(Error):
"""An unexpected error was returned by the server."""
class PayloadError(ServerError):
"""Something about a data payload was bad."""
def buffered_sha1(data, buffer_size=BUFFER_SIZE):
"""Calculates the sha1 hash of some data.
Args:
data: A string of data to write or an open file-like object. File-like
objects will be seeked back to their original position before this
function returns.
buffer_size: How much data to munge at a time.
Returns:
Hex sha1 string.
"""
compute = hashlib.sha1()
if isinstance(data, basestring):
compute.update(data)
else:
start = data.tell()
while True:
line = data.read(buffer_size)
if line == '':
break
compute.update(line)
data.seek(start)
return compute.hexdigest()
class CamliOp(object):
"""Camlistore client class that is single threaded, using one socket."""
def __init__(self,
server_address,
buffer_size=BUFFER_SIZE,
create_connection=httplib.HTTPConnection,
auth=None,
basepath=""):
"""Initializer.
Args:
server_address: hostname:port for the server.
buffer_size: Byte size to use for in-memory buffering for various
client-related operations.
create_connection: Use for testing.
auth: Optional. 'username:password' to use for HTTP basic auth.
basepath: Optional path suffix. e.g. if the server is at
"localhost:3179/bs", the basepath should be "/bs".
"""
self.server_address = server_address
self.buffer_size = buffer_size
self._create_connection = create_connection
self._connection = None
self._authorization = ''
self.basepath = ""
if auth:
if len(auth.split(':')) != 2:
# Default to dummy username; current server doesn't care
# TODO(jrabbit): care when necessary
auth = "username:" + auth #If username not given use the implicit default, 'username'
self._authorization = ('Basic ' + base64.encodestring(auth).strip())
if basepath:
if '/' not in basepath:
raise NameError("basepath must be in form '/bs'")
if basepath[-1] == '/':
basepath = basepath[:-1]
self.basepath = basepath
def _setup_connection(self):
"""Sets up the HTTP connection."""
self.connection = self._create_connection(self.server_address)
def put_blobs(self, blobs):
"""Puts a set of blobs.
Args:
blobs: List of (data, blobref) tuples; list of open files; or list of
blob data strings.
Returns:
The set of blobs that were actually uploaded. If all blobs are already
present this set will be empty.
Raises:
ServerError if the server response is bad.
PayloadError if the server response is not in the right format.
OSError or IOError if reading any blobs breaks.
"""
if isinstance(blobs, dict):
raise TypeError('Must pass iterable of tuples, open files, or strings.')
blobref_dict = {}
for item in blobs:
if isinstance(item, tuple):
blob, blobref = item
else:
blob, blobref = item, None
if blobref is None:
blobref = 'sha1-' + buffered_sha1(blob, buffer_size=self.buffer_size)
blobref_dict[blobref] = blob
preupload = {'camliversion': '1'}
for index, blobref in enumerate(blobref_dict.keys()):
preupload['blob%d' % (index+1)] = blobref
# TODO: What is the max number of blobs that can be specified in a
# preupload request? The server probably has some reasonable limit and
# after that we need to do batching in smaller groups.
self._setup_connection()
if self.basepath:
fullpath = self.basepath + '/camli/stat'
else:
fullpath = '/camli/stat'
self.connection.request(
'POST', fullpath, urllib.urlencode(preupload),
{'Content-Type': 'application/x-www-form-urlencoded',
'Authorization': self._authorization})
response = self.connection.getresponse()
logging.debug('Preupload HTTP response: %d %s',
response.status, response.reason)
if response.status != 200:
raise ServerError('Bad preupload response status: %d %s' %
(response.status, response.reason))
data = response.read()
try:
response_dict = simplejson.loads(data)
except simplejson.decoder.JSONDecodeError:
raise PayloadError('Server returned bad preupload response: %r' % data)
logging.debug('Parsed preupload response: %r', response_dict)
if 'stat' not in response_dict:
raise PayloadError(
'Could not find "stat" in preupload response: %r' %
response_dict)
if 'uploadUrl' not in response_dict:
raise PayloadError(
'Could not find "uploadUrl" in preupload response: %r' %
response_dict)
already_have_blobrefs = set()
for blobref_json in response_dict['stat']:
if 'blobRef' not in blobref_json:
raise PayloadError(
'Cannot find "blobRef" in preupload response: %r',
response_dict)
already_have_blobrefs.add(blobref_json['blobRef'])
logging.debug('Already have blobs: %r', already_have_blobrefs)
missing_blobrefs = set(blobref_dict.iterkeys())
missing_blobrefs.difference_update(already_have_blobrefs)
if not missing_blobrefs:
logging.debug('All blobs already present.')
return
# TODO(bslatkin): Figure out the 'Content-Length' header value by looking
# at the size of the files by seeking; required for multipart POST.
out = cStringIO.StringIO()
boundary = mimetools.choose_boundary()
boundary_start = '--' + boundary
blob_number = 0
for blobref in blobref_dict.iterkeys():
if blobref in already_have_blobrefs:
logging.debug('Already have blobref=%s', blobref)
continue
blob = blobref_dict[blobref]
blob_number += 1
out.write(boundary_start)
out.write('\r\nContent-Type: application/octet-stream\r\n')
out.write('Content-Disposition: form-data; name="%s"; '
'filename="%d"\r\n\r\n' % (blobref, blob_number))
if isinstance(blob, basestring):
out.write(blob)
else:
while True:
buf = blob.read(self.buffer_size)
if buf == '':
break
out.write(buf)
out.write('\r\n')
out.write(boundary_start)
out.write('--\r\n')
request_body = out.getvalue()
pieces = list(urlparse.urlparse(response_dict['uploadUrl']))
# TODO: Support upload servers on another base URL.
pieces[0], pieces[1] = '', ''
relative_url = urlparse.urlunparse(pieces)
self.connection.request(
'POST', relative_url, request_body,
{'Content-Type': 'multipart/form-data; boundary="%s"' % boundary,
'Content-Length': str(len(request_body)),
'Authorization': self._authorization})
response = self.connection.getresponse()
logging.debug('Upload response: %d %s', response.status, response.reason)
if response.status not in (200, 301, 302, 303):
raise ServerError('Bad upload response status: %d %s' %
(response.status, response.reason))
while response.status in (301, 302, 303):
# TODO(bslatkin): Support connections to servers on different addresses
# after redirects. For now just send another request to the same server.
location = response.getheader('Location')
pieces = list(urlparse.urlparse(location))
pieces[0], pieces[1] = '', ''
new_relative_url = urlparse.urlunparse(pieces)
logging.debug('Redirect %s -> %s', relative_url, new_relative_url)
relative_url = new_relative_url
self.connection.request('GET', relative_url)
response = self.connection.getresponse()
if response.status != 200:
raise ServerError('Bad upload response status: %d %s' %
(response.status, response.reason))
data = response.read()
try:
response_dict = simplejson.loads(data)
except simplejson.decoder.JSONDecodeError:
raise PayloadError('Server returned bad upload response: %r' % data)
if 'received' not in response_dict:
raise PayloadError('Could not find "received" in upload response: %r' %
response_dict)
received_blobrefs = set()
for blobref_json in response_dict['received']:
if 'blobRef' not in blobref_json:
raise PayloadError(
'Cannot find "blobRef" in upload response: %r',
response_dict)
received_blobrefs.add(blobref_json['blobRef'])
logging.debug('Received blobs: %r', received_blobrefs)
missing_blobrefs.difference_update(received_blobrefs)
if missing_blobrefs:
# TODO: Try to upload the missing ones.
raise ServerError('Some blobs not uploaded: %r', missing_blobrefs)
logging.debug('Upload of %d blobs successful.', len(blobref_dict))
return received_blobrefs
def get_blobs(self,
blobref_list,
start_out=None,
end_out=None,
check_sha1=True):
"""Gets a set of blobs.
Args:
blobref_list: A single blobref as a string or an iterable of strings that
are blobrefs.
start_out: Optional. A function taking the blobref's key, returns a
file-like object to which the blob should be written. Called before
the blob has started any writing.
end_out: Optional along with start_out. A function that takes the
blobref and open file-like object that does proper cleanup and closing
of the file. Called when all of the file's contents have been written.
check_sha1: Double-check that the file's contents match the blobref.
Returns:
If start_out is not supplied, then all blobs will be kept in memory. If
blobref_list is a single blobref, then the return value will be a string
with the blob data or None if the blob was not present. If blobref_list
was iterable, the return value will be a dictionary mapping blobref to
blob data for each blob that was found.
If start_out is supplied, the return value will be None. Callers can
check for missing blobs by comparing their own input of the blobref_list
argument to the blobrefs that are passed to start_out.
Raises:
ServerError if the server response is invalid for whatever reason.
OSError or IOError if writing to any files breaks.
"""
multiple = not isinstance(blobref_list, basestring)
result = {}
if start_out is None:
def start_out(blobref):
buffer = cStringIO.StringIO()
return buffer
def end_out(blobref, file_like):
result[blobref] = file_like.getvalue()
else:
result = None # Rely on user-supplied start_out for reporting blobrefs.
if end_out is None:
def end_out(blobref, file_like):
file_like.close()
self._setup_connection()
# Note, we could use a 'preupload' here as a quick, bulk existence check,
# but that may not always work depending on the access the user has.
# It's possible the user has read-only access, and thus can only do
# GET or HEAD on objects.
for blobref in blobref_list:
logging.debug('Getting blobref=%s', blobref)
if self.basepath:
fullpath = self.basepath + '/camli/'
else:
fullpath = '/camli/'
self.connection.request('GET', fullpath + blobref)
response = self.connection.getresponse()
if response.status == 404:
logging.debug('Server does not have blobref=%s', blobref)
continue
elif response.status != 200:
raise ServerError('Bad response status: %d %s' %
(response.status, response.reason))
if check_sha1:
compute_hash = hashlib.sha1()
out_file = start_out(blobref)
while True:
buf = response.read(self.buffer_size)
if buf == '':
end_out(blobref, out_file)
break
if check_sha1:
compute_hash.update(buf)
out_file.write(buf)
if check_sha1:
found = 'sha1-' + compute_hash.hexdigest()
if found != blobref:
raise ValueError('sha1 hash of blobref does not match; '
'found %s, expected %s' % (found, blobref))
if result and not multiple:
return result.values()[0]
return result