FastAPI 與外部服務整合 ── ElasticSearch / Meilisearch 搜尋

---

簡介

在現代 Web 應用中，即時且精準的搜尋功能往往是提升使用者體驗的關鍵。傳統的資料庫查詢雖能滿足基本需求，但在面對大量文字資料、模糊匹配、分詞與相關度排序時，效能與功能會迅速受限。此時，專門的搜尋引擎（如 ElasticSearch 或 Meilisearch）就能發揮威力。

FastAPI 以其 高效能、型別安全 的特性，成為構建 API 的首選框架。將 FastAPI 與 ElasticSearch / Meilisearch 結合，不僅可以在 Python 程式碼中直接呼叫強大的搜尋能力，還能保持簡潔的路由與自動生成的 OpenAPI 文件。本文將一步步帶你了解概念、實作範例，以及在實務上應注意的細節，適合剛接觸 FastAPI 的新手與想提升搜尋功能的中級開發者。

---

核心概念

1. 為什麼選擇 ElasticSearch 或 Meilisearch

項目	ElasticSearch	Meilisearch
成熟度	10+ 年的企業級部署經驗，支援分散式叢集	新興輕量級搜尋，適合中小型專案
安裝與部署	需要 JVM、較多資源	單一二進位檔，部署簡單
功能	完整的分析管線、聚合、腳本	預設即支援即時搜索、拼寫校正、同義詞
效能	大規模資料時表現優秀	小至中等規模資料快速回應
語言支援	官方 Python 客戶端 elasticsearch-py	官方 Python 客戶端 meilisearch-python

選擇依據：如果你的系統需要高可用、複雜聚合或大量日誌資料，建議使用 ElasticSearch；若是快速開發、需求較簡單且想降低維運成本，Meilisearch 是不錯的選擇。

2. 基本的連線與客戶端設定

ElasticSearch

# elastic_client.py
from elasticsearch import Elasticsearch

def get_es_client() -> Elasticsearch:
    """
    建立並回傳 Elasticsearch 客戶端。
    這裡使用環境變數或設定檔管理連線資訊，避免硬編碼。
    """
    es = Elasticsearch(
        hosts=["http://localhost:9200"],
        http_auth=("elastic", "your_password"),   # 若有啟用基本認證
        timeout=30,
    )
    return es

Meilisearch

# meili_client.py
from meilisearch import Client

def get_meili_client() -> Client:
    """
    建立並回傳 Meilisearch 客戶端。
    Meilisearch 預設不需要認證，若有設定 API Key，請在此加入。
    """
    client = Client("http://127.0.0.1:7700", api_key="masterKey")
    return client

小技巧：將客戶端建立封裝成函式或 Singleton，讓 FastAPI 的依賴注入（Dependency Injection）可以共用同一個連線，減少資源浪費。

3. 建立索引（Index）與資料映射（Mapping）

ElasticSearch 範例

# create_index.py
from elastic_client import get_es_client

def create_product_index():
    es = get_es_client()
    mapping = {
        "mappings": {
            "properties": {
                "title": {"type": "text"},
                "description": {"type": "text"},
                "price": {"type": "float"},
                "tags": {"type": "keyword"},
                "created_at": {"type": "date"},
            }
        }
    }
    # 若索引已存在會拋出例外，這裡簡化處理
    es.indices.create(index="products", body=mapping, ignore=400)

Meilisearch 範例

# create_index_meili.py
from meili_client import get_meili_client

def create_product_index():
    client = get_meili_client()
    # Meilisearch 會自動根據第一筆資料推斷欄位類型
    # 這裡僅示範建立索引名稱
    client.index("products")  # 若不存在會自動建立

注意：ElasticSearch 需要明確定義 Mapping，才能利用分詞器（analyzer）做中文斷詞；Meilisearch 則在後端自動處理，對中文支援較新，需要自行設定語言模型。

4. 資料寫入（Indexing）

範例 1：單筆資料寫入（ElasticSearch）

# index_one.py
from elastic_client import get_es_client

def index_product(product: dict):
    es = get_es_client()
    # 使用自動產生的 ID，或自行指定 _id
    es.index(index="products", document=product)

範例 2：批次寫入（Bulk）

# bulk_index.py
from elastic_client import get_es_client
from elasticsearch import helpers

def bulk_index_products(products: list[dict]):
    es = get_es_client()
    actions = [
        {
            "_index": "products",
            "_source": prod,
        }
        for prod in products
    ]
    helpers.bulk(es, actions)

範例 3：Meilisearch 單筆寫入

# index_one_meili.py
from meili_client import get_meili_client

def index_product(product: dict):
    client = get_meili_client()
    index = client.index("products")
    # Meilisearch 會回傳任務 ID，可用於監控進度（非同步）
    task = index.add_documents([product])
    return task

範例 4：Meilisearch 批次寫入

# bulk_index_meili.py
from meili_client import get_meili_client

def bulk_index_products(products: list[dict]):
    client = get_meili_client()
    index = client.index("products")
    task = index.add_documents(products)
    return task

5. 在 FastAPI 中實作搜尋 API

5.1 基本搜尋（ElasticSearch）

# main.py
from fastapi import FastAPI, Depends, Query
from elastic_client import get_es_client
from typing import List

app = FastAPI(title="商品搜尋 API")

def es_client():
    return get_es_client()

