SQLAlchemy Sphinx

SQLAlchemy Sphinx is a dialect for SQLalchemy which converts SQLAlchemy model into compatible sql for sphinx.

This dialect works for both python 2 and 3. Currently you need to import sqlalchemy_sphinx to properly register for python 3.

Installation

SQLAlchemy Sphinx is available on pypi under the package name sqlalchemy-sphinx, you can get it by running:

pip install sqlalchemy-sphinx

Usage

Defining a Sphinx SQLAlchemy is exactly the same way you would create a sqlalchemy model.

from sqlalchemy import create_engine, Column, Integer, String, BigInteger, Unicode, Enum, or_, not_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import deferred, sessionmaker

sphinx_engine = create_engine('sphinx://your.sphinx.host:9008')
SphinxBase = declarative_base(bind=sphinx_engine)
SphinxSession = sessionmaker(bind=sphinx_engine)
sphinx_session = SphinxSession()


class MockSphinxModel(Base):
    __tablename__ = "mock_table"
    name = Column(String)
    id = Column(Integer, primary_key=True)
    country = Column(String)
    ranker = deferred(Column(String))
    group_by_dummy = deferred(Column(String))
    max_matches = deferred(Column(String))
    field_weights = deferred(Column(String))

After the model is created we can run queries against the model:

query = session.query(MockSphinxModel).limit(100)
# 'SELECT name, id, country FROM mock_table LIMIT 0, 100'

We can also do matching

base_query = session.query(MockSphinxModel.id)
query = base_query.filter(MockSphinxModel.country.match("US"))
# "SELECT id FROM mock_table WHERE MATCH('(@country US)')"

query = base_query.filter(MockSphinxModel.name.match("adriel"), MockSphinxModel.country.match("US"))
# "SELECT id FROM mock_table WHERE MATCH('(@name adriel) (@country US)')"

query = base_query.filter(not_(MockSphinxModel.name).match("US"))
# "SELECT id FROM mock_table WHERE MATCH('(@!name US)')"

query = base_query.filter(or_(MockSphinxModel.name, MockSphinxModel.country).match("US"))
# "SELECT id FROM mock_table WHERE MATCH('(@(name,country) US)')"

query = base_query.filter(not_(or_(MockSphinxModel.name, MockSphinxModel.country)).match("US"))
# "SELECT id FROM mock_table WHERE MATCH('(@!(name,country) US)')"

Options:

query = session.query(MockSphinxModel.id)
query = query.filter(func.options(MockSphinxModel.field_weights == ["title=10", "body=3"]))
# 'SELECT id FROM mock_table OPTION field_weights=(title=10, body=3)'

query = session.query(MockSphinxModel.id)
query = query.filter(MockSphinxModel.country.match("US"), func.options(MockSphinxModel.max_matches == 1))
# "SELECT id FROM mock_table WHERE MATCH('(@country US)') OPTION max_matches=1"