电气SQL 的脚本客户端

现代应用的实时Postgres同步.

电气提供 HTTP 接口 到 Postgres 使大量客户端能够查询并实时更新数据库子集,称为 形状通过这种方式,Electric将Postgres转换成实时数据库.

TypeScript客户端帮助从浏览器和其他JavaScript环境中的HTTP API读取形状,如边缘功能和服务器侧节点/Bun/Deno应用程序. 它支持精细和粗细的活性模式——你可以订阅看到每一个变化的行,或者只需订阅就可以在变化时获得整个形状. 客户端也通过基于函数的参数和头来支持动态选项,使其易于处理认证符,用户上下文等运行时值.

安装

客户在国家预防机制上公布为: @electric-sql/client数字 :

npm i @electric-sql/client

如何使用

客户输出 a ShapeStream 逐行更新到形状以及 Shape 用于获取整个形状的更新.

ShapeStream

import { ShapeStream } from '@electric-sql/client'

const BASE_URL = 'http://localhost:3000'

// Passes subscribers rows as they're inserted, updated, or deleted
const stream = new ShapeStream({
  url: `${BASE_URL}/v1/shape`,
  params: {
    table: `foo`,
  },
})

// You can also add custom headers and URL parameters
const streamWithParams = new ShapeStream({
  url: `${BASE_URL}/v1/shape`,
  headers: {
    Authorization: 'Bearer token',
  },
  params: {
    table: `foo`,
    'custom-param': 'value',
  },
})

stream.subscribe((messages) => {
  // messages is an array with one or more row updates
  // and the stream will wait for all subscribers to process them
  // before proceeding
})

Shape

import { ShapeStream, Shape } from '@electric-sql/client'

const BASE_URL = 'http://localhost:3000'

const stream = new ShapeStream({
  url: `${BASE_URL}/v1/shape`,
  params: {
    table: `foo`,
  },
})
const shape = new Shape(stream)

// Returns promise that resolves with the latest shape data once it's fully loaded
await shape.rows

// passes subscribers shape data when the shape updates
shape.subscribe(({ rows }) => {
  // rows is an array of the latest value of each row in a shape.
})

反应原生/ Expo 应用程序生命周期

大都会和博览会解决软件包 react-native 用于本地建筑的导出, 因此标准导入自动电线 React Introduction AppState 输入 ShapeStream。如果您的捆绑器设置不能解决根 react-native 条件,使用明示 @electric-sql/client/react-native 入境点或通过 runtimeVisibility 当然 见 类型脚本客户端 Docs 详细情况.

处理错误

ShepeStream 以自动重试支持提供强大的错误处理:

1. 有重试控制的流层错误处理器

这个 onError 处理器给予您对错误回收的全部控制 :

import { ShapeStream, FetchError } from '@electric-sql/client'

const BASE_URL = 'http://localhost:3000'
const getRefreshedToken = () => 'refreshed-token'
const fallbackUserId = 'fallback-user-id'

const stream = new ShapeStream({
  url: `${BASE_URL}/v1/shape`,
  params: { table: `foo` },
  onError: (error) => {
    console.error('Stream error:', error)

    // IMPORTANT: Return an object to keep syncing!
    // Return void/undefined to stop syncing permanently.

    // Note: 5xx errors and network errors are automatically retried,
    // so onError is mainly for handling client errors (4xx)

    if (error instanceof FetchError) {
      if (error.status === 401) {
        // Unauthorized - refresh token and retry
        const newToken = getRefreshedToken()
        return {
          headers: {
            Authorization: `Bearer ${newToken}`,
          },
        }
      }

      if (error.status === 403) {
        // Forbidden - maybe change user context
        return {
          params: {
            table: `foo`,
            where: `user_id = $1`,
            params: [fallbackUserId],
          },
        }
      }
    }

    // Stop syncing for other errors (return void)
  },
})

关键编号: onError 调用返回值控制同步是否继续 :

• 返回对象 (甚至为空) {}重试同步 :
  • {} - 使用相同的参数和标题重试
  • { params } - 使用经修改的参数重试
  • { headers } - 用修改后的标题重试
  • { params, headers } - 用两个修改过的重试
• 返回为空/ 未定义 以永久停止溪流

处理器支持同步操作 :

import { FetchError } from '@electric-sql/client'

const refreshAuthToken = async () => 'refreshed-token'

const options = {
  onError: async (error: Error) => {
    if (error instanceof FetchError && error.status === 401) {
      // Perform async token refresh
      const newToken = await refreshAuthToken()
      return {
        headers: { Authorization: `Bearer ${newToken}` },
      }
    }

    return {} // Retry other errors
  },
}

自动重试:客户端自动重试5xx服务器出错,网络出错,以及429个速率限制并有指数回放. 这个 onError 只有在这些重试被用尽后,或者对于像4xx客户端错误这样的不可重复的错误才引用回调.

无 onError: 如果没有 onError 提供了处理器,将抛出不可重复的错误(如4xx客户端错误)并停止流.

2. 订阅级错误回调

个人订阅者可以处理其订阅时特有的错误:

stream.subscribe(
  (messages) => {
    // Handle messages
  },
  (error) => {
    // Handle errors for this specific subscription
    console.error('Subscription error:', error)
  }
)

注意: 订阅错误召回无法控制重试行为 - 使用流级 onError 对于这个.

常见错误类型

设置错误 :

• MissingShapeUrlError: 缺少所需的 URL 参数
• InvalidSignalError无效的中止实例
• ReservedParamError: 使用保留参数名称

运行时间出错 :

• FetchError: HTTP 在取形状时出错(包括 status, (中文(简体) ), url, (中文(简体) ), headers(中文(简体) )
• FetchBackoffAbortError: 使用中止符号获取中止
• MissingShapeHandleError: 缺少所需的形状手柄
• ParserNullValueError: NULL 值在不可变列中

见 类型脚本客户端 Docs 更多细节.

总而言之,看看 文件网站 财务报告和审定财务报表 实例 更多信息.

开发

在 repo 根上安装 pnpm 工作空间 :

pnpm install

构建软件包 :

cd packages/typescript-client
pnpm build

测试

在一个终端中,启动后端运行:

cd ../sync-service
mix deps.get
mix stop_dev && mix compile && mix start_dev && iex -S mix

在此文件夹中:

pnpm test