ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

search: multi-word queries match by word, not the exact phrase

Browse Source 
Global search did a single ilike('%CSF Roles%'), so any query with more than one
word required the exact contiguous phrase and usually returned nothing. Add
_word_match: split the query into words and AND them (OR across the searched
columns per word), so 'CSF Roles' matches a record with both words in any field,
any order. Applied across every domain (assets, applications, KB, employees
[selfhosted + external HR], notifications, hostnames, IP, custom fields,
vendor/model/type). External HR path uses a parameterized per-word LIKE.
This commit is contained in:
cproudlock
2026-07-29 07:43:31 -04:00
parent b22701a444
commit 7a7c7f37d5
2 changed files with 96 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

117
shopdb/core/api/search.py
View File

@@ -23,6 +23,24 @@ logger = logging.getLogger(__name__)
search_bp = Blueprint('search', __name__)
def _word_match(query, *columns):
"""SQL clause matching rows that contain EVERY word of the query, each word
in any of the columns, in any order.
A plain `col.ilike('%CSF Roles%')` needs the exact contiguous phrase, so a
multi-word search returned nothing. This splits the query into words and ANDs
them (OR across the columns per word): 'CSF Roles' matches a row with 'CSF'
and 'Roles' anywhere in the searched fields.
"""
words = [w for w in query.split() if w]
if not words:
words = ['']
return db.and_(*[
db.or_(*[c.ilike(f'%{w}%') for c in columns])
for w in words
])
def _require_enabled(name):
"""Raise ImportError when the named plugin is disabled.
@@ -194,10 +212,7 @@ def _search_applications(query, search_term):
try:
apps = Application.query.filter(
Application.isactive == True,
db.or_(
Application.appname.ilike(search_term),
Application.appdescription.ilike(search_term)
)
_word_match(query, Application.appname, Application.appdescription)
).limit(10).all()
for app in apps:
@@ -228,10 +243,8 @@ def _search_knowledgebase(query, search_term):
from plugins.knowledgebase.models import KnowledgeBase
kb_articles = KnowledgeBase.query.filter(
KnowledgeBase.isactive == True,
db.or_(
KnowledgeBase.shortdescription.ilike(search_term),
KnowledgeBase.keywords.ilike(search_term)
)
_word_match(query, KnowledgeBase.shortdescription,
KnowledgeBase.keywords)
).limit(20).all()
for kb in kb_articles:
@@ -255,7 +268,7 @@ def _search_knowledgebase(query, search_term):
return results
def _employee_rows(search_term):
def _employee_rows(query):
"""Employee rows matching search_term, from the SELFHOSTED directory table
or the external HR DB per the employee_directory_mode setting. Returns dicts
shaped like the external query (SSO/First_Name/Last_Name/Team/Role)."""
@@ -267,25 +280,30 @@ def _employee_rows(search_term):
except ImportError:
return []
rows = DirectoryEmployee.query.filter(
db.or_(
DirectoryEmployee.firstname.ilike(search_term),
DirectoryEmployee.lastname.ilike(search_term),
db.cast(DirectoryEmployee.sso, db.String).ilike(search_term),
)
_word_match(query, DirectoryEmployee.firstname,
DirectoryEmployee.lastname,
db.cast(DirectoryEmployee.sso, db.String))
).order_by(DirectoryEmployee.lastname, DirectoryEmployee.firstname).limit(10).all()
return [{'SSO': e.sso, 'First_Name': e.firstname or '',
'Last_Name': e.lastname or '', 'Team': e.team, 'Role': e.role}
for e in rows]
# External HR DB: shared env-backed connection helper; never hardcode creds.
# Word-AND so "First Last" matches (each word in any name/SSO field), not just
# the exact contiguous phrase. Parameterized - one LIKE triple per word.
from shopdb.utils.employee_db import employee_connection
words = [w for w in query.split() if w] or ['']
clauses, params = [], []
for w in words:
clauses.append('(First_Name LIKE %s OR Last_Name LIKE %s '
'OR CAST(SSO AS CHAR) LIKE %s)')
like = f'%{w}%'
params += [like, like, like]
where = ' AND '.join(clauses)
emp_conn = employee_connection()
with emp_conn.cursor() as cur:
cur.execute('''
SELECT SSO, First_Name, Last_Name, Team, Role
FROM employees
WHERE First_Name LIKE %s OR Last_Name LIKE %s OR CAST(SSO AS CHAR) LIKE %s
ORDER BY Last_Name, First_Name LIMIT 10
''', (search_term, search_term, search_term))
cur.execute(
'SELECT SSO, First_Name, Last_Name, Team, Role FROM employees '
f'WHERE {where} ORDER BY Last_Name, First_Name LIMIT 10', params)
rows = cur.fetchall()
emp_conn.close()
return rows
@@ -295,7 +313,7 @@ def _search_employees(query, search_term):
"""Search employees (selfhosted directory or external HR DB, per mode)."""
results = []
try:
for emp in _employee_rows(search_term):
for emp in _employee_rows(query):
full_name = f"{emp['First_Name'].strip()} {emp['Last_Name'].strip()}"
sso_str = str(emp['SSO'])
@@ -329,12 +347,8 @@ def _search_assets(query, search_term):
joinedload(Asset.location),
).filter(
Asset.isactive == True,
db.or_(
Asset.assetnumber.ilike(search_term),
Asset.name.ilike(search_term),
Asset.serialnumber.ilike(search_term),
Asset.notes.ilike(search_term)
)
_word_match(query, Asset.assetnumber, Asset.name,
Asset.serialnumber, Asset.notes)
).limit(15).all()
for asset in assets:
@@ -372,12 +386,8 @@ def _search_measuringtools(query, search_term):
joinedload(Asset.location),
).filter(
Asset.isactive == True,
db.or_(
Asset.assetnumber.ilike(search_term),
Asset.name.ilike(search_term),
Asset.serialnumber.ilike(search_term),
Asset.gaugelabreference.ilike(search_term),
)
_word_match(query, Asset.assetnumber, Asset.name,
Asset.serialnumber, Asset.gaugelabreference)
).limit(15).all()
for asset in assets:
@@ -421,7 +431,7 @@ def _search_customfields(query, search_term):
Asset.isactive == True,
CustomField.searchable == True,
CustomField.isactive == True,
CustomFieldValue.value.ilike(search_term),
_word_match(query, CustomFieldValue.value),
).limit(15).all()
for asset, field in rows:
@@ -438,7 +448,7 @@ def _search_by_ip(query, search_term):
results = []
try:
comms = Communication.query.filter(
Communication.ipaddress.ilike(search_term)
_word_match(query, Communication.ipaddress)
).options(
joinedload(Communication.asset).joinedload(Asset.assettype),
joinedload(Communication.asset).joinedload(Asset.location),
@@ -498,7 +508,7 @@ def _search_hostnames(query, search_term):
_require_enabled('computers')
from plugins.computers.models import Computer
computers = Computer.query.filter(
Computer.hostname.ilike(search_term)
_word_match(query, Computer.hostname)
).options(
joinedload(Computer.asset).joinedload(Asset.assettype),
joinedload(Computer.asset).joinedload(Asset.location),
@@ -520,11 +530,8 @@ def _search_hostnames(query, search_term):
_require_enabled('printers')
from plugins.printers.models import Printer
printers = Printer.query.filter(
db.or_(
Printer.hostname.ilike(search_term),
Printer.sharename.ilike(search_term),
Printer.windowsname.ilike(search_term),
)
_word_match(query, Printer.hostname, Printer.sharename,
Printer.windowsname)
).options(
joinedload(Printer.asset).joinedload(Asset.assettype),
joinedload(Printer.asset).joinedload(Asset.location),
@@ -547,7 +554,7 @@ def _search_hostnames(query, search_term):
_require_enabled('network')
from plugins.network.models import NetworkDevice
devices = NetworkDevice.query.filter(
NetworkDevice.hostname.ilike(search_term)
_word_match(query, NetworkDevice.hostname)
).options(
joinedload(NetworkDevice.asset).joinedload(Asset.assettype),
joinedload(NetworkDevice.asset).joinedload(Asset.location),
@@ -577,10 +584,8 @@ def _search_notifications(query, search_term):
notifications = Notification.query.options(
joinedload(Notification.notificationtype)
).filter(
db.or_(
Notification.notification.ilike(search_term),
Notification.ticketnumber.ilike(search_term)
)
_word_match(query, Notification.notification,
Notification.ticketnumber)
).order_by(Notification.starttime.desc()).limit(15).all()
now = datetime.now(timezone.utc).replace(tzinfo=None)
@@ -635,11 +640,8 @@ def _search_vendor_model_type(query, search_term):
joinedload(Asset.location),
).filter(
Asset.isactive == True,
db.or_(
Vendor.vendor.ilike(search_term),
Model.modelnumber.ilike(search_term),
MachineType.machinetype.ilike(search_term)
)
_word_match(query, Vendor.vendor, Model.modelnumber,
MachineType.machinetype)
).limit(10).all()
for asset in machine_assets:
@@ -666,11 +668,8 @@ def _search_vendor_model_type(query, search_term):
joinedload(Asset.location),
).filter(
Asset.isactive == True,
db.or_(
Vendor.vendor.ilike(search_term),
Model.modelnumber.ilike(search_term),
PrinterType.printertype.ilike(search_term)
)
_word_match(query, Vendor.vendor, Model.modelnumber,
PrinterType.printertype)
).limit(10).all()
for asset in printer_assets:
@@ -695,10 +694,8 @@ def _search_vendor_model_type(query, search_term):
joinedload(Asset.location),
).filter(
Asset.isactive == True,
db.or_(
Vendor.vendor.ilike(search_term),
NetworkDeviceType.networkdevicetype.ilike(search_term)
)
_word_match(query, Vendor.vendor,
NetworkDeviceType.networkdevicetype)
).limit(10).all()
for asset in netdev_assets:

39
tests/test_core/test_search_multiword.py Normal file
View File

@@ -0,0 +1,39 @@
"""Global search matches multi-word queries by WORD (every word, any order or
field), not just the exact contiguous phrase.
Regression: 'CSF Roles' returned nothing because search did a single
ilike('%CSF Roles%'). Now each word must match (ORed across fields), so a record
with both words - even reversed or split - is found; a record with only one is
not.
"""
from shopdb.core.models import AssetType, Asset
def _seed(client, db, auth_headers, assetnumber, name):
if not AssetType.query.filter_by(assettype='computer').first():
db.session.add(AssetType(assettype='computer', pluginname='computer',
tablename='computer', description='c'))
db.session.commit()
r = client.post('/api/computers', json={'assetnumber': assetnumber},
headers=auth_headers)
assert r.status_code == 201, r.get_json()
asset = Asset.query.filter_by(assetnumber=assetnumber).first()
asset.name = name
db.session.commit()
def _titles(client, auth_headers, term):
r = client.get(f'/api/search?q={term}', headers=auth_headers)
assert r.status_code == 200, r.get_json()
return {h['title'] for h in r.get_json()['data']['results']}
def test_multiword_matches_all_words_any_order(client, db, auth_headers):
_seed(client, db, auth_headers, 'MW-1', 'CSF Roles Alpha') # contiguous
_seed(client, db, auth_headers, 'MW-2', 'Roles for the CSF Beta') # split/reversed
_seed(client, db, auth_headers, 'MW-3', 'CSF only') # one word only
titles = _titles(client, auth_headers, 'CSF Roles')
assert 'CSF Roles Alpha' in titles # exact phrase still works
assert 'Roles for the CSF Beta' in titles # words split/reversed now match
assert 'CSF only' not in titles # missing a word -> excluded