nayota-tdenginedb-sdk

概述

这是一个基于 Node.js 的 TDengine 数据库 ORM 库，模仿 Mongoose 的 API 设计，提供了完整的数据库操作功能。主要包含四个核心模块：TDengineConnection、Schema、Model 和 WindowQuery。

特性

• 🚀 简单易用: 类似 Mongoose 的 API 设计，学习成本低
• 🔧 功能完整: 支持超级表、普通表、窗口查询等 TDengine 核心功能
• 📊 窗口查询: 支持时间窗口、状态窗口、会话窗口等多种查询类型
• 🔍 智能验证: 自动关键字验证和数据类型转换
• 📝 文档管理: 完整的文档实例管理和变更跟踪
• 🔄 自动重连: 内置连接管理和自动重连机制

目录结构

server/utils/td/
├── TDengineConnection.js    # 数据库连接管理
├── Schema.js               # 数据表结构定义
├── Model.js                # 数据模型和文档操作
├── WindowQuery.js          # 窗口查询功能
├── index.js                # 主入口文件
└── README.md               # 使用说明文档

核心模块详解

1. TDengineConnection - 数据库连接管理

功能特性

• 基于 WebSocket 的 TDengine 连接
• 自动重连机制
• 数据库和超级表管理
• 流计算管理器集成
• 模型表自动创建

基本用法

const { TDengineConnection } = require('nayota-tdenginedb-sdk')

// 创建连接实例
const connection = new TDengineConnection()

// 连接到数据库
await connection.connect('127.0.0.1', 'root', 'taosdata', 6041, 'testdb')

// 创建数据库
await connection.createDatabase('mydb')

// 切换数据库
await connection.useDatabase('mydb')

// 执行SQL
await connection.execute('CREATE TABLE test (ts timestamp, val int)')

// 查询数据
const result = await connection.query('SELECT * FROM test LIMIT 10')

// 关闭连接
await connection.close()

主要方法

| 方法 | 参数 | 说明 | |------|------|------| | connect(ip, user, pwd, port, db) | 连接参数 | 连接到TDengine数据库 | | reconnect() | 无 | 重新连接数据库 | | createDatabase(dbName) | 数据库名 | 创建数据库 | | useDatabase(dbName) | 数据库名 | 切换数据库 | | execute(sql) | SQL语句 | 执行SQL语句 | | query(sql) | SQL语句 | 查询数据 | | model(name, schema) | 模型名和结构 | 创建数据模型 | | close() | 无 | 关闭连接 |

2. Schema - 数据表结构定义

功能特性

• 支持 TDengine 数据类型映射
• 自动关键字验证
• 超级表和普通表支持
• 标签列和数据列区分
• 字段别名支持

基本用法

const { Schema } = require('nayota-tdenginedb-sdk')

// 定义普通表结构
const userSchema = new Schema({
  // 数据列定义
  name: { type: String, required: true },
  age: { type: Number },
  email: { type: 'nchar(100)' },
  recordTime: { type: Date }  // 时间戳字段
}, {
  timestamps: true,  // 自动添加时间戳字段
  strict: false      // 是否严格模式
})

// 定义超级表结构 - 方式1：在字段定义中指定标签
const alarmSchema = new Schema({
  recordTime: { type: Date },           // 数据列
  val: { type: Number },                // 数据列
  valueJSON: { type: String },          // 数据列
  valueStr: { type: String },           // 数据列
  isSimulation: { type: Boolean, default: false }, // 数据列
  alarm: {                              // 标签列
    type: String,
    isTag: true,
    isTbName: true  // 用于生成子表名
  }
}, {
  tbname: 'alarm'  // 子表前缀
})

#### 字段配置选项

| 选项 | 类型 | 说明 |
|------|------|------|
| `type` | String/Function | 字段类型，支持JavaScript类型或TDengine类型字符串 |
| `isTag` | Boolean | 是否为标签列（超级表专用） |
| `isTbName` | Boolean | 是否用于生成子表名 |
| `alias` | String | 字段别名 |
| `required` | Boolean | 是否必填 |
| `default` | Any | 默认值 |

#### 子表名生成规则

