Object mapper for Amazon's DynamoDB

I do the following:

class AppDataModel(Model): """ The application data storage model. The indexed data includes:

    _country_code - ISO 3166-1 alpha-3, 3 character country code
    _app_type - application type, i.e. android, ios, etc. Always lower case
    _app_id - unique application identifier, usually a hash of some kind
    _app_version - version of application
    _analyzer_name - name of the analyzer that is responsible for some specific generated data
    _data_type - type of recorded data, i.e. log, call, output
    _creation_datetime - the ISO 8601 formatted date and time of when the data was created

The following indices are available:

    Global index gcc-index allowing searching by _country_code
    Global index gat-index allowing searching by _app_type
    Global index lan-index allowing searching by _analyzer_name

    Primary key:
        Hash key composed of _country_code, _app_type, _app_id, _app_version
        Range key composed of _analyzer_name

    Local index ldt-index allow searching within an application by _data_type
    Local index lc-index allow searching within an application by _creation_datetime
"""
__metadata__ = {
   'global_indexes': [
        GlobalIndex.all( 'gcc-index','country_code' ),
        GlobalIndex.all( 'gat-index','app_type' ),
        GlobalIndex.all( 'gboth-index','app_type','app_id')
    ]
}

country_code = Field( data_type=STRING )
app_type = Field( data_type=STRING )
app_id = Field( data_type=STRING )
app_version = Field( data_type=STRING )
composite_appid = Composite( 'country_code', 'app_type', 'app_id', 'app_version', hash_key=True )
analyzer_name = Field( data_type=STRING, index='lan-index' )
data_type = Field( data_type=STRING, index='ldt-index' )
data_index = Field( data_type=NUMBER )
creation_datetime = Field( data_type=datetime, index='lc-index' )
r_key = Composite( 'analyzer_name', 'data_type', 'data_index', 'creation_datetime', merge=c_composite, range_key=True)

def AppDataEngine(*_kwargs): # Connect to DynamoDB dyn = DynamoDBConnection( *_kwargs )

# Connect to the server and create the table schema for the 'appdata' table.
eng = Engine( dynamo=dyn, namespace=['appdata'] )
eng.register( AppDataModel )
eng.create_schema()

# Return the engine.
return( eng )

Later on I do this:

eng = AppDataEngine(**kwargs)

followed by this:

    # Create the instance for us to use...
    i = AppDataModel( country_code=self.cc, app_type=self.at,
                      app_id=self.aid, app_version=self.av,
                      analyzer_name=self.an, data_type='CALL',
                      creation_datetime=self.dt,
                    )

    # Write out the control data as a JSON struct
    i.data = { "WhoCalledMe":who_called_me, "args":args } )
    self.eng.save( i )

When I execute this code I get the following error returned:

oto.exception.JSONResponseError: JSONResponseError: 400 Bad Request {u'Message': u'Cannot do operations on a non-existent table', u'__type': u'com.amazonaws.dynamodb.v20120810#ResourceNotFoundException'}

Since I'm running the "local dynamodb" system I can check the database and see all the stuff in Sqlite. So I know the table is there since I see 'appdata-AppDataModel' in the list of tables.

There seems to be a bug where flywheel is trying to decode any string value for not defined name fields.

Conditions seem to be:

• name field present in dynamodb table while no corresponding field defined in model
• custom table name defined using __metadata__ and _name
• objects are obtained from dynamodb using scan and not get (did not check query yet)

Sample Code

from flywheel import Model, Field, Engine
from flywheel.fields.types import StringType
import uuid

class Item(Model):
    __metadata__ = {
        '_name': 'items'
    }
    uuid = Field(data_type=StringType(), hash_key=True, nullable=False)
    # name = Field(data_type=StringType())  # only not having this field defined will let the script crash

# setup
engine = Engine()
engine.connect_to_region('eu-west-1')
engine.register(Item)
engine.create_schema()

# create an item
unique_id = str(uuid.uuid1())
i = Item()
i.uuid = unique_id
i.name = 'a name' # using name (which is not a field) while _name is defined in __metadata__
engine.save(i)

# scan for
engine.get(Item, [unique_id]) # no crash
engine.scan(Item).filter().all() # crash happens here

Stack Trace

