Hacks

Collected hacks using peewee. Have a cool hack you’d like to share? Open an issue on GitHub or contact me.

Optimistic Locking

Optimistic locking is useful in situations where you might ordinarily use a SELECT FOR UPDATE (or in SQLite, BEGIN IMMEDIATE). For example, you might fetch a user record from the database, make some modifications, then save the modified user record. Typically this scenario would require us to lock the user record for the duration of the transaction, from the moment we select it, to the moment we save our changes.

In optimistic locking, on the other hand, we do not acquire any lock and instead rely on an internal version column in the row we’re modifying. At read time, we see what version the row is currently at, and on save, we ensure that the update takes place only if the version is the same as the one we initially read. If the version is higher, then some other process must have snuck in and changed the row – to save our modified version could result in the loss of important changes.

It’s quite simple to implement optimistic locking in Peewee, here is a base class that you can use as a starting point:

  1. from peewee import *
  3. class BaseVersionedModel(Model):
  4.     version = IntegerField(default=1, index=True)
  6.     def save_optimistic(self):
  7.         if not self.id:
  8.             # This is a new record, so the default logic is to perform an
  9.             # INSERT. Ideally your model would also have a unique
  10.             # constraint that made it impossible for two INSERTs to happen
  11.             # at the same time.
  12.             return self.save()
  14.         # Update any data that has changed and bump the version counter.
  15.         field_data = dict(self._data)
  16.         current_version = field_data.pop('version', 1)
  17.         field_data = self._prune_fields(field_data, self.dirty_fields)
  18.         if not field_data:
  19.             raise ValueError('No changes have been made.')
  21.         ModelClass = type(self)
  22.         field_data['version'] = ModelClass.version + 1  # Atomic increment.
  24.         query = ModelClass.update(**field_data).where(
  25.             (ModelClass.version == current_version) &
  26.             (ModelClass.id == self.id))
  27.         if query.execute() == 0:
  28.             # No rows were updated, indicating another process has saved
  29.             # a new version. How you handle this situation is up to you,
  30.             # but for simplicity I'm just raising an exception.
  31.             raise ConflictDetectedException()
  32.         else:
  33.             # Increment local version to match what is now in the db.
  34.             self.version += 1
  35.             return True

Here’s an example of how this works. Let’s assume we have the following model definition. Note that there’s a unique constraint on the username – this is important as it provides a way to prevent double-inserts.

  1. class User(BaseVersionedModel):
  2.     username = CharField(unique=True)
  3.     favorite_animal = CharField()

Example:

  1. >>> u = User(username='charlie', favorite_animal='cat')
  2. >>> u.save_optimistic()
  3. True
  5. >>> u.version
  6. 1
  8. >>> u.save_optimistic()
  9. Traceback (most recent call last):
  10.   File "<stdin>", line 1, in <module>
  11.   File "x.py", line 18, in save_optimistic
  12.     raise ValueError('No changes have been made.')
  13. ValueError: No changes have been made.
  15. >>> u.favorite_animal = 'kitten'
  16. >>> u.save_optimistic()
  17. True
  19. # Simulate a separate thread coming in and updating the model.
  20. >>> u2 = User.get(User.username == 'charlie')
  21. >>> u2.favorite_animal = 'macaw'
  22. >>> u2.save_optimistic()
  23. True
  25. # Now, attempt to change and re-save the original instance:
  26. >>> u.favorite_animal = 'little parrot'
  27. >>> u.save_optimistic()
  28. Traceback (most recent call last):
  29.   File "<stdin>", line 1, in <module>
  30.   File "x.py", line 30, in save_optimistic
  31.     raise ConflictDetectedException()
  32. ConflictDetectedException: current version is out of sync

Top object per group

These examples describe several ways to query the single top item per group. For a thorough discuss of various techniques, check out my blog post Querying the top item by group with Peewee ORM. If you are interested in the more general problem of querying the top N items, see the section below Top N objects per group.

In these examples we will use the User and Tweet models to find each user and their most-recent tweet.

The most efficient method I found in my testing uses the MAX() aggregate function.

