dynamodb-mock / ddbmock / router /

The default branch has multiple heads

# -*- coding: utf-8 -*-

# This module is not imported unless connect_boto is called thus making 'boto'
# an optional dependancy

# boto entry point: semantic name
from __future__ import absolute_import

import json, time, boto

from ddbmock.router import router
from ddbmock.errors import ValidationException, ConditionalCheckFailedException, DDBError
from boto.exception import DynamoDBResponseError
from boto.dynamodb.exceptions import (DynamoDBValidationError as DDBValidationErr,

# DDB to Boto exception
def _ddbmock_exception_to_boto_exception(err):
    if isinstance(err, ValidationException):
        return DDBValidationErr(err.status, err.status_str, err.to_dict())
    elif isinstance(err, ConditionalCheckFailedException):
        return DynamoDBConditionalCheckFailedError(err.status, err.status_str, err.to_dict())
        return DynamoDBResponseError(err.status, err.status_str, err.to_dict())

# Boto lib version entry point
def boto_router(self, action, body='', object_hook=None):
    """Replace boto.layer1.make_request to avoid the network layer. It also
    skips the authentication part. The external behavior is strictly compatible"""
    target = '%s_%s.%s' % (self.ServiceName, self.Version, action)
    start = time.time()

    post = json.loads(body)

        ret = router(action, post)
    except DDBError as e:
        raise _ddbmock_exception_to_boto_exception(e)
        boto.log.debug('RequestId: %s', post['request_id'])
        elapsed = (time.time() - start) * 1000'dynamodb %s: id=%s time=%sms', target, post['request_id'], int(elapsed))

    # FIXME: dump followed by load... can be better...
    ret = json.dumps(ret)
    return json.loads(ret, object_hook=object_hook)