Traceback (most recent call last):
  File "addition_field.py", line 27, in <module>
    engine.scan(Item).filter().all() # crash happens here
  File "/usr/local/lib/python3.5/site-packages/flywheel/query.py", line 115, in all
    exclusive_start_key=exclusive_start_key))
  File "/usr/local/lib/python3.5/site-packages/flywheel/query.py", line 301, in gen
    yield self.model.ddb_load_(self.engine, result)
  File "/usr/local/lib/python3.5/site-packages/flywheel/models.py", line 493, in ddb_load_
    obj.set_ddb_val_(key, val)
  File "/usr/local/lib/python3.5/site-packages/flywheel/models.py", line 485, in set_ddb_val_
    setattr(self, key, Field.ddb_load_overflow(val))
  File "/usr/local/lib/python3.5/site-packages/flywheel/fields/__init__.py", line 265, in ddb_load_overflow
    return json.loads(val)
  File "/usr/local/Cellar/python3/3.5.2/Frameworks/Python.framework/Versions/3.5/lib/python3.5/json/__init__.py", line 319, in loads
    return _default_decoder.decode(s)
  File "/usr/local/Cellar/python3/3.5.2/Frameworks/Python.framework/Versions/3.5/lib/python3.5/json/decoder.py", line 339, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/usr/local/Cellar/python3/3.5.2/Frameworks/Python.framework/Versions/3.5/lib/python3.5/json/decoder.py", line 357, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

edit: I noticed something more. Actually running the script above works fine. But when you have a closer look at the content of the table / raw data returned by dynamo3for the scan you will notice the additional ": {'name': '"a name"', 'uuid': '5ebc745a-4def-11e6-864d-20c9d07f4883'}

Manually removing them (e.g. {'name': 'a name', 'uuid': '6b87e066-4def-11e6-a7b2-20c9d07f4883'}) will trigger the exception/bug described above. I didn't dig deeper why and where these " get inserted. But there is definitely a problem relying on them when the data in the table also gets touched by other software.

This is an absolutely fantastic library, by the way.

Since attribute names count against the 64kb limit for items and against provisioned iops (docs), it would be great to specify a short attribute name, and still refer to the field by a longer name.

For example:

class MyModel(Model):
    my_long_id = Field(hash_key=True, data_type=BINARY, attribute_name='id')
    unreasonably_long_field_name_for_data = Field(attribute_name='data')

I imagine that all external references would still use the model's meta field names - my_long_id - which would require an extra field_name -> attribute_name translation when constructing queries. Fallback when attribute_name isn't specified would use the existing behavior.

If you're interested, I can get to work on a pr.

This came about because I tried to change the name of a field before mapping it to an object as a workaround, but it didn't persist. That seems odd - maybe it's worth checking if field name is None here before overwriting it?

A simple test case:

f = Field(hash_key=True)
f.name = 'custom name'
print(f.name)  # prints 'custom name'
class MyModel(Model):
    id = f
print(MyModel.id.name)  # prints 'id'
print(f.name)  # prints 'id'

Here is the code I execute:

def c_composite( analyzer_name, data_type, data_idx, creation_time ): """ Custom composite handler for range key of AppDataModel """ return( '%s:%s:%s:%s' % (analyzer_name,data_type,str(data_idx),str(creation_time)) )

class AppDataModel(Model): metadata = { 'global_indexes': [ GlobalIndex.all( 'gcc-index','country_code' ), GlobalIndex.all( 'gat-index','app_type' ) ] }

country_code = Field( data_type=STRING )
app_type = Field( data_type=STRING )
app_id = Field( data_type=STRING )
app_version = Field( data_type=STRING )
composite_appid = Composite( 'country_code', 'app_type', 'app_id', 'app_version', hash_key=True )
analyzer_name = Field( data_type=STRING, index='lan-index' )
data_type = Field( data_type=STRING, index='ldt-index' )
data_index = Field( data_type=NUMBER )
creation_datetime = Field( data_type=datetime, index='lc-index' )
r_key = Composite( 'analyzer_name', 'data_type', 'data_index', 'creation_datetime', merge=c_composite, range_key=True)

def init(....)

Later on I do this:

dyn = DynamoDBConnection( **args )
eng = Engine( dynamo=dyn, namespace=['appdata'] )