We will perform the aggregation in a non-correlated subquery, so we can be confident this method will be performant. The idea is that we will select the posts, grouped by their author, whose timestamp is equal to the max observed timestamp for that user.

  1. # When referencing a table multiple times, we'll call Model.alias() to create
  2. # a secondary reference to the table.
  3. TweetAlias = Tweet.alias()
  5. # Create a subquery that will calculate the maximum Tweet create_date for each
  6. # user.
  7. subquery = (TweetAlias
  8.             .select(
  9.                 TweetAlias.user,
  10.                 fn.MAX(TweetAlias.create_date).alias('max_ts'))
  11.             .group_by(TweetAlias.user)
  12.             .alias('tweet_max_subquery'))
  14. # Query for tweets and join using the subquery to match the tweet's user
  15. # and create_date.
  16. query = (Tweet
  17.          .select(Tweet, User)
  18.          .join(User)
  19.          .switch(Tweet)
  20.          .join(subquery, on=(
  21.              (Tweet.create_date == subquery.c.max_ts) &
  22.              (Tweet.user == subquery.c.user_id))))

SQLite and MySQL are a bit more lax and permit grouping by a subset of the columns that are selected. This means we can do away with the subquery and express it quite concisely:

  1. query = (Tweet
  2.          .select(Tweet, User)
  3.          .join(User)
  4.          .group_by(Tweet.user)
  5.          .having(Tweet.create_date == fn.MAX(Tweet.create_date)))

Top N objects per group

These examples describe several ways to query the top N items per group reasonably efficiently. For a thorough discussion of various techniques, check out my blog post Querying the top N objects per group with Peewee ORM.

In these examples we will use the User and Tweet models to find each user and their three most-recent tweets.

Postgres lateral joins

Lateral joins are a neat Postgres feature that allow reasonably efficient correlated subqueries. They are often described as SQL for each loops.

The desired SQL is:

  1. SELECT * FROM
  2.   (SELECT t2.id, t2.username FROM user AS t2) AS uq
  3.    LEFT JOIN LATERAL
  4.   (SELECT t2.message, t2.create_date
  5.    FROM tweet AS t2
  6.    WHERE (t2.user_id = uq.id)
  7.    ORDER BY t2.create_date DESC LIMIT 3)
  8.   AS pq ON true

To accomplish this with peewee we’ll need to express the lateral join as a Clause, which gives us greater flexibility than the join() method.

  1. # We'll reference `Tweet` twice, so keep an alias handy.
  2. TweetAlias = Tweet.alias()
  4. # The "outer loop" will be iterating over the users whose
  5. # tweets we are trying to find.
  6. user_query = User.select(User.id, User.username).alias('uq')
  8. # The inner loop will select tweets and is correlated to the
  9. # outer loop via the WHERE clause. Note that we are using a
  10. # LIMIT clause.
  11. tweet_query = (TweetAlias
  12.                .select(TweetAlias.message, TweetAlias.create_date)
  13.                .where(TweetAlias.user == user_query.c.id)
  14.                .order_by(TweetAlias.create_date.desc())
  15.                .limit(3)
  16.                .alias('pq'))
  18. # Now we join the outer and inner queries using the LEFT LATERAL
  19. # JOIN. The join predicate is *ON TRUE*, since we're effectively
  20. # joining in the tweet subquery's WHERE clause.
  21. join_clause = NodeList((
  22.     user_query,
  23.     SQL('LEFT JOIN LATERAL'),
  24.     tweet_query,
  25.     SQL('ON %s', [True])))
  27. # Finally, we'll wrap these up and SELECT from the result.
  28. query = (Tweet
  29.          .select(user_query.c.username, tweet_query.c.message,
  30.                  tweet_query.c.create_date)
  31.          .from_(join_clause))

Window functions

Window functions, which are supported by peewee, provide scalable, efficient performance.

The desired SQL is:

  1. SELECT subq.message, subq.username
  2. FROM (
  3.     SELECT
  4.         t2.message,
  5.         t3.username,
  6.         RANK() OVER (
  7.             PARTITION BY t2.user_id
  8.             ORDER BY t2.create_date DESC
  9.         ) AS rnk
  10.     FROM tweet AS t2
  11.     INNER JOIN user AS t3 ON (t2.user_id = t3.id)
  12. ) AS subq
  13. WHERE (subq.rnk <= 3)

To accomplish this with peewee, we will wrap the ranked Tweets in an outer query that performs the filtering.

  1. TweetAlias = Tweet.alias()
  3. # The subquery will select the relevant data from the Tweet and
  4. # User table, as well as ranking the tweets by user from newest
  5. # to oldest.
  6. subquery = (TweetAlias
  7.             .select(
  8.                 TweetAlias.message,
  9.                 User.username,
  10.                 fn.RANK().over(
  11.                     partition_by=[TweetAlias.user],
  12.                     order_by=[TweetAlias.create_date.desc()]).alias('rnk'))
  13.             .join(User, on=(TweetAlias.user == User.id))
  14.             .alias('subq'))
  16. # Since we can't filter on the rank, we are wrapping it in a query
  17. # and performing the filtering in the outer query.
  18. query = (Tweet
  19.          .select(subquery.c.message, subquery.c.username)
  20.          .from_(subquery)
  21.          .where(subquery.c.rnk <= 3))

