数据库

pg_net: 异步网络扩展

pg_net 使 Postgres 能够在 SQL 中发起异步的 HTTP/HTTPS 请求。它与 http 扩展的不同之处在于默认就是异步的，这使得它在阻塞性函数（如触发器）中特别有用。

该扩展消除了服务器需要持续轮询数据库变更的需求，转而允许数据库主动通知外部资源关于重要事件的发生。

启用扩展

  1. 进入 Dashboard 中的数据库页面
  2. 点击侧边栏中的 扩展
  3. 搜索 "pg_net" 并启用该扩展

http_get

创建一个 HTTP GET 请求并返回请求 ID。HTTP 请求会在事务提交后才会实际发起。

签名 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
net.http_get(    -- 请求的URL    url text,    -- 将被URL编码并附加到`url`的键值对    params jsonb default '{}'::jsonb,    -- 包含在请求头中的键值对    headers jsonb default '{}'::jsonb,    -- 请求在被取消前允许的最大毫秒数    timeout_milliseconds int default 2000)    -- 请求ID引用    returns bigint    strict    volatile    parallel safe    language plpgsql

使用方法 

1
2
3
4
5
6
7
select    net.http_get('https://news.ycombinator.com')    as request_id;request_id----------         1(1 row)

http_post

创建一个带有JSON正文的HTTP POST请求，返回请求ID。HTTP请求在事务提交前不会启动。

正文的字符集编码与数据库的server_encoding设置匹配。

签名 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
net.http_post(    -- 请求的URL    url text,    -- POST请求的正文    body jsonb default '{}'::jsonb,    -- 将被URL编码并附加到`url`的键值对    params jsonb default '{}'::jsonb,    -- 包含在请求头中的键值对    headers jsonb default '{"Content-Type": "application/json"}'::jsonb,    -- 请求在被取消前允许的最大毫秒数    timeout_milliseconds int default 2000)    -- 请求ID引用    returns bigint    volatile    parallel safe    language plpgsql

使用方法 

1
2
3
4
5
6
7
8
9
select    net.http_post(        url:='https://httpbin.org/post',        body:='{"hello": "world"}'::jsonb    ) as request_id;request_id----------         1(1 row)

http_delete

创建一个 HTTP DELETE 请求，返回请求的 ID。HTTP 请求将在事务提交后才会开始执行。

函数签名 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
net.http_delete(    -- 请求的URL    url text,    -- 将被URL编码并附加到`url`的键值对    params jsonb default '{}'::jsonb,    -- 包含在请求头中的键值    headers jsonb default '{}'::jsonb,    -- 请求在被取消前允许的最大毫秒数    timeout_milliseconds int default 2000)    -- 返回请求ID引用    returns bigint    strict    volatile    parallel safe    language plpgsql    security definer

使用方法 

1
2
3
4
5
6
7
select    net.http_delete(        'https://dummy.restapiexample.com/api/v1/delete/2'    ) as request_id;----------         1(1 row)

分析响应

等待处理的请求存储在 net.http_request_queue 表中。执行完成后，这些请求会被删除。

1
2
3
4
5
6
7
8
9
CREATE UNLOGGED TABLE    net.http_request_queue (        id bigint NOT NULL DEFAULT nextval('net.http_request_queue_id_seq'::regclass),        method text NOT NULL,        url text NOT NULL,        headers jsonb NOT NULL,        body bytea NULL,        timeout_milliseconds integer NOT NULL    )

当响应返回时，默认情况下会存储在 net._http_response 表中6小时。

1
2
3
4
5
6
7
8
9
10
11
CREATE UNLOGGED TABLE    net._http_response (        id bigint NULL,        status_code integer NULL,        content_type text NULL,        headers jsonb NULL,        content text NULL,        timed_out boolean NULL,        error_msg text NULL,        created timestamp with time zone NOT NULL DEFAULT now()    )

可以通过以下查询查看响应数据：

1
select * from net._http_response;

也可以通过 Supabase 仪表盘的 SQL 编辑器在 net 模式中查看这些数据。

调试请求

检查请求数据

Postman Echo API 会返回与请求相同的响应体和内容，可用于检查发送的数据。

向 echo API 发送 POST 请求

1
2
3
4
5
select    net.http_post(        url := 'https://postman-echo.com/post',        body := '{"key1": "value", "key2": 5}'::jsonb    ) as request_id;

检查 echo API 响应内容以确保包含正确的请求体

1
2
3
4
5
6
select    "content"from net._http_responsewhere id = <request_id>-- 返回请求相关信息-- 包括发送的请求体: {"key": "value", "key": 5}