test = AppDataModel( country_code='united states',app_type='android',app_id='123', app_version='1.0')
test.eng = eng

i = test.eng.query(AppDataModel).all()
for r in i:
    print r

I get this traceback:

Traceback (most recent call last): ... File .../flywheel/query.py line 79 in all

File ../flywheel/query.py line 48 in gen

File ../flywheel/fields/conditions.py line 54

ValueError: Bad query arguments. You must provide a hash key and may optionally constrain on exactly one range key

In fact when I check composite_appid it is set to u'USA:android:123:1.0' which is the composited hash key. Am I missing something?

Observation/Situation

When working with a multi-tenant AWS account, it's a reasonable practice to prefix table names with some sort of token to indicate what tenant/application a table belongs to. This same design pattern is also relevant for having tables for different environments, such as dev/qa/prod.

Hacky Failed Attempt

I attempted to handle this with a sort of configuration parameter for a table prefix, which I could them update the class models' __metadata__ attributes with, but it did not work:

APP_NAME = 'abc'
APP_ENV = 'dev'
TABLE_NAME_PREFIX = '-'.join([
    APP_NAME,
    APP_ENV,
])

# Register our models with the engine
for this_model in [
    OAuthUser,
    User,
]:
    # Insert the table name prefix
    if '_name' in this_model.__metadata__:
        print("Updating {}.__metadata__".format(this_model.__name__))
        print("  (old)_name='{}'".format(this_model.__metadata__['_name']))

        this_model.__metadata__.update({
            '_name': TABLE_NAME_PREFIX + this_model.__metadata__['_name'],
        })

        print("  (new)_name='{}'".format(this_model.__metadata__['_name']))

    engine.register(this_model)

engine.create_schema()

>>> Updating OAuthUser.__metadata__
>>>   (old)_name='oauth_user'
>>>   (new)_name='abc-dev_oauth_user'
>>> Updating User.__metadata__
>>>   (old)_name='user'
>>>   (new)_name='abc-dev_user'

This resulted in the creation of tables that were still only named oauth_user and user.

Potential Resolutions

• Add ability to the engine to set an engine-wide table prefix string

• Add ability to the engine to set an engine-wide table name function

  def table_namer(original_name):
      return "my-prefix_" + original_name
  
  engine.set_name_function(table_namer)
  

  This implementation may also resolve @stevearc's enhancement request in #4 as the table_namer function could independently handle the naming of

  the table with a timestamp:

  import datetime
  
  def table_namer(original_name):
      if original_name == 'timeseries_table':
          now = datetime.datetime.now()
          return ''.join([
              original_name,
              str(now.year),
              str(now.month),
          ])
      else:
          return original_name
  

First of all, thanks for creating flywheel. Very helpful.

I do have the following issue:

Version is 0.44 running on python 3.43 under Windows 7 The db structure is:

class scrobble(Model):
    artist = Field(hash_key=True)
    ts = Field(data_type=datetime, range_key=True)

Works fine when database is queried through boto or when exploring through the DynamoDB Web interface. I believe the issue is that for some reason the timestamps have a dozen or more digits to the right of the decimal point when I view them in the DynamoDB.

Whether that's it or not, all queries are throwing an Exception as follows:

>>> z = engine(scrobble).filter(artist="Lucinda Williams")
>>> z.first()
Traceback (most recent call last):
   File "<stdin>", line 1, in <module>
  File "C:\Python34\lib\site-packages\flywheel\query.py", line 155, in first
attributes=attributes, filter_or=filter_or):
  File "C:\Python34\lib\site-packages\flywheel\query.py", line 80, in gen
yield self.model.ddb_load_(self.engine, result)
  File "C:\Python34\lib\site-packages\flywheel\models.py", line 458, in ddb_load_
obj.set_ddb_val_(key, val)
  File "C:\Python34\lib\site-packages\flywheel\models.py", line 450, in set_ddb_val_
setattr(self, key, Field.ddb_load_overflow(val))
  File "C:\Python34\lib\site-packages\flywheel\fields\__init__.py", line 265, in ddb_load_overflow
return json.loads(val)
  File "C:\Python34\lib\json\__init__.py", line 318, in loads
