Skip to content

Instantly share code, notes, and snippets.

@xflr6
Last active June 4, 2022 13:37
Show Gist options
  • Save xflr6/c7c87befd9e88e8b1fcb0f042610eea4 to your computer and use it in GitHub Desktop.
Save xflr6/c7c87befd9e88e8b1fcb0f042610eea4 to your computer and use it in GitHub Desktop.
Register a Python stdlib re handler with sqlite3 create_function() to use the SQLite REGEXP operator under SQLAlchemy
"""Use Python re for sqlite3 REGEXP operator wíth SQLAlchemy.
added in https://docs.sqlalchemy.org/en/14/changelog/migration_14.html#support-for-sql-regular-expression-operators
"""
import re
from typing import Optional
import sqlalchemy as sa
import sqlalchemy.orm
ENGINE = sa.create_engine('sqlite://', echo=True)
REGISTRY = sa.orm.registry()
@sa.event.listens_for(sa.engine.Engine, 'connect')
def sqlite_engine_connect(dbapi_conn, connection_record) -> None:
dbapi_conn.create_function('regexp', 2, _regexp)
def _regexp(pattern: str, value: Optional[str]) -> Optional[bool]:
if value is None:
return None
return re.search(pattern, value) is not None
@REGISTRY.mapped
class Spam:
__tablename__ = 'spam'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Text)
REGISTRY.metadata.create_all(ENGINE)
sa.insert(Spam, bind=ENGINE).execute([{'name': 'spam'},
{'name': 'eggs'},
{'name': 'ham'},
{'name': None}])
query = sa.select(Spam).where(Spam.name.op('REGEXP')(r'(\w)\1'))
with ENGINE.connect() as conn:
print(conn.execute(query).fetchall())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment