multi-column flask sqlalchemy full-text-search with postgresql

Fredy Yudiawan
1 min readOct 7, 2021

--

this is how to do full text search with postgresql without implementing elastic-search for startups.

Implementing dedicated search server like elasticsearch may not be cost effective depending on your project. Your database may only contains less than 1 million rows.

from datetime import datetime
from app import db, login_manager
...
from flask import current_app
from sqlalchemy import Index, func, cast, text
from sqlalchemy.dialects import postgresql
(model.py)
def to_tsvector_ix(*columns):
s = " || ' ' || ".join(columns)
return func.to_tsvector('english', text(s))
class Estate(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(150), nullable=False)
address = db.Column(db.String(512), nullable=False)
city = db.Column(db.String(100), nullable=False)
district = db.Column(db.String(100), nullable=False)
zip_code = db.Column(db.String(10))
land_area = db.Column(db.Integer())
bedroom_number = db.Column(db.Integer())
bathroom_number = db.Column(db.Integer())
maid_room = db.Column(db.Integer())
maid_bathroom = db.Column(db.Integer())
agent_name = db.Column(db.String(150))
agent_phone = db.Column(db.String(150))
is_rent = db.Column(db.Boolean())
price = db.Column(db.Float(), nullable=False)
lat = db.Column(db.Float())
lon = db.Column(db.Float())
posted = db.Column(db.DateTime, default=datetime.now)
last_update = db.Column(db.DateTime, default=datetime.now)
# add columns you want to search in __ts_vector__ variable
__ts_vector__ = to_tsvector_ix(
'name', 'address', 'city', 'district', 'zip_code'
)

__table_args__ = (Index('idx_estate_fts', __ts_vector__, postgresql_using='gin'),)

def __repr__(self):
return f'estate({self.name})'

after defining variable in the ORM model, you can use the __ts_vector__ in endpoints like this:

@search_test.route('/m_search/<search_param>', methods=['GET', 'POST'])
def estate_search(search_param):
results = Estate.query.filter(Estate.__ts_vector__.match(search_param, postgresql_regconfig='english')).all()
if results:
for result in results:
print(result.name)
else:
print("no result found")
return 'ok'

--

--