腳本開發 / InfluxDB

InfluxDB 連接器操作對象為 Python 第三方包 influxdb（版本 5.2.3）的封裝，主要提供一些用於查詢 InfluxDB 的方法。 本連接器兼容以下數據庫：

• 阿里雲時序數據庫 InfluxDB 版

本連接器基於 HTTP 協議連接

DFF.CONN(...) 參數如下：

參數	類型	必須 / 默認值	説明
connector_id	str	必須	連接器 ID
database	str	None	指定數據庫

.query(...)

執行 InfluxQL 語句，參數如下：

參數	類型	必須 / 默認值	説明
sql	str	必須	InfluxQL 語句，可包含綁定參數佔位符，形式為 $var_name
bind_params	dict	None	綁定參數
database	str	None	本次查詢指定數據庫
dict_output	dict	False	返回數據自動轉換為{列名：值}形式

sql = 'SELECT * FROM demo WHERE city = $city LIMIT 5'
bind_params = {'city': 'hangzhou'}
result = db.query(sql, bind_params=bind_params, database='demo')

# {'series': [{'columns': ['time', 'city', 'hostname', 'status', 'value'], 'name': 'demo', 'values': [['2018-12-31T16:00:10Z', 'hangzhou', 'webserver', 'UNKNOW', 90], ['2018-12-31T16:00:20Z', 'hangzhou', 'jira', 'running', 40], ['2018-12-31T16:00:50Z', 'hangzhou', 'database', 'running', 50], ['2018-12-31T16:01:00Z', 'hangzhou', 'jira', 'stopped', 40], ['2018-12-31T16:02:00Z', 'hangzhou', 'rancher', 'UNKNOW', 90]]}]}

sql = 'SELECT * FROM demo WHERE city = $city LIMIT 5'
bind_params = {'city': 'hangzhou'}
result = db.query(sql, bind_params=bind_params, database='demo', dict_output=True)

# {'series': [[{'city': 'hangzhou', 'hostname': 'webserver', 'status': 'UNKNOW', 'time': '2018-12-31T16:00:10Z', 'value': 90}, {'city': 'hangzhou', 'hostname': 'jira', 'status': 'running', 'time': '2018-12-31T16:00:20Z', 'value': 40}, {'city': 'hangzhou', 'hostname': 'database', 'status': 'running', 'time': '2018-12-31T16:00:50Z', 'value': 50}, {'city': 'hangzhou', 'hostname': 'jira', 'status': 'stopped', 'time': '2018-12-31T16:01:00Z', 'value': 40}, {'city': 'hangzhou', 'hostname': 'rancher', 'status': 'UNKNOW', 'time': '2018-12-31T16:02:00Z', 'value': 90}]]}

.query2(...)

query2(...) 方法同樣用於執行 InfluxQL 語句，但參數佔位符不同，使用問號 ? 作為參數佔位符。參數如下：

參數	類型	必須 / 默認值	説明
sql	str	必須	InfluxQL 語句，可包含參數佔位符。 ? 表示需要轉義的參數； ?? 表示不需要轉義的參數
sql_params	list	None	InfluxQL 參數
database	str	None	本次查詢指定數據庫
dict_output	dict	False	返回數據自動轉換為 {"列名": "值"} 形式

sql = 'SELECT * FROM ?? WHERE city = ? LIMIT 5'
sql_params = ['demo', 'hangzhou']
result = db.query2(sql, sql_params=sql_params, dict_output=True)

動態 SQL 語句

query2(...) 內部使用 DFF.SQL(...) 構造 SQL 語句並支持構造複雜動態的 SQL 語句。

如 WHERE IN (...) 不確定數量的值，或者 INSERT INTO ... VALUES ... 進行批量寫入數據等。

詳情請參考 腳本開發 / SQL 構造 DFF.SQL

.write_point(...)

於 1.1.13 版本新增

寫入單個數據點。參數如下：

參數	類型	必須 / 默認值	説明
measurement	str	必須	指標集
fields	dict{str: str/int/float/bool}	必須	字段 鍵名必須為 str 鍵值可以為 str/int/float/bool
tags	dict{str: str}	None	標籤 鍵名、鍵值必須為都為 str
timestamp	str/int	None	時間 ISO 格式，如：2020-01-01T01:02:03Z UNIX 時間戳如：1577840523
database	str	None	本次寫入指定數據庫

fields = { 'cpu': 100, 'mem': 0.5 }
tags   = { 'host': 'web001' }
result = db.write_point(measurement='host_monitor', fields=fields, tags=tags)

.write_points(...)

於 1.1.13 版本新增

批量寫入數據點。參數如下：

參數	類型	必須 / 默認值	説明
points	list	必須	數據點數組
points[#]['measurement']	str	必須	指標集
points[#]['fields']	dict{str: str/int/float/bool}	必須	字段 鍵名必須為 str 鍵值可以為 str/int/float/bool
points[#]['tags']	dict{str: str}	None	標籤 鍵名、鍵值必須為都為 str
points[#]['time']	str/int	None	時間 ISO 格式，如：2020-01-01T01:02:03Z UNIX 時間戳如：1577840523
database	str	None	本次寫入指定數據庫

points = [
    {
        'measurement': 'host_monitor',
        'fields'     : { 'cpu': 100, 'mem': 0.5 },
        'tags'       : { 'host': 'web001' },
    }
]
result = db.write_points(points)