return _default_decoder.decode(s)
  File "C:\Python34\lib\json\decoder.py", line 343, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "C:\Python34\lib\json\decoder.py", line 361, in raw_decode
    raise ValueError(errmsg("Expecting value", s, err.value)) from None
ValueError: Expecting value: line 1 column 1 (char 0)

There probably aren't many types that would care, but I came across this code while looking into attribute_names.

Any reason why None is always returning None, instead of delegating it to the data_type.ddb_dump call? I'd expect most custom types to check None anyway, and it doesn't leave much room for Types that care.

Unless there's a strong case for not supporting custom load/dump for None, this condition should probably be removed.

import uuid
import dynamo3
from flywheel import Field, Model, Engine
from flywheel.fields.types import TypeDefinition, STRING

class NonNullType(TypeDefinition):
    data_type = int
    ddb_data_type = STRING

    def coerce(self, value, force):
        return value

    def ddb_dump(self, value):
        if value is None:
            return ''
        return str(value)

    def ddb_load(self, value):
        if value == '':
            return None
        return int(value)

class MyModel(Model):
    __metadata__ = {
        '_name': str(uuid.uuid4())
    }
    pkey = Field(data_type=NonNullType(), hash_key=True)

dynamo = dynamo3.DynamoDBConnection.connect(...)
engine = Engine(dynamo=dynamo)
engine.register(MyModel)
engine.create_schema()

pk = None

from_typedef = NonNullType().ddb_dump(pk)
print(type(from_typedef))

m = MyModel(pkey=pk)
from_model = m.ddb_dump_field_('pkey')
print(type(from_model))

engine.sync(m)

And the output:

Traceback (most recent call last):
  File "crossj.py", line 77, in <module>
    engine.sync(m)
  File "/home/crossj/ws/flywheel/flywheel/engine.py", line 615, in sync
    self.save(item, overwrite=False)
  File "/home/crossj/ws/flywheel/flywheel/engine.py", line 465, in save
    **expected)
  File "build/bdist.linux-x86_64/egg/dynamo3/connection.py", line 382, in put_item
  File "build/bdist.linux-x86_64/egg/dynamo3/connection.py", line 207, in call
  File "build/bdist.linux-x86_64/egg/dynamo3/exception.py", line 46, in raise_if_error
dynamo3.exception.DynamoDBError: ValidationException: One or more parameter values were \
  invalid: Missing the key pkey in the item
Args: {'expected': {'pkey': {'ComparisonOperator': 'NULL'}},
 'item': {},
 'return_consumed_capacity': 'NONE',
 'return_values': 'NONE',
 'table_name': '91f18196-2d5c-41c3-ad1e-15d8f007e6a8'}

My company Runscope is probably going to use Flywheel pretty heavily as we expand our use of DynamoDB. We're already familiar with the basic ORM approach from our experience with SQLAlchemy. I personally built one project on top of Flywheel so far, and I've been pretty happy with it.

Anyway, we use table namespaces similar to what Flywheel supports, but we've already standardized on "." as the separator between elements. For example, our table names are prod.atlas.hosts and test.scorekeeper.requests. This commit allows Flywheel's namespace feature to work for us too.

This is my first pull request to the Flywheel project, so let me know if there are any stylistic norms that I should follow.

Unless I'm mistaken, undeclared / overflow field support was removed in version 5.0.0, https://github.com/stevearc/flywheel/commit/5cf757e1e1a2d6b3caf6ebf6732e7af36ba4b6cc

but support for undeclared fields is still mentioned in the documentation here: https://github.com/stevearc/flywheel/blame/master/doc/topics/models/basics.rst#L18

I'm new to your library, so I hope I'm not missing something. If so, please disregard.

Sync seems to update all changed attributes of a model.

Is there a way to update just one attribute (for instance for status updates)?

Similar to ActiveRecord's update_column.

It would be good to have support for dynamo3's connect method instead of having two separate methods connecting: connect_to_host and connect_to_region.

Also, dynamo3 seems to have deprecated the connect_to_host and connect_to_region methods in the version which is used by the latest flywheel release. So there are DeprecationWarnings everywhere.

Version: flywheel==0.4.4

I have encountered with an empty result issue when using "first" function on flywheel query object, but got result when using the "all" function (with the same query)

