Writing select functions for customized Pony

Question

A lot of times, I wrote queries like the following:

pony.orm.select(u for u in User if u.available and u.friends > 0 and ...)

So, I would like to write my own version of select, an alternative of it. Something like to avoid me writing every time the first part of the predicate, if u.available and u.friends > 0.

My question is more general: how can I write a function like select that it accepts arguments like those that select method or count method can accept.

Answer 1

Let's define a User entity in the following way:

from datetime import datetime, timedelta
from pony.orm import *

db = Database('sqlite', ':memory:')

class User(db.Entity):
    username = Required(str, unique=True)
    password = Required(str)
    friends = Set("User", reverse='friends')  # many-to-many symmetric relation
    online = Required(bool, default=False)    # if user is currently online
    last_visit = Optional(datetime)           # last visit time
    disabled = Required(bool, default=False)  # if user is disabled by administrator

sql_debug(True)
db.generate_mapping(create_tables=True)

Now we can define some convenient functions to retrieve most frequently used types of users. The first function will return the users who are not disabled by the admin:

def active_users():
    return User.select(lambda user: not user.disabled)

In that function I use select method of User entity which accepts a lambda function, but the same function can be written using global select function which accepts a generator expression:

def active_users():
    return select(u for u in User if not user.disabled)

The result of active_users function is a query object. You can call filter method of the query object to produce more specific query. For example, I can use active_users function to select active users whose names start with the 'A' letter:

users = active_users().filter(lambda user: user.name.startswith('A')) \
                      .order_by(User.name)[:10]

Now I want to find users who visit the site in a few last days. I can define another function which uses the query returned from the previous function and augment it in the following way:

def recent_users(days=1):
    return active_users().filter(lambda u: u.last_visit > datetime.now() - timedelta(days))

In this example I pass the days argument to the function and use its value inside the filter.

You can define a set of such functions which will form data access layer of your application. Some more examples:

def users_with_at_least_n_friends(n=1):
    return active_users().filter(lambda u: count(u.friends) >= n)

def online_users():
    return User.select(lambda u: u.online)

def online_users_with_at_least_n_online_friends(n=1):
    return online_users().filter(lambda u: count(f for f in u.friends if f.online) >= n)

users = online_users_with_at_least_n_online_friends(n=10) \
            .order_by(User.name)[:10]

In the examples above I define global functions. Another option is to define these functions as a classmethods of the User entity:

class User(db.Entity):
    username = Required(str, unique=True)
    ...
    @classmethod
    def name_starts_with(cls, prefix):
        return cls.select(lambda user: not user.disabled
                                       and user.name.startswith(prefix))

...
users = User.name_starts_with('A').order_by(desc(User.last_visit))[:10]

If you might want to have a generic function which can be applied to different entity classes, then you need to pass the entity class as a parameter. For example, if a number of different classes have the deleted attribute, and you want to have a generic method to select only non-deleted objects, you can write something like that:

def select_active(cls):
    return cls.select(lambda obj: not obj.deleted)

select_active(Message).filter(lambda msg: msg.author == current_user)

All functions provided above have one drawback - they are not composable. You cannot get a query from one function and augment it with another function. If you want to have a function which can augment existing query, that function should accept the query as an argument. Example:

def active_users():
    return User.select(lambda user: not user.disabled)

def name_starts_with(query, prefix):
    return query.filter(lambda user: user.name.startswith('prefix'))

The name_starts_with function can be applied to another query:

users1 = name_starts_with(active_users(), 'A').order_by(User.last_visited)
users2 = name_starts_with(recent_users(), 'B').filter(lambda user: user.online)

Also we are working on the query extension API which will allow a programmer to write custom query methods. When we release this API it will be possible to just chain custom query methods together in the following way:

select(u for u in User).recent(days=3).name_starts_with('A')[:10]

Hope I answered your question. If this is the case, please accept the answer as a correct one.