SQLAlchemy custom sorting algorithms using SQL indexes

Question

Is it possible to write custom sorting functions with indexes in SQLAlchemy? SQLite, for example, allows you to specify a sort function at the C level as sqlite3_create_collation().

An implementation of some Unicode sorting algorithm was provided by James Tauber here , which, for example, sorts all the closure “a” together, they have emphasis on them or not.

Other examples of why this might be useful are different orders of the alphabet (languages ​​other than English) and sorting of numerical values ​​(sorting 10after 9, not the order of codes).

Is this possible in SQLAlchemy? If not, is it supported by modules pysqlite3or MySQLdbor for any other SQL database modules supported by python?

Any information would be greatly appreciated.

Answer 1

The following is an example demonstrating the unicode sort algorithm for sqlite:

from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from pyuca import Collator

metadata = MetaData()
Base = declarative_base(metadata=metadata)

class Item(Base):
    __tablename__ = 'Item'
    id = Column(Integer, primary_key=True)
    value = Column(String, nullable=False)

collator = Collator('allkeys.txt')

def collate_unicode(value1, value2):
    return cmp(collator.sort_key(value1), collator.sort_key(value2))

engine = create_engine('sqlite://')
engine.raw_connection().create_collation('unicode', collate_unicode)
metadata.create_all(engine)
session = sessionmaker(engine)()

for word in [u"ĉambr", u"ĉar", u"car'", u"carin'", u"ĉe", u"ĉef'",
             u"centjar'", u"centr'", u"cerb'", u"cert'", u"ĉes'", u"ceter'"]:
    item = Item(value=word)
    session.add(item)
    session.commit()

for item in session.query(Item).order_by(collate(Item.value, 'unicode')):
    print item.value

Answer 2

I modified Denis Otkidach to answer a bit, so I will add my changes as a community wiki in case anyone else is interested:

# -*- coding: utf-8 -*-
from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import types
from pyuca import Collator

class MyUnicode(types.TypeDecorator):
    impl = types.Unicode
    def get_col_spec(self):
        # Return a new Unicode type sorted with 
        # the `mycollation` function
        return 'Unicode COLLATE mycollation'

# Create the collator (sorting) function/instance
collator = Collator('allkeys.txt')
def mycollation(value1, value2):
    if False:
        # Use pyuca for sorting
        return cmp(collator.sort_key(value1), 
                   collator.sort_key(value2))
    else:
        # Normalize to lowercased combining characters for sorting
        import unicodedata
        return cmp(unicodedata.normalize('NFD', unicode(value1)).lower(),
                   unicodedata.normalize('NFD', unicode(value2)).lower())

# Create a new metadata/base/table
metadata = MetaData()
Base = declarative_base(metadata=metadata)
class Item(Base):
    __tablename__ = 'CollatedTable'
    id = Column(Integer, primary_key=True)
    # (Note the `unique=True` in the next line so that an index 
    #  is created, therefore stored in collated order for faster SELECTs)
    value = Column(MyUnicode(), nullable=False, unique=True)

# Create a new database connection
engine = create_engine('sqlite://')
engine.echo = True # Print the SQL
engine.raw_connection().create_collation('mycollation', mycollation)
metadata.create_all(engine)
session = sessionmaker(engine)()

# Add some test data
for word in [u"ĉambr", u"ĉar", u"car'", u"carin'", u"ĉe", u"ĉef'",
             u"centjar'", u"centr'", u"cerb'", u"cert'", u"ĉes'", u"ceter'",

             u"zimble", u'bumble', 
             u'apple', u'ápple', u'ãpple',
             u'đjango', u'django']:
    item = Item(value=word)
    session.add(item)
session.commit()

for item in session.query(Item).order_by(Item.value): # collate(Item.value, 'mycollation')
    print item.value