1. Michael Bayer
  2. sqlalchemy

Wiki

Clone wiki

sqlalchemy / UsageRecipes / UniqueObject

Unique Object

Various patterns for instantiating an object that may or may not correspond to an existing row, kept unique on some field or set of fields.

Prerequisites

We'll define a function _unique() that will provide the "guts" to the unique recipe. This function is given a Session to work with, and associates a dictionary with the Session() which keeps track of current "unique" keys.

def _unique(session, cls, hashfunc, queryfunc, constructor, arg, kw):
    cache = getattr(session, '_unique_cache', None)
    if cache is None:
        session._unique_cache = cache = {}

    key = (cls, hashfunc(*arg, **kw))
    if key in cache:
        return cache[key]
    else:
        with session.no_autoflush:
            q = session.query(cls)
            q = queryfunc(q, *arg, **kw)
            obj = q.first()
            if not obj:
                obj = constructor(*arg, **kw)
                session.add(obj)
        cache[key] = obj
        return obj

The above function is given everything we need to check for an existing instance, query for it, or create:

widget = _unique(
            session,
            Widget,
            lambda name:name,
            lambda query, name:query.filter(Widget.name == name)
            Widget,
            (),
            {"name":"some name"}
            )

Some methods of integrating this method follows.

Explicit classmethod and session

This is the most straightforward way to go; a @classmethod is used to indicate the desire for a "unique" object. The Session is passed explicitly. The target class is augmented using a mixin, and it defines additional methods which produce the correct functionality for the "hash" and "query filter" functions.

class UniqueMixin(object):
    @classmethod
    def unique_hash(cls, *arg, **kw):
        raise NotImplementedError()

    @classmethod
    def unique_filter(cls, query, *arg, **kw):
        raise NotImplementedError()

    @classmethod
    def as_unique(cls, session, *arg, **kw):
        return _unique(
                    session,
                    cls,
                    cls.unique_hash,
                    cls.unique_filter,
                    cls,
                    arg, kw
               )

Usage:

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

engine = create_engine('sqlite://', echo=True)

Session = sessionmaker(bind=engine)

class Widget(UniqueMixin, Base):
    __tablename__ = 'widget'

    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True, nullable=False)

    @classmethod
    def unique_hash(cls, name):
        return name

    @classmethod
    def unique_filter(cls, query, name):
        return query.filter(Widget.name == name)

Base.metadata.create_all(engine)

session = Session()

w1, w2, w3 = Widget.as_unique(session, name='w1'), \
                Widget.as_unique(session, name='w2'), \
                Widget.as_unique(session, name='w3')
w1b = Widget.as_unique(session, name='w1')

assert w1 is w1b
assert w2 is not w3
assert w2 is not w1

session.commit()

Implicit constructor and scoped session

This version overrides __new__() to achieve a similar result. It also uses a global scoped_session. The target class is augmented using a class decorator, where lambdas are given to provide the functionality for the "hash" and "query filter" functions.

def unique_constructor(scoped_session, hashfunc, queryfunc):
    def decorate(cls):
        def _null_init(self, *arg, **kw):
            pass
        def __new__(cls, bases, *arg, **kw):
            # no-op __new__(), called
            # by the loading procedure
            if not arg and not kw:
                return object.__new__(cls)

            session = scoped_session()

            def constructor(*arg, **kw):
                obj = object.__new__(cls)
                obj._init(*arg, **kw)
                return obj

            return _unique(
                        session,
                        cls,
                        hashfunc,
                        queryfunc,
                        constructor,
                        arg, kw
                   )

        # note: cls must be already mapped for this part to work
        cls._init = cls.__init__
        cls.__init__ = _null_init
        cls.__new__ = classmethod(__new__)
        return cls

    return decorate

Usage:

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

engine = create_engine('sqlite://', echo=True)

Session = scoped_session(sessionmaker(bind=engine))

@unique_constructor(Session,
        lambda name: name,
        lambda query, name: query.filter(Widget.name == name)
)
class Widget(Base):
    __tablename__ = 'widget'

    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True, nullable=False)

Base.metadata.create_all(engine)

w1, w2, w3 = Widget(name='w1'), \
                Widget(name='w2'), \
                Widget(name='w3')
w1b = Widget(name='w1')

assert w1 is w1b
assert w2 is not w3
assert w2 is not w1

Session.commit()

Any combination of the above two!

The above approach can be implemented with any combination of: mixins or class decorators, additional methods on Widget or lambdas, scoped session or explicit session, @classmethod or overriding __new__().

Which one is more Pythonic? Likely the patterns present in the first version :).

Updated