SQLAlchemy getting substring in a query

Question

I have a dev db with SQLite (which unit test run as well) and prod db with MySQL.

I need to write a query with SQLAlchemy which uses a substring in a WHERE statement. I was trying to use func, but it translates it without modification for a specific DB engine.

My query is:

MyTable.field == func.substring_index(OtherTable.other_field, ":", 1)

So I basically want to split a value by ":" and take the first part.

The problem is it's translated with substring_index for SQLite, which is incorrect. Is there a way to use substring in a WHERE clause?

Answer 1

SQLAlchemy supports custom SQL constructs and compilation extensions and registering named functions. With these you can register substring_index() as a function with special treatment for SQLite:

from sqlalchemy.sql.functions import GenericFunction
from sqlalchemy.types import String
from sqlalchemy.ext.compiler import compiles

class substring_index(GenericFunction):
    type = String

@compiles(substring_index, 'sqlite')
def compile_substring_index_sqlite(element, compiler, **kw):
    s, delim, count = element.clauses

    # This assumes that count is a `bindparam`, produced from passing
    # literal integer to `func.substring_index()`.
    assert count.value == 1, "INSTR(X, Y) only supports first occurrence"

    s = compiler.process(s, **kw)
    delim = compiler.process(delim, **kw)

    return f"substr({s}, 1, instr({s}, {delim}) - 1)"

Another option would be to register a Python function in SQLite as substring_index():

from sqlalchemy import event

def sqlite_substring_index(s, delim, count):
    parts = s.split(delim)

    if count > 0:
        parts = parts[:count]

    else:
        parts = parts[count:]

    return delim.join(parts)

# In your SQLite branch, before anything else DB related is performed:
@event.listens_for(engine, 'connect')
def create_functions(dbapi_connection, connection_record):
    dbapi_connection.create_function('substring_index', 3, sqlite_substring_index)

With the function in place you can call it just as you would in MySQL.

---

As to why this is not a part of the SQLAlchemy library out of the box, it would be an endless battle due to different DBMS supporting wildly different functions. For example the names of some trigonometric functions vary, and SQLite does not provide them at all out of the box. Supporting different SQL DBMS in a single code base is non-trivial, and more often than not just not worth it.

Answer 2

You could check the database dialect name, and based on that make the substring. For example:

def substring(column, delimeter):
    if session.bind.dialect.name == 'sqlite':
        return func.substr(column, 1, func.instr(column, delimeter) - 1)
    elif session.bind.dialect.name == 'mysql':
        return func.substring_index(column, delimeter, 1)

And then replace your filter condition with:

MyTable.field == substring(OtherTable.other_field, ":")