sqlalchemy startswith

# -*- coding: utf-8 -*-

"""
Illustration of behaviour of startswith on PostgreSQL caused by lack of _ escaping.

"""

DB_URI="postgresql+psycopg2://marcink:qiwicudura@localhost/testdb"

import sqlalchemy
from sqlalchemy import orm, schema, sql, types

engine = sqlalchemy.create_engine(DB_URI)
metadata = sqlalchemy.MetaData(bind = engine)

######################################################################
# Create test table
######################################################################

test_table = sqlalchemy.Table(
    "test_table", metadata,
    schema.Column('value', types.Unicode(120)))

metadata.create_all(checkfirst=True)

######################################################################
# Insert test data
######################################################################

dbconn = engine.connect()

# Delete previous data if present
dbconn.execute(
    test_table.delete())

dbconn.execute(
    test_table.insert(),
    [
        {'value': u"a/b/c/d"},
        {'value': u"a_b/c/d"},
        {'value': u"a/b_c/d"},
        {'value': u"a_b_c_d"},
        {'value': u"x/y/z"},
        {'value': u"x%y%z"},
    ]
)

######################################################################
# Query
######################################################################

print u"* All items:"
qry = sql.select([test_table])
for row in engine.execute(qry):
    print u"   ", row['value']

print u"* Items starting with a/b/c:"
qry = sql.select([test_table]).where(test_table.c.value.startswith(u"a/b/c"))
#print qry
for row in engine.execute(qry):
    print u"   ", row['value']

print u"* Items starting with a_b_c:"
qry = sql.select([test_table]).where(test_table.c.value.startswith(u"a_b_c"))
for row in engine.execute(qry):
    print u"   ", row['value']

print u"* Items starting with a_b_c (manually escaped):"
qry = sql.select([test_table]).where(test_table.c.value.startswith(u"a\\_b\\_c"))
for row in engine.execute(qry):
    print u"   ", row['value']

print u"* Items starting with a_b_c (SA escaped):"
qry = sql.select([test_table]).where(test_table.c.value.startswith(u"a#_b#_c", escape='#'))
for row in engine.execute(qry):
    print u"   ", row['value']

print u"* Items starting with x/y:"
qry = sql.select([test_table]).where(test_table.c.value.startswith(u"x/y"))
#print qry
for row in engine.execute(qry):
    print u"   ", row['value']

print u"* Items starting with x%y:"
qry = sql.select([test_table]).where(test_table.c.value.startswith(u"x%y"))
for row in engine.execute(qry):
    print u"   ", row['value']

print u"* Items starting with x%y (manually escaped):"
qry = sql.select([test_table]).where(test_table.c.value.startswith(u"x\\%y"))
for row in engine.execute(qry):
    print u"   ", row['value']

print u"* Items starting with x%y (SA escaped):"
qry = sql.select([test_table]).where(test_table.c.value.startswith(u"x#%y", escape='#'))
for row in engine.execute(qry):
    print u"   ", row['value']