Other methods

If you’re not using Postgres, then unfortunately you’re left with options that exhibit less-than-ideal performance. For a more complete overview of common methods, check out this blog post. Below I will summarize the approaches and the corresponding SQL.

Using COUNT, we can get all tweets where there exist less than N tweets with more recent timestamps:

  1. TweetAlias = Tweet.alias()
  3. # Create a correlated subquery that calculates the number of
  4. # tweets with a higher (newer) timestamp than the tweet we're
  5. # looking at in the outer query.
  6. subquery = (TweetAlias
  7.             .select(fn.COUNT(TweetAlias.id))
  8.             .where(
  9.                 (TweetAlias.create_date >= Tweet.create_date) &
  10.                 (TweetAlias.user == Tweet.user)))
  12. # Wrap the subquery and filter on the count.
  13. query = (Tweet
  14.          .select(Tweet, User)
  15.          .join(User)
  16.          .where(subquery <= 3))

We can achieve similar results by doing a self-join and performing the filtering in the HAVING clause:

  1. TweetAlias = Tweet.alias()
  3. # Use a self-join and join predicates to count the number of
  4. # newer tweets.
  5. query = (Tweet
  6.          .select(Tweet.id, Tweet.message, Tweet.user, User.username)
  7.          .join(User)
  8.          .switch(Tweet)
  9.          .join(TweetAlias, on=(
  10.              (TweetAlias.user == Tweet.user) &
  11.              (TweetAlias.create_date >= Tweet.create_date)))
  12.          .group_by(Tweet.id, Tweet.content, Tweet.user, User.username)
  13.          .having(fn.COUNT(Tweet.id) <= 3))

The last example uses a LIMIT clause in a correlated subquery.

  1. TweetAlias = Tweet.alias()
  3. # The subquery here will calculate, for the user who created the
  4. # tweet in the outer loop, the three newest tweets. The expression
  5. # will evaluate to `True` if the outer-loop tweet is in the set of
  6. # tweets represented by the inner query.
  7. query = (Tweet
  8.          .select(Tweet, User)
  9.          .join(User)
  10.          .where(Tweet.id << (
  11.              TweetAlias
  12.              .select(TweetAlias.id)
  13.              .where(TweetAlias.user == Tweet.user)
  14.              .order_by(TweetAlias.create_date.desc())
  15.              .limit(3))))

Writing custom functions with SQLite

SQLite is very easy to extend with custom functions written in Python, that are then callable from your SQL statements. By using the SqliteExtDatabase and the func() decorator, you can very easily define your own functions.

Here is an example function that generates a hashed version of a user-supplied password. We can also use this to implement login functionality for matching a user and password.

  1. from hashlib import sha1
  2. from random import random
  3. from playhouse.sqlite_ext import SqliteExtDatabase
  5. db = SqliteExtDatabase('my-blog.db')
  7. def get_hexdigest(salt, raw_password):
  8.     data = salt + raw_password
  9.     return sha1(data.encode('utf8')).hexdigest()
  11. @db.func()
  12. def make_password(raw_password):
  13.     salt = get_hexdigest(str(random()), str(random()))[:5]
  14.     hsh = get_hexdigest(salt, raw_password)
  15.     return '%s$%s' % (salt, hsh)
  17. @db.func()
  18. def check_password(raw_password, enc_password):
  19.     salt, hsh = enc_password.split('$', 1)
  20.     return hsh == get_hexdigest(salt, raw_password)

Here is how you can use the function to add a new user, storing a hashed password:

  1. query = User.insert(
  2.     username='charlie',
  3.     password=fn.make_password('testing')).execute()

If we retrieve the user from the database, the password that’s stored is hashed and salted:

  1. >>> user = User.get(User.username == 'charlie')
  2. >>> print user.password
  3. b76fa$88be1adcde66a1ac16054bc17c8a297523170949

To implement login-type functionality, you could write something like this:

  1. def login(username, password):
  2.     try:
  3.         return (User
  4.                 .select()
  5.                 .where(
  6.                     (User.username == username) &
  7.                     (fn.check_password(password, User.password) == True))
  8.                 .get())
  9.     except User.DoesNotExist:
  10.         # Incorrect username and/or password.
  11.         return False