Wiki

Clone wiki

sqlalchemy / UsageRecipes / FunctionAttribute

FunctionAttribute

The goal is to have an attribute on an object that persists its value through a SQL function, and when loaded returns a value passed through another (or the same) SQL function. Queries should work too.

The synonym() construct makes this possible. We'll use as an example a field that on the Python side uses lowercase text but is persisted using uppercase text - all conversion is done via the database.

What we're really doing here is mapping two attributes, and then creating a third attribute as a Python descriptor which interacts with one of the attributes on the read side, and the other on the write side.

Here's the declarative version:

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

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

class Foo(Base):
    __tablename__ = 'foo'
    id = Column(Integer, primary_key=True)
    data = Column(String)

    _some_uppercase_thing = column_property(func.lower(data))

    def _get_some_uppercase_thing(self):
        return self._some_uppercase_thing

    def _set_some_uppercase_thing(self, data):
        self.data = func.upper(data)

    some_uppercase_thing = synonym(
                                '_some_uppercase_thing', 
                                descriptor=property(
                                            _get_some_uppercase_thing, 
                                            _set_some_uppercase_thing
                                            )
                                )

Base.metadata.create_all(engine)

sess = sessionmaker(engine)()

sess.add_all([
    Foo(some_uppercase_thing="value 1"),
    Foo(some_uppercase_thing="value 2"),
    Foo(some_uppercase_thing="value 3"),
    Foo(some_uppercase_thing="value 4"),
])

sess.commit()

assert sess.query(Foo.some_uppercase_thing).all() == [
    ("value 1", ),
    ("value 2", ),
    ("value 3", ),
    ("value 4", ),
]

assert sess.query(Foo.some_uppercase_thing).filter(Foo.some_uppercase_thing == 'value 2').all() == [
    ("value 2", ),
]

assert sess.execute(select([Foo.__table__.c.data])).fetchall() == [
    ("VALUE 1", ),
    ("VALUE 2", ),
    ("VALUE 3", ),
    ("VALUE 4", ),
]

Here's the plain mapper version:

from sqlalchemy import *
from sqlalchemy.orm import *

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

metadata = MetaData()

foo_table = Table('foo', metadata,
    Column('id', Integer, primary_key=True),
    Column('data', String)
)

class Foo(object):
    def __init__(self, some_uppercase_thing):
        self.some_uppercase_thing = some_uppercase_thing



    def _get_some_uppercase_thing(self):
        return self._some_uppercase_thing

    def _set_some_uppercase_thing(self, data):
        self.data = func.upper(data)


mapper(Foo, foo_table, properties={
    '_some_uppercase_thing' : column_property(func.lower(foo_table.c.data)),
    'some_uppercase_thing' : synonym(
                                '_some_uppercase_thing', 
                                descriptor=property(
                                            Foo._get_some_uppercase_thing, 
                                            Foo._set_some_uppercase_thing
                                            )
                                )
})

metadata.create_all(engine)

sess = sessionmaker(engine)()

sess.add_all([
    Foo(some_uppercase_thing="value 1"),
    Foo(some_uppercase_thing="value 2"),
    Foo(some_uppercase_thing="value 3"),
    Foo(some_uppercase_thing="value 4"),
])

sess.commit()

assert sess.query(Foo.some_uppercase_thing).all() == [
    ("value 1", ),
    ("value 2", ),
    ("value 3", ),
    ("value 4", ),
]

assert sess.query(Foo.some_uppercase_thing).filter(Foo.some_uppercase_thing == 'value 2').all() == [
    ("value 2", ),
]

assert sess.execute(select([foo_table.c.data])).fetchall() == [
    ("VALUE 1", ),
    ("VALUE 2", ),
    ("VALUE 3", ),
    ("VALUE 4", ),
]

Updated