DB schema:

class User(Modal):
__metadata__ = {
        '_name': 'test_table',
        'global_indexes': [
            GlobalIndex('name-age-index', 'name', 'age_min').throughput(5, 3),
        ]
    }
name = Field(hash_key=True)
age_min = Field(data_type=int)      # OPTIONAL
age_max = Field(data_type=int)       # OPTIONAL

Query:

1. engine.query(User).filter(User.name == 'Erik', User.age_min <= age, User.age_max >= 18).first()
result: None

2.  engine.query(User).filter(User.name == 'Erik', User.age_min <= age, User.age_max >= 18).all()
result: [<object>]

It would be great if flywheel supported the creation of DynamoDB streams.

Has any work been done (or planned) on streams support? It's something I need, and may have time to take a look at integrating directly into flywheel, if you think it's useful. (Until then, I can dive in and do it imperatively through boto.)

I've observed a breaking bug in version 0.5 of flywheel. Took me awhile to figure out the issue.

........
    else:
                print('Setting {}'.format(arg))
                setattr(user, arg, args[arg])

        print(user.secret_question)
        print(user.ddb_dump_())
        user.update()

The result, I would see the secret question I just set on my user object, but in the result of user.ddb_dump_() all of my newly attrs were gone (and it appeared some values for attrs defined in the model statically were missing). Thus, none of my newly added attributes to my user object were saving and my project and tests were breaking when trying to update a model.

There is something wrong with the user.ddb_dump_() method.. I believe that method is called for every model object to save to the database?

For updating I use the engine.save(self, overwrite=True)

I had to downgrade to 0.4.11 and everything started saving and working again.

When a field is set to None due to an explicit None value, there could be the option to set it to its default value before validation/

I've written a field validator loop (https://github.com/mathcamp/flywheel/issues/29) and I've just recently added this check to it. Ideally you would want to check if nullable is false first but it looks like nullable isn't an attribute (it's just part of the init to append the not null check.)

It might be possible to do what I propose in a check method, but python isn't my strongest language.

  def validate(self):
    self._fields_with_errors = []
    for field in self.meta_.fields.values():
      try:
        #check if the field's value is None and if it has a default, if it's none, set it to it's default.
        if field.resolve(self) is None and field.default is not None:
          setattr(self, field.name, field.default)

        field.validate(self)
      except ValueError as e:
        res = re.search('Validation check on field (.*) failed for value (.*)', str(e))
        self._fields_with_errors.append(res.group(1))
    return len(self._fields_with_errors) == 0

My other suggestion is to treat "" as None for the nullable check for str field types. Or to at least have that option. This one's not a big deal and can easily be overcome with a check.

lambda x: x != ""

When there are more than 100 table in user's account and a table for a model already exists engine.create_schema() will fail with: ResourceInUseException: Table already exists: table-name

This happens because in engine.py line 204 the call to:

        tablenames = set(self.dynamo.list_tables())

will return 100 first tables and in model_meta.py line 440:

        if tablename in tablenames:
            return None

will miss the fact that the table already exists.

Proposed solution: For each table to create deliberately validate that it doesn't exist. Like (proto-code):

        for model in self._ddb_engine.models.itervalues():
            if not self._ddb_engine.dynamo.describe_table(model.meta_.ddb_tablename()):
                model.meta_.create_dynamo_schema(self._ddb_engine.dynamo, tablenames=[], wait=True)

I’d like to be able to provide an API to our users that is backed by a query that returns a large number of results. We’d like to be able to retrieve and store a cursor from a result set (query or scan), then create a new result set that picks up where the last one left off, possibly minutes later. I know that dynamo3 supports this internally, but it's not exported via flywheel yet.

Steve replied via email:

I think you'd have to have a new terminator on the query (like gen(), all()). I think you'd need page() to fetch a single page from DynamoDB so that you have some knowledge of where the database query left off. The default gen() implementation just provides a straight-up iterator that handles it all transparently. We could make page() return an object that extends list and has a 'cursor' attribute. Then you make the query terminators also take 'cursor' as a keyword argument, pass that through to dynamo3 and it should just work. I'll have to make a couple of changes in dynamo3 to support fetching pages instead of just iterators, but that should be pretty easy.