@app.get("/search", summary="ElasticSearch 商品搜尋")
def search(
    q: str = Query(..., description="關鍵字"),
    page: int = Query(1, ge=1, description="第幾頁"),
    size: int = Query(10, ge=1, le=100, description="每頁筆數"),
    client: Elasticsearch = Depends(es_client),
):
    body = {
        "from": (page - 1) * size,
        "size": size,
        "query": {
            "multi_match": {
                "query": q,
                "fields": ["title^3", "description", "tags"]
            }
        }
    }
    resp = client.search(index="products", body=body)
    hits = [hit["_source"] for hit in resp["hits"]["hits"]]
    total = resp["hits"]["total"]["value"]
    return {"total": total, "page": page, "size": size, "items": hits}

5.2 基本搜尋（Meilisearch）

# main_meili.py
from fastapi import FastAPI, Depends, Query
from meili_client import get_meili_client
from typing import List

app = FastAPI(title="商品搜尋 API (Meilisearch)")

def meili_client():
    return get_meili_client()

@app.get("/search", summary="Meilisearch 商品搜尋")
def search(
    q: str = Query(..., description="關鍵字"),
    page: int = Query(1, ge=1),
    size: int = Query(10, ge=1, le=100),
    client = Depends(meili_client),
):
    index = client.index("products")
    resp = index.search(
        q,
        {
            "offset": (page - 1) * size,
            "limit": size,
            "attributesToHighlight": ["title", "description"]
        }
    )
    return {
        "total": resp["nbHits"],
        "page": page,
        "size": size,
        "items": resp["hits"]
    }

5.3 高階功能：拼寫校正（Meilisearch）

@app.get("/suggest", summary="搜尋建議（拼寫校正）")
def suggest(
    q: str = Query(..., description="使用者輸入"),
    client = Depends(meili_client),
):
    index = client.index("products")
    # Meilisearch 內建 typo tolerance，直接使用 search 即可
    resp = index.search(q, {"limit": 5})
    return {"suggestions": [hit["title"] for hit in resp["hits"]]}

5.4 高階功能：聚合分析（ElasticSearch）

@app.get("/stats/price", summary="價格分布統計")
def price_stats(
    client: Elasticsearch = Depends(es_client),
):
    body = {
        "size": 0,  # 不返回文件
        "aggs": {
            "price_ranges": {
                "range": {
                    "field": "price",
                    "ranges": [
                        {"to": 100},
                        {"from": 100, "to": 500},
                        {"from": 500}
                    ]
                }
            }
        }
    }
    resp = client.search(index="products", body=body)
    buckets = resp["aggregations"]["price_ranges"]["buckets"]
    return {"price_distribution": buckets}

---

常見陷阱與最佳實踐

陷阱	說明	最佳做法
連線資源未釋放	每次請求都重新建立 ElasticSearch 連線會導致大量 socket 開銷。	使用 依賴注入 並在程式啟動階段建立單例，或使用 async 客戶端（elasticsearch-async）
中文斷詞不正確	ElasticSearch 預設使用英文字分析器，中文會被視為單一 token。	安裝 analysis-ik 或 analysis-nori 等中文分詞插件，並在 Mapping 中指定 analyzer: "ik_max_word"
搜尋結果過大	直接返回全部欄位會浪費頻寬，尤其在大文件時。	使用 _source 或 attributesToRetrieve 控制返回欄位；在 Meilisearch 使用 attributesToRetrieve
安全性漏洞	將 Elasticsearch/Meilisearch 的 API 直接暴露在公開網路上。	設置防火牆、IP 白名單，或在 FastAPI 前加上 JWT/OAuth2 驗證層
同步阻塞	es.search() 為同步呼叫，會阻塞 event loop。	轉換為 async 版（elasticsearch[async]）或使用 ThreadPoolExecutor 包裝
索引映射不一致	變更 Mapping 後舊資料仍使用舊結構，導致查詢錯誤。	使用 Reindex API 重新建立索引或建立新索引後切換別名（alias）

---

實際應用場景

1. 電子商務平台

   • 商品名稱、描述、標籤需要即時搜尋與排序。
   • 使用 ElasticSearch 的 function_score 調整相關度，使新品或促銷商品排名提升。

2. 部落格或新聞站

   • 大量文字內容需要支援 全文檢索、同義詞、關鍵字高亮。
   • Meilisearch 可快速部署，且內建 highlight 功能，只要在 FastAPI 回傳結果時直接使用即可。

3. 客服系統

   • 透過搜尋歷史工單、FAQ，提供 即時建議（autocomplete）與 拼寫容錯。
   • Meilisearch 的 typo tolerance 與 search-as-you-type 設定非常合適。

4. 資料分析儀表板

   • 需要 聚合統計（如價格分布、庫存量）給前端圖表。
   • ElasticSearch 的聚合管線（Aggregations）提供彈性且效能佳的計算方式。

---

總結

FastAPI 與 ElasticSearch / Meilisearch 的結合，讓開發者能以 簡潔的程式碼 交付 高效、可擴充 的搜尋服務。

• ElasticSearch：適合大型、需要複雜聚合與自訂分析的場景。
• Meilisearch：輕量、上手快，適合中小型專案或快速原型。

在實作時，記得把 連線管理、中文斷詞、資料安全 放在首位，並善用 FastAPI 的依賴注入 讓客戶端資源共享。透過本文提供的範例，你可以快速建立商品搜尋 API，並根據需求擴充拼寫校正、聚合分析或即時建議等功能。祝你在專案中玩得開心，搜尋功能大放異彩！