在postgresql中用SQL封装python

“ 对于一个数据库来说，SQL大家肯定最熟悉不过了。但是作为数据库开发者，我认为数据库不应该只是支持SQL语言。应该支持更多编程语言。比如python、java、c++等更多编程语言，让数据库在多种语言之间的管理、功能上实现最大便捷，这才是未来数据库最大的发展趋势。也是国产数据库在实现崛起的弯道之一。就像手机不仅能打电话，还要能拍照。汽车不仅需要发动机，也需要冰箱、彩电、大沙发。”

01

—

简介

        在postgresql数据库中，需要安装plpythonu的extension，才可以使用python语言，plpythonu分两种。

        一种是python2的版本plpython2u，

        一种是python3的版本plpython3u。

02

—

postgresql16.1安装

安装依赖

yum install -y bison flex readline-devel zlib-devel zlib zlib-devel gcc  gcc-c++ openssl-devel python3  python3-devel libicu-devel ncurses-devel sqlite-devel tk-devel gcc make
yum install -y bison flex readline-devel zlib-devel zlib zlib-devel gcc  gcc-c++ openssl-devel python3  python3-devel libicu-devel ncurses-devel sqlite-devel tk-devel gcc make

添加用户

useradd postgres 
vim /etc/sudo

在101行加入以下内容

postgres ALL=(ALL)     NOPASSWD: ALL
postgres ALL=(ALL)     NOPASSWD: ALL

进入官网找到链接，这里使用源码安装。

wget https://ftp.postgresql.org/pub/source/v16.1/postgresql-16.1.tar.gz
wget https://ftp.postgresql.org/pub/source/v16.1/postgresql-16.1.tar.gz

解压

useradd postgres 
 mv postgresql-16.1.tar.gz /home/postgres
 su - postgres 
 tar -zxf postgresql-16.1.tar.gz
 cd postgresql-16.1

这里编译python支持还是很重要。--with-python 自行构建plpython3u插件

./configure --prefix=/home/postgres/pg --with-openssl  --with-python

echo $?
echo $?

返回值是零便是编译无报错。

进行构建

make && make install
make && make install

cd /home/postgres/pg/share/extension
cd /home/postgres/pg/share/extension

此时可以看见plpython3u.control文件已经存在

编辑环境变量

cd 
vim .bash_profile

加入以下变量

export PATH=/home/postgres/pg/bin:$PATH 
export PGDATA=/home/postgres/pg/data

加载

source ~/.bash_profile
source ~/.bash_profile

初始化数据库

initdb -D $PGDATA -U postgres -W 
(输入超级用户密码两次)
pg_ctl start 
pg_ctl status

03

—

创建拓展

postgres=# create extension plpython3u ;
CREATE EXTENSION
postgres=# select  *  from pg_extension ;
  oid  |  extname   | extowner | extnamespace | extrelocatable | extversion | extconfig | extcondition 
-------+------------+----------+--------------+----------------+------------+-----------+--------------
 14270 | plpgsql    |       10 |           11 | f              | 1.0        |           | 
 16384 | plpython3u |       10 |           11 | f              | 1.0        |           | 
(2 rows)

在已安装plpython3u的情况下，可以下载最新的plpython3u。然后在数据库中加关键字CASCADE，对其进行升级

CREATE EXTENSION plpython3u CASCADE;
CREATE EXTENSION plpython3u CASCADE;

04

—

跨IP文本文件的实时同步

    在postgresql数据库中编写python，一样遵循其“缩进规则”，

 在业务中，在一些业务服务器上往往会在每隔一段时间生成一个文档存储在指定路径，此时，我们需要实时读取该文件内容让其数据落地，本文举例说明。

\! pip3 install -i https://pypi.tuna.tsinghua.edu.cn/simple paramiko

在postgresql数据库中创建函数用于读取远程服务器的文件内容，读取完之后本文将其删掉（在生产中可以将其移动到归档路径中去）。

drop function if exists query_data(host varchar(100), username varchar(100), password varchar(100), remote_file_path varchar(100));
CREATE OR REPLACE FUNCTION query_data(host varchar(100), username varchar(100), password varchar(100), remote_file_path varchar(100))
RETURNS varchar(9999)
AS $$
import paramiko

def read_and_delete_remote_file(host, username, password, remote_file_path):
    try:
        with paramiko.SSHClient() as client:
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            client.connect(host, username=username, password=password)

            with client.open_sftp() as sftp:
                with sftp.file(remote_file_path, 'r') as remote_file:
                    file_content = remote_file.read()

                # 删除远程文件
                sftp.remove(remote_file_path)

        lines = file_content.decode('utf-8').splitlines()

        # 生成插入语句的后半段
        values_part = ', '.join(["('" + "', '".join(line.split(',')) + "')" for line in lines[1:]])

        # 返回插入语句的后半段
        return f'VALUES {values_part};'

    except Exception as e:
        print(f"Error: {e}")

insert_statement = read_and_delete_remote_file(host, username, password, remote_file_path)
return insert_statement

$$ LANGUAGE plpython3u ;

注：本博主没找到合适的返回值类型，所以将其返回为字符串，在后面的函数调用中使用动态SQL将其拼接并执行。读者对于这里的返回值类型有更好的建议，望私信告知。

创建一个装载表

CREATE TABLE IF NOT EXISTS public.texttable
(
    name character varying(100) ,
    age character varying(100) ,
    city character varying(100) ,
    occupation character varying(100) ,
    salary character varying(100) ,
    load_time timestamp without time zone DEFAULT now()
);

再创建一个函数用于在指定时间中去调用query_data函数

CREATE OR REPLACE FUNCTION insert_text(host varchar(100), username varchar(100), password varchar(100), remote_file_path varchar(100))
RETURNS void AS $$
declare data_query varchar(4000);
BEGIN
    SELECT query_data(host,username,password,remote_file_path) into data_query;
    EXECUTE 'INSERT INTO texttable (Name, Age, City, Occupation, Salary)' ||data_query;
END;
$$ LANGUAGE plpgsql;

进行调用

select insert_text('10.0.0.107','root','123456','/root/text.txt'); 
 select  *  from texttable

可以看到数据成功进入。

本博主有考虑将其实时调用这个问题

这里可以使用使用循环将函数改写，使其能一直读取该路径的文件内容

CREATE OR REPLACE FUNCTION insert_text_sleep(host VARCHAR(100), username VARCHAR(100), password VARCHAR(100), remote_file_path VARCHAR(100))
RETURNS VOID AS $$
DECLARE
    data_query VARCHAR(4000);
  i    integer :=0;
BEGIN
    WHILE true LOOP
   RAISE NOTICE '已经执行第%次数',i;
        SELECT query_data(host, username, password, remote_file_path) INTO data_query;
    if data_query is null or trim(data_query) = 'VALUES' then
    PERFORM   pg_sleep(10);
    else 
        EXECUTE 'INSERT INTO texttable (Name, Age, City, Occupation, Salary) ' || data_query;
    end if ;
    i=i+1;
    END LOOP;
END;
$$ LANGUAGE plpgsql;

查看锁状态的时候

select  *  from pg_locks;
select  *  from pg_locks;

oid=16424  为texttable 表，此时会被加上RowExclusiveLock锁，如果事务不中断，select 就无法查看到最新数据。要想实现实时调度，还是需要借用外部工具，例如kettle   contrab 等。

select insert_text('10.0.0.107','root','123456','/root/text.txt');