或者，通过将请求封装在数据库函数中，可以记录或返回发送的行数据用于检查和调试。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
create or replace function debugging_example (row_id int)returns jsonb as $$declare    -- 存储载荷数据    row_data_var jsonb;begin    -- 检索行数据并转换为JSON    select to_jsonb("<example_table>".*) into row_data_var    from "<example_table>"    where "<example_table>".id = row_id;    -- 初始化HTTP POST请求到URL    perform        net.http_post(            url := 'https://postman-echo.com/post',            -- 使用行数据作为载荷            body := row_data_var        ) as request_id;    -- 可选：在Supabase仪表板的Postgres日志中记录行数据或其他数据用于检查    raise log '以JSON格式记录整行数据 (%)', row_data_var;    -- 返回行数据用于检查    return row_data_var;-- 需要时可在此处处理异常exception    when others then        raise exception '发生错误: %', SQLERRM;end;$$ language plpgsql;-- 调用函数select debugging_example(<row_id>);

检查失败的请求

查找所有失败的请求

1
2
3
4
5
select  *from net._http_responsewhere "status_code" >= 400 or "error_msg" is not nullorder by "created" desc;

配置

该扩展默认配置为可靠执行每秒最多200个请求。响应消息仅存储6小时以防止不必要的积累。可以通过重写配置变量来修改默认行为。

获取当前设置

1
2
3
4
5
select  "name",  "setting"from pg_settingswhere "name" like 'pg_net%';

修改设置

更改变量：

1
2
alter role "postgres" set pg_net.ttl to '24 hours';alter role "postgres" set pg_net.batch_size to 500;

然后重新加载设置并通过以下命令重启 pg_net 后台工作进程：

1
select net.worker_restart();

示例

调用 Supabase 边缘函数

向 Supabase 边缘函数发起 POST 请求，附带认证头和 JSON 请求体：

1
2
3
4
5
6
select    net.http_post(        url:='https://project-ref.supabase.co/functions/v1/function-name',        headers:='{"Content-Type": "application/json", "Authorization": "Bearer <YOUR_ANON_KEY>"}'::jsonb,        body:='{"name": "pg_net"}'::jsonb    ) as request_id;

使用 pg_cron 每分钟调用端点

pg_cron 扩展使 Postgres 能够成为自己的 cron 服务器。通过它，您可以安排精确到分钟的定期端点调用。

1
2
3
4
5
6
7
8
9
10
11
12
13
select cron.schedule(	'cron-job-name',	'* * * * *', -- 每分钟执行一次(cron语法)	$$	    -- SQL查询	    select "net"."http_post"(            -- 边缘函数URL            url:='https://project-ref.supabase.co/functions/v1/function-name',            headers:='{"Authorization": "Bearer <YOUR_ANON_KEY>"}'::jsonb,            body:='{"name": "pg_net"}'::jsonb	    ) as "request_id";	$$);

在触发器中执行 pg_net

当触发事件发生时调用外部端点。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
-- 触发器调用的函数create or replace function <function_name>()    returns trigger    language plpgSQLas $$begin    -- 调用pg_net函数net.http_post    -- 发送请求到postman API    perform "net"."http_post"(      'https://postman-echo.com/post'::text,      jsonb_build_object(        'old_row', to_jsonb(old.*),        'new_row', to_jsonb(new.*)      ),      headers:='{"Content-Type": "application/json"}'::jsonb    ) as request_id;    return new;END $$;-- 表更新触发器create trigger <trigger_name>    after update on <table_name>    for each row    execute function <function_name>();

在单个请求中发送多行表数据

1
2
3
4
5
6
7
8
9
10
11
12
13
with "selected_table_rows" as (    select        -- 将所有行转换为JSONB数组        jsonb_agg(to_jsonb(<table_name>.*)) as JSON_payload    from <table_name>    -- 最佳实践是限制最大行数)select    net.http_post(        url := 'https://postman-echo.com/post'::text,        body := JSON_payload    ) AS request_idFROM "selected_table_rows";

更多示例可在扩展的GitHub页面查看

限制说明

  • 为了提高速度和性能，请求和响应数据存储在非日志表中，这些表在崩溃或不正常关机时不会被保留
  • 默认情况下，响应数据仅保存6小时
  • 仅支持使用JSON数据格式的POST请求，不支持其他数据格式
  • 设计处理能力上限为每秒200个请求，提高请求速率可能导致系统不稳定
  • 不支持PATCH/PUT请求
  • 一次只能操作一个数据库，默认使用postgres数据库

相关资源