原文：https://dev.to/priyasundaram/add-real-search-to-your-python-web-app-no-elasticsearch-no-separate-service-8la（作者 @priyasundaram）
每个应用最终都会遇到比 WHERE title LIKE '%...%' 更好的搜索需求。LIKE 无法对结果排序，不知道 running 应该匹配 run，而且一旦表数据量可观，查询就会变慢。常规的下一步——搭一个 Elasticsearch 或 OpenSearch——意味着引入 JVM、一个需要维护的集群，以及一个需要保持同步的独立数据存储。对大量应用来说，这就像用起重机挂相框。
whoosh3 是一个纯 Python 全文搜索库（Whoosh 的维护版延续）。pip install whoosh3 之后，你就拥有 BM25 排序、词干提取、真正的查询语言，以及一个只是磁盘上一个文件夹的索引——它在你的进程内运行，不需要服务器。下面介绍如何把它接入 Web 应用，包括那些容易踩坑的运维细节。
按你的需求设计排序的 Schema
from whoosh.fields import Schema, TEXT, ID, NUMERIC
from whoosh.analysis import StemmingAnalyzer
schema = Schema(
id=ID(stored=True, unique=True),
title=TEXT(analyzer=StemmingAnalyzer(), stored=True, field_boost=2.0),
body=TEXT(analyzer=StemmingAnalyzer(), stored=True),
views=NUMERIC(stored=True, sortable=True),
)
这里有三个选择起着实际作用。id 上的 unique=True 让索引支持 upsert——用相同 id 写入两次会替换该行而不是重复插入。StemmingAnalyzer 让 widget 能匹配到 widgets。field_boost=2.0 表示标题中的匹配权重是正文的两倍，这样正确的结果会自动浮到顶部，无需手动打分。
每个应用都需要的三个操作
索引只创建一次（它存在于一个目录中），之后所有操作就是 upsert、删除和搜索：
from whoosh import index
from whoosh.qparser import MultifieldParser, OrGroup
from whoosh.writing import AsyncWriter
ix = index.create_in("indexdir", schema) # or index.open_dir("indexdir")
def upsert(doc: dict):
w = AsyncWriter(ix)
w.update_document(**doc) # insert-or-replace on the unique `id`
w.commit()
def delete(doc_id: str):
w = AsyncWriter(ix)
w.delete_by_term("id", doc_id)
w.commit()
def search(q: str, k: int = 5):
with ix.searcher() as s:
parser = MultifieldParser(["title", "body"], ix.schema, group=OrGroup)
hits = s.search(parser.parse(q), limit=k)
return [(h["id"], h["title"], round(h.score, 3)) for h in hits]
update_document 就是完整的同步方案：每当真实数据库中的某行发生变化，就用相同的 id 调用 upsert，索引就会跟上。看看这对排序的影响：
upsert({"id":"1","title":"Getting started with widgets",
"body":"install and configure your first widget","views":10})
upsert({"id":"2","title":"Widget troubleshooting",
"body":"fix common widget errors and crashes","views":99})
search("widget")
# [('2', 'Widget troubleshooting', 1.435), ('1', 'Getting started with widgets', 1.397)]
upsert({"id":"1","title":"Getting started with gadgets",
"body":"install your first gadget","views":10}) # doc 1 edited in place
search("widget")
# [('2', 'Widget troubleshooting', 2.386)] <- doc 1 dropped out, score adjusted
search("gadget")
# [('1', 'Getting started with gadgets', 3.432)]
MultifieldParser 同时搜索标题和正文；OrGroup 让多词查询能匹配到任意词项（更倾向召回），而 BM25 仍然会把最接近的匹配排在最前面。
一个陷阱：同一时间只能有一个 writer
这是朴素集成在高负载下崩溃的地方。Whoosh 索引同一时间只允许一个 writer——在另一个 writer 未提交时打开第二个 ix.writer() 会得到 LockError。在并发请求的 Web 应用中，这种情况一定会发生。两种稳健模式：
- `AsyncWriter`（上面用到的）：如果索引被锁，它会在后台线程中重试而不是抛异常。适合中低写入频率——评论、编辑、偶尔的管理员变更。
- 单一 writer / 写入队列： 把全部写入汇集到一个 worker（后台任务、Celery worker、专用线程）并批量提交。写入频繁时最佳，因为每次
commit()都有固定开销，批量提交可以摊薄它。
读取没有这种限制：每个请求打开一个新的 ix.searcher()（它们很轻量，能看到最后一次提交的状态），绝不要在请求之间持有同一个 searcher。commit() 返回后，新 searcher 立即可见变更——接近实时，无需调整刷新间隔。
接入 FastAPI
端点只是那三个函数的薄封装：
from fastapi import FastAPI
app = FastAPI()
@app.put("/docs/{doc_id}")
def put(doc_id: str, doc: dict):
upsert({"id": doc_id, **doc}); return {"ok": True}
@app.delete("/docs/{doc_id}")
def remove(doc_id: str):
delete(doc_id); return {"ok": True}
@app.get("/search")
def query(q: str, k: int = 10):
return [{"id": i, "title": t, "score": s} for i, t, s in search(q, k)]
这就是一个生产形态的搜索 API——排序、词干提取、增量更新——除了应用本身无需部署任何服务，搜索索引通过复制一个文件夹就能备份。FastAPI、Flask 和 Django 的完整可运行版本在仓库的 `examples/` 目录中。
什么时候仍需要 Elasticsearch？当你真的超出了单节点、进程内索引的规模——超大体量语料、分布式分片、集群分析。在此之前，嵌入式引擎要运行的东西更少、更容易出问题的地方也更少，而且完全够用。
原文：https://dev.to/priyasundaram/add-real-search-to-your-python-web-app-no-elasticsearch-no-separate-service-8la（作者 @priyasundaram）