对于超级表，子表名的生成有以下两种方式：

1. **指定 isTbName 字段**（推荐）：
   - 当某个标签字段设置了 `isTbName: true` 时，使用该字段的值生成子表名
   - 格式：`{tbname}_{isTbName字段值}`
   - 示例：`alarm_device001`

2. **兼容模式**（自动拼接所有标签）：
   - 当没有任何标签字段设置 `isTbName: true` 时，自动使用所有标签字段的值拼接生成子表名
   - 格式：`{tbname}_{tag1值}_{tag2值}_{tag3值}...`
   - 特殊字符会被替换为下划线，连续下划线会被合并
   - 示例：`device_thermometer_room101_buildinga`

```javascript
// 方式1：使用 isTbName 字段
const schema1 = new Schema({
  temperature: { type: 'float' },
  device_id: { type: 'nchar(32)', isTag: true, isTbName: true },
  location: { type: 'nchar(64)', isTag: true }
}, { tbname: 'sensor' })
// 生成子表名：sensor_device001

// 方式2：兼容模式（所有标签拼接）
const schema2 = new Schema({
  temperature: { type: 'float' },
  device_type: { type: 'nchar(32)', isTag: true },
  location: { type: 'nchar(64)', isTag: true },
  building: { type: 'nchar(32)', isTag: true }
}, { tbname: 'device' })
// 生成子表名：device_thermometer_room101_buildinga

数据类型映射

| JavaScript类型 | TDengine类型 | 说明 | |----------------|--------------|------| | Date | timestamp | 时间戳 | | Number | double | 浮点数 | | String | nchar(256) | 字符串 | | Boolean | bool | 布尔值 | | 字符串 | 原样使用 | 如 'int', 'varchar(50)', 'binary(32)' |

实际使用示例

// 报警记录Schema
const AlarmRecordSchema = new Schema({
  recordTime: { type: Date },                    // 时间戳字段
  val: { type: Number },                         // 数值字段
  valueJSON: { type: String },                   // JSON字符串
  valueStr: { type: String },                    // 字符串值
  isSimulation: { type: Boolean, default: false }, // 布尔值带默认值
  alarm: {                                       // 标签列
    type: String,
    isTag: true,
    isTbName: true  // 用于生成子表名
  }
}, {
  tbname: 'alarm'  // 子表前缀
})

// 检查记录Schema
const CheckRecordSchema = new Schema({
  recordTime: { type: 'timestamp' },             // 直接使用TDengine类型
  val: {
    type: 'float',
    alias: 'value'  // 字段别名
  },
  valueJSON: { type: 'nchar(256)' },             // 指定长度的字符串
  valueStr: { type: 'nchar(256)' },
  check: {                                        // 标签列
    type: 'binary(32)',
    isTag: true,
    isTbName: true
  },
  isSimulation: { type: Boolean, default: false }
}, {
  tbname: 'check'
})

// 召回记录Schema
const RecallSchema = new Schema({
  recordTime: { type: 'timestamp' },
  modelBody: {                                    // 标签列
    type: 'binary(32)',
    isTag: true
  },
  onModel: {                                      // 既是标签又是子表名
    type: 'binary(32)',
    isTbName: true,
    isTag: true
  },
  code: { type: 'binary(32)' }                   // 普通数据列
}, {
  tbname: 'recall'
})

主要方法

| 方法 | 说明 | |------|------| | getColumnsString() | 获取列定义字符串 | | getTagsString() | 获取标签定义字符串 | | validateKeywords() | 验证关键字使用 | | getSafeDefinition() | 获取安全的Schema定义 |

3. Model - 数据模型和文档操作

功能特性

• 类似 Mongoose 的 API 设计
• 支持超级表和普通表
• 自动子表创建
• 文档实例管理
• 数据验证和变更跟踪
• 丰富的查询方法

基本用法

const { Model, Schema } = require('nayota-tdenginedb-sdk')

// 定义Schema
const userSchema = new Schema({
  name: { type: String, required: true },
  age: { type: Number },
  email: { type: 'nchar(100)' }
}, {
  timestamps: true
})

// 创建模型
const User = Model('users', userSchema, connection)

// 创建表
await User.createTable()

// 插入数据
await User.insert({
  name: 'John Doe',
  age: 30,
  email: '[email protected]'
})

// 查询数据
const users = await User.find({ age: { $gte: 25 } })

// 更新数据
await User.update({ name: 'John Doe' }, { age: 31 })

// 删除数据
await User.delete({ name: 'John Doe' })

超级表操作

// 定义超级表Schema - 方式1：在字段中定义标签
const alarmSchema = new Schema({
  recordTime: { type: Date },
  val: { type: Number },
  valueJSON: { type: String },
  valueStr: { type: String },
  isSimulation: { type: Boolean, default: false },
  alarm: {                              // 标签列
    type: String,
    isTag: true,
    isTbName: true  // 用于生成子表名
  }
}, {
  tbname: 'alarm'
})

const AlarmRecord = Model('alarmrecords', alarmSchema, connection)

// 创建超级表
await AlarmRecord.createSuperTable()

// 向超级表插入数据（自动创建子表）
await AlarmRecord.insert({
  recordTime: new Date(),
  val: 25.5,
  valueJSON: '{"status": "normal"}',
  valueStr: 'normal',
  isSimulation: false,
  alarm: 'temperature_high'  // 这个值会用于生成子表名
})

// 定义超级表Schema - 方式2：使用 tags 选项
const deviceSchema = new Schema({
  temperature: { type: Number },
  humidity: { type: Number },
  recordTime: { type: Date }
}, {
  tags: [
    { name: 'deviceId', type: 'nchar(50)', isTag: true, isTbName: true },
    { name: 'location', type: 'nchar(100)', isTag: true }
  ],
  tbname: 'device_data'
})

const Device = Model('devices', deviceSchema, connection)

// 创建超级表
await Device.createSuperTable()

// 手动创建子表
await Device.createSubTable('device_001', {
  deviceId: '001',
  location: 'Beijing'
})

// 向超级表插入数据
await Device.insert({
  tbname: 'device_001',
  deviceId: '001',
  location: 'Beijing',
  temperature: 25.5,
  humidity: 60.2,
  recordTime: new Date()
})

文档实例操作

// 创建文档实例
const user = User.create({
  name: 'Jane Doe',
  age: 28,
  email: '[email protected]'
})

// 检查状态
console.log(user.isNew())        // true
console.log(user.isModified())   // false

// 修改数据
user.age = 29
user.set('email', '[email protected]')

// 检查修改
console.log(user.isModified())           // true
console.log(user.modifiedPaths())        // ['age', 'email']
console.log(user.getChanges('age'))      // { original: 28, current: 29 }

// 保存到数据库
await user.save()

// 查询并获取文档实例
const foundUser = await User.findOne({ name: 'Jane Doe' })
if (foundUser) {
  foundUser.age = 30
  await foundUser.save()
}

主要静态方法

| 方法 | 参数 | 说明 | |------|------|------| | createTable() | 无 | 创建普通表 | | createSuperTable() | 无 | 创建超级表 | | createSubTable(name, tags) | 子表名和标签 | 创建子表 | | insert(data) | 数据对象或数组 | 插入数据 | | find(query, options) | 查询条件和选项 | 查找数据 | | findOne(query, options) | 查询条件和选项 | 查找单个数据 | | update(query, update) | 查询条件和更新数据 | 更新数据 | | delete(query) | 查询条件 | 删除数据 | | count(query) | 查询条件 | 统计数量 |

主要实例方法

| 方法 | 参数 | 说明 | |------|------|------| | save() | 无 | 保存文档 | | remove() | 无 | 删除文档 | | set(path, value) | 字段路径和值 | 设置字段值 | | get(path) | 字段路径 | 获取字段值 | | isNew() | 无 | 是否为新文档 | | isModified(path) | 字段路径 | 是否被修改 | | validateSync() | 无 | 同步验证 |

4. WindowQuery - 窗口查询功能

功能特性

• 支持多种窗口类型
• 时间窗口、状态窗口、会话窗口等
• 灵活的查询构建器
• 自动填充选项
• 时间戳伪列支持

基本用法

const { WindowQuery } = require('nayota-tdenginedb-sdk')

// 创建窗口查询实例
const windowQuery = new WindowQuery(Device)

// 时间窗口查询
const timeWindowResults = await windowQuery
  .select(['AVG(temperature) as avg_temp', 'MAX(humidity) as max_humidity'])
  .where({ deviceId: '001' })
  .interval('10m')  // 10分钟窗口
  .fill('PREV')     // 前值填充
  .orderBy('ts')
  .exec()

// 按标签分组的时间窗口
const tagWindowResults = await windowQuery
  .select(['deviceId', 'AVG(temperature) as avg_temp'])
  .partitionBy('deviceId')
  .interval('1h')
  .addTimestampPseudoColumns(['_WSTART', '_WEND'])
  .exec()

窗口类型

1. 时间窗口 (INTERVAL)

// 基本时间窗口
windowQuery.interval('10m')  // 10分钟窗口

// 带滑动窗口
windowQuery.interval('10m', { sliding: '5m' })

// 带偏移量
windowQuery.interval('1h', { offset: '30m' })

// 带填充
windowQuery.interval('10m', { fill: 'PREV' })

2. 状态窗口 (STATE_WINDOW)

// 基本状态窗口
windowQuery.stateWindow('status')

// 带TRUE_FOR参数
windowQuery.stateWindow('status', '10m')

3. 会话窗口 (SESSION)

// 会话窗口
windowQuery.session('ts', '30m')  // 30分钟容忍时间

4. 事件窗口 (EVENT_WINDOW)

// 事件窗口
windowQuery.eventWindow('temperature > 30', 'temperature < 25')

5. 计数窗口 (COUNT_WINDOW)

// 基本计数窗口
windowQuery.countWindow(100)

// 带滑动
windowQuery.countWindow(100, 50)

// 带计数列
windowQuery.countWindow(100, 50, 'deviceId')

填充选项

// 预定义填充类型
windowQuery.fill('NULL')    // 填充NULL
windowQuery.fill('PREV')    // 前值填充
windowQuery.fill('NEXT')    // 后值填充
windowQuery.fill('LINEAR')  // 线性插值

// 自定义填充值
windowQuery.fill({ value: 0 })

// 不填充（TDengine 3.0+）
windowQuery.fill({ type: 'NONE' })

快捷方法

// 时间间隔统计
const results1 = await Device.timeInterval('10m', 'AVG', 'temperature')

// 按标签分组的时间统计
const results2 = await Device.timeIntervalByTag('deviceId', '1h', 'MAX', 'temperature')

// 按子表分组的时间统计
const results3 = await Device.timeIntervalBySubTable('30m', 'COUNT', '*')

主要方法

| 方法 | 参数 | 说明 | |------|------|------| | select(fields) | 字段数组或字符串 | 设置查询字段 | | where(condition) | 查询条件 | 添加WHERE条件 | | partitionBy(fields) | 分组字段 | 设置PARTITION BY | | interval(interval, options) | 时间间隔和选项 | 时间窗口 | | stateWindow(column, trueFor) | 状态列和持续时间 | 状态窗口 | | session(tsColumn, tolerance) | 时间戳列和容忍时间 | 会话窗口 | | eventWindow(start, end, trueFor) | 开始和结束条件 | 事件窗口 | | countWindow(count, sliding, columns) | 计数参数 | 计数窗口 | | fill(fillType) | 填充类型 | 设置填充方式 | | orderBy(sort) | 排序规则 | 设置排序 | | limit(limit) | 限制数量 | 设置LIMIT | | offset(offset) | 偏移量 | 设置OFFSET | | exec() | 无 | 执行查询 |

完整使用示例

1. 基本设置

const { TDengineConnection, Schema, model } = require('nayota-tdenginedb-sdk')

// 创建连接
const connection = new TDengineConnection()
await connection.connect('127.0.0.1', 'root', 'taosdata', 6041, 'iotdb')

// 定义传感器数据Schema
const sensorSchema = new Schema({
  recordTime: { type: Date },           // 时间戳字段
  temperature: { type: Number },        // 温度
  humidity: { type: Number },           // 湿度
  pressure: { type: Number },           // 压力
  voltage: { type: Number },            // 电压
  valueJSON: { type: String },          // JSON数据
  valueStr: { type: String },           // 字符串值
  isSimulation: { type: Boolean, default: false }, // 是否模拟数据
  deviceId: {                           // 设备ID标签
    type: 'nchar(50)',
    isTag: true,
    isTbName: true
  },
  location: {                           // 位置标签
    type: 'nchar(100)',
    isTag: true
  },
  type: {                               // 类型标签
    type: 'nchar(50)',
    isTag: true
  }
}, {
  tbname: 'sensor_data'  // 子表前缀
})

// 创建模型
const Sensor = model('sensors', sensorSchema, connection)

2. 数据操作

// 创建超级表
await Sensor.createSuperTable()

// 插入传感器数据
await Sensor.insert([
  {
    recordTime: new Date(),
    temperature: 25.5,
    humidity: 60.2,
    pressure: 1013.25,
    voltage: 3.3,
    valueJSON: '{"status": "normal"}',
    valueStr: 'normal',
    isSimulation: false,
    deviceId: '001',  // 标签值，会自动生成子表名
    location: 'Beijing',
    type: 'temperature'
  },
  {
    recordTime: new Date(),
    temperature: 28.3,
    humidity: 75.8,
    pressure: 1015.30,
    voltage: 3.2,
    valueJSON: '{"status": "normal"}',
    valueStr: 'normal',
    isSimulation: false,
    deviceId: '002',
    location: 'Shanghai',
    type: 'humidity'
  }
])

// 查询数据
const sensors = await Sensor.find({
  location: 'Beijing',
  temperature: { $gte: 20 }
})

// 窗口查询 - 每10分钟的平均温度
const avgTemp = await Sensor.windowQuery()
  .select(['AVG(temperature) as avg_temp'])
  .where({ deviceId: '001' })
  .interval('10m')
  .fill('PREV')
  .addTimestampPseudoColumns(['_WSTART', '_WEND'])
  .exec()

console.log(avgTemp)

3. 高级查询

// 按设备分组的时间窗口统计
const deviceStats = await Sensor.windowQuery()
  .select(['deviceId', 'AVG(temperature) as avg_temp', 'MAX(humidity) as max_humidity'])
  .partitionBy('deviceId')
  .interval('1h')
  .where({ ts: { $gte: new Date('2024-01-01') } })
  .addTimestampPseudoColumns(['_WSTART', '_WEND'])
  .orderBy('_WSTART')
  .exec()

// 状态窗口查询 - 高温状态持续时间
const highTempSessions = await Sensor.windowQuery()
  .select(['deviceId', 'COUNT(*) as duration'])
  .stateWindow('temperature > 30', '5m')
  .where({ deviceId: '001' })
  .exec()

注意事项

1. 表名大小写: 所有表名会自动转换为小写
2. 关键字验证: 建议开启严格模式避免使用TDengine保留关键字
3. 超级表结构: 超级表创建后会自动同步结构变更
4. 时间戳处理: 时间戳字段会自动转换为Date对象
5. 连接管理: 确保在应用关闭时调用connection.close()
6. 错误处理: 建议对所有数据库操作进行适当的错误处理

安装

npm install nayota-tdenginedb-sdk

环境要求

• Node.js >= 12.0.0
• TDengine >= 2.0.0
• @tdengine/websocket 包（已包含在SDK中）

快速开始

const { TDengineConnection, Schema, model } = require('nayota-tdenginedb-sdk')

// 创建连接
const connection = new TDengineConnection()
await connection.connect('127.0.0.1', 'root', 'taosdata', 6041, 'mydb')

// 定义Schema
const userSchema = new Schema({
  name: { type: String, required: true },
  age: { type: Number }
}, { timestamps: true })

// 创建模型
const User = model('users', userSchema, connection)

// 创建表
await User.createTable()

// 插入数据
await User.insert({ name: 'John Doe', age: 30 })

// 查询数据
const users = await User.find({ age: { $gte: 25 } })
console.log(users)

许可证

本项目采用 MIT 许可证。