撰寫新的查詢執行器

快速導覽

簡介

Redash 已經連接到許多資料庫和 REST API。若要在 Redash 中新增對新資料來源類型的支援，您需要為它實作一個查詢執行器 (Query Runner)。查詢執行器是一個 Python 類別。這個文件頁面展示了撰寫新查詢執行器的過程。它以 Firebolt 查詢執行器為例。

首先在 /redash/query_runner 目錄中建立一個新的 firebolt.py 檔案，並實作 BaseQueryRunner 類別

from redash.query_runner import BaseQueryRunner, register

class Firebolt(BaseQueryRunner):
    def run_query(self, query, user):
        pass

您必須實作的唯一方法是 run_query 方法，它接受一個 query 參數（字串）和調用此查詢的 user。使用者對於大多數查詢執行器來說是無關緊要的，可以忽略。

設定

通常，查詢執行器需要一些設定才能使用，因此我們需要實作 configuration_schema 類別方法。這些欄位屬於 properties 鍵之下

@classmethod
def configuration_schema(cls):
    return {
        "type": "object",
        "properties": {
            "api_endpoint": {"type": "string", "default": DEFAULT_API_URL},
            "engine_name": {"type": "string"},
            "DB": {"type": "string"},
            "user": {"type": "string"},
            "password": {"type": "string"}
        },
        "order": ["user", "password", "api_endpoint", "engine_name", "DB"],
        "required": ["user", "password", "engine_name", "DB"],
        "secret": ["password"],
    }

此方法會傳回一個 JSON schema 物件。

每個屬性都必須指定一個 type。屬性支援的類型有 string、number 和 boolean。對於類似檔案的欄位，請參閱下一個標題。

您也可以選擇性地指定一個 default 值和 title，它們將顯示在 UI 中。如果您沒有指定 title，則會使用屬性名稱。沒有預設值的屬性將會是空白的。

另請注意 required 欄位，它定義了必要的屬性（除了本例中的 api_endpoint 之外的所有屬性）和 secret，它定義了機密欄位（不會送回 UI）。

這些設定的值可以作為字典在查詢執行器物件的 self.configuration 欄位中存取。

檔案上傳

當使用者建立資料來源的實例時，Redash 會將設定儲存在其 metadata 資料庫中。某些資料來源會要求使用者上傳檔案（例如 SSL 憑證或金鑰檔案）。若要處理此問題，請定義名稱以 File 結尾且類型為 string 的屬性。例如

    "properties": {
        "someFile": {"type": "string"},
    }

Redash 前端會將任何名稱以 File 結尾且類型為 string 的屬性，呈現為檔案上傳選擇器元件。儲存時，檔案的內容將會被加密並以位元組的形式儲存到 metadata 資料庫中。在您的查詢執行器程式碼中，您可以將 self.configuration['someFile'] 的值讀入 Python 內建 tempfile 程式庫中的其中一個固定裝置。從那裡，您可以像處理儲存在磁碟上的任何檔案一樣處理這些位元組。您可以在 PostgreSQL 查詢執行器程式碼中看到一個範例。

執行查詢

現在我們已經定義了設定，我們可以實作 run_query 方法

def run_query(self, query, user):
    connection = connect(
        api_endpoint=(self.configuration.get("api_endpoint") or DEFAULT_API_URL),
        engine_name=(self.configuration.get("engine_name") or None),
        username=(self.configuration.get("user") or None),
        password=(self.configuration.get("password") or None),
        database=(self.configuration.get("DB") or None),
    )

    cursor = connection.cursor()

    try:
        cursor.execute(query)
        columns = self.fetch_columns(
            [(i[0], TYPES_MAP.get(i[1], None)) for i in cursor.description]
        )
        rows = [
            dict(zip((column["name"] for column in columns), row)) for row in cursor
        ]

        data = {"columns": columns, "rows": rows}
        error = None
        json_data = json_dumps(data)
    finally:
        connection.close()

    return json_data, error

這是最少的必要程式碼。以下是它的作用

連接到已設定的 Firebolt 端點，或使用從官方 Firebolt Python API 用戶端匯入的 DEFAULT_API_URL。
執行查詢。
將結果轉換為 Redash [期望]({% link _kb/data-sources/querying/json-api %}#Required-Data-Structure) 的格式。

將資料行類型對應至 Redash 類型

請注意這些行

columns = self.fetch_columns(
    [(i[0], TYPES_MAP.get(i[1], None)) for i in cursor.description]
)

BaseQueryRunner 包含一個輔助函式 (fetch_columns)，它可以消除重複的資料行名稱，並為資料行指派類型（如果已知）。如果沒有指派類型，則預設為字串。TYPES_MAP 字典是我們在檔案頂端定義的自訂字典。它在不同的查詢執行器之間會有所不同。

run_query 方法的傳回值是 JSON 編碼的結果和錯誤字串的元組。如果您想要傳回某種自訂錯誤訊息，則會使用錯誤字串，否則您可以讓例外情況傳播（這在您首次開發查詢執行器時很有用）。

擷取資料庫綱要

到目前為止，我們已經展示了執行查詢的最少必要項目。如果您還希望 Redash 顯示資料庫綱要並啟用自動完成功能，您需要實作 get_schema 方法

def get_schema(self, get_stats=False):
    query = """
    SELECT TABLE_SCHEMA,
            TABLE_NAME,
            COLUMN_NAME
    FROM INFORMATION_SCHEMA.COLUMNS
    WHERE TABLE_SCHEMA <> 'INFORMATION_SCHEMA'
    """

    results, error = self.run_query(query, None)

    if error is not None:
        raise Exception("Failed getting schema.")

    schema = {}
    results = json_loads(results)

    for row in results["rows"]:
        table_name = "{}.{}".format(row["table_schema"], row["table_name"])

        if table_name not in schema:
            schema[table_name] = {"name": table_name, "columns": []}

        schema[table_name]["columns"].append(row["column_name"])

    return list(schema.values())

get_schema 的實作方式會因您新增支援的資料來源而異，但傳回值必須是一個字典陣列，其中每個字典都有一個 name 鍵（資料表名稱）和 columns 鍵（作為字串的資料行名稱陣列）。

在綱要瀏覽器中包含資料行類型

如果您希望 Redash 綱要瀏覽器也顯示資料行類型，您可以調整 get_schema 方法，以便 columns 鍵包含一個具有 name 和 type 鍵的字典陣列。

以下是一個不包含資料行類型的範例

[
  {
    "name": "Table1",
    "columns": ["field1", "field2", "field3"]
  }
]

以下是一個包含資料行類型的範例

[
  {
    "name": "Table1",
    "columns": [
      {
        "name": "field1",
        "type": "VARCHAR"
      },
      {
        "name": "field2",
        "type": "BIGINT"
      },
      {
        "name": "field3",
        "type": "DATE"
      }
    ]
  }
]

請注意，資料行類型字串僅用於協助查詢作者。如果它出現在 get_schema 的輸出中，Redash 會信任它，不會將它與 run_query 傳回的類型資訊進行比較。因此，綱要瀏覽器中顯示的類型可能與資料庫中的資料行類型不同。我們建議針對已知綱要進行手動測試，以確保正確的類型出現在綱要瀏覽器中。

新增測試連線支援

您也可以實作「測試連線」按鈕支援。「測試連線」按鈕會出現在資料來源設定和配置畫面中。您可以為您的查詢執行器提供一個 noop_query 屬性，或自行實作 test_connection 方法。在這個範例中，我們選擇了第一個選項

class Firebolt(BaseQueryRunner):
    noop_query = "SELECT 1"

支援 SQL 資料庫的自動限制

Redash 前端包含一個核取方塊，可自動限制查詢結果。這有助於避免大型結果集使 Redash Web 應用程式過載。對於大多數 SQL 樣式的資料庫，您可以透過繼承 BaseSQLQueryRunner 而不是 BaseQueryRunner 來自動新增自動限制支援。

from redash.query_runner import BaseSQLQueryRunner, register

class Firebolt(BaseSQLQueryRunner):
    def run_query(self, query, user):
        pass

只要選取查詢編輯器中的核取方塊，BaseSQLQueryRunner 就會使用 sqplarse 在執行之前，將 LIMIT 1000 智慧地附加到查詢中。對於使用不同語法的資料庫（特別是 Microsoft SQL Server 或任何 NoSQL 資料庫），您可以繼續繼承 BaseQueryRunner 並實作以下項目

@property
def supports_auto_limit(self):
    return True

def apply_auto_limit(self, query_text: str, should_apply_auto_limit: bool):
    ...

對於 BaseQueryRunner，supports_auto_limit 屬性預設為 false，且 apply_auto_limit 會傳回未修改的查詢文字。

檢查必要的相依性

如果查詢執行器需要某些外部 Python 套件，我們會使用 try/except 區塊包裝這些匯入，以防止在沒有此套件的情況下造成部署崩潰

try:
    from firebolt.db import connect
    from firebolt.client import DEFAULT_API_URL
    enabled = True
except ImportError:
    enabled = False

稍後會在查詢執行器的 enabled 類別方法中使用 enabled 變數

@classmethod
def enabled(cls):
    return enabled

如果它傳回 False，則不會啟用查詢執行器。

完成

在檔案的頂部，匯入 register 函數，並在 firebolt.py 的底部呼叫它。

# top of file

try:
    from firebolt.db import connect
    from firebolt.client import DEFAULT_API_URL
    enabled = True
except ImportError:
    enabled = False

from redash.query_runner import BaseQueryRunner, register
from redash.query_runner import TYPE_STRING, TYPE_INTEGER, TYPE_BOOLEAN
from redash.utils import json_dumps, json_loads

TYPES_MAP = {1: TYPE_STRING, 2: TYPE_INTEGER, 3: TYPE_BOOLEAN}

# ... implementation

# bottom of file
register(Firebolt)

通常，連接器會需要一些額外的 Python 套件，我們會將它們添加到 requirements_all_ds.txt 檔案中。如果所需的 Python 套件沒有任何特殊依賴關係（例如某些系統套件），我們通常會將查詢執行器添加到 redash/settings/__init__.py 中的 default_query_runners 中。

您可以在這裡查看 Firebolt 查詢執行器的完整 Pull Request。

摘要

Redash 查詢執行器是一個 Python 類別，至少會實作一個 run_query 方法，該方法會以 Redash 期望的格式傳回結果。可配置的資料來源設定由 configuration_schema 類別方法定義，該方法會傳回 JSON 綱要。您可以選擇性地實作連線測試、綱要擷取和自動限制。您可以透過將其添加到設定中的 default_query_runners 清單中，或設定 ADDITIONAL_QUERY_RUNNERS 環境變數來啟用資料來源。

在 GitHub 上編輯