程序示例

连接数据库

1. psycopg2

import psycopg2

database = "unvbd" user = "unvbd" password = "123456"
host = "192.168.65.129"
port = "5678"

if __name__ == "__main__":
    try:
        conn = psycopg2.connect(
            "dbname={} user={} password={} host={} port={}"
            .format(database, user, password, host, port))
    except Exception, e:
        print(e)
        print("error")
    else:
        print("success")

2. psycopg3

import psycopg

database = "unvdb"
user = "unvdb"
password = "123"
host = "localhost"
port = "6666"

if __name__== "__main__":
        try:
            conn = psycopg.connect(
                "dbname={} user={} password={} host={} port={}"
                .format(database, user, password, host, port))
            cur = conn.cursor()

            cur.execute('select version()')
            rows = cur.fetchall()
            for row in rows:
                print(row)
        except Exception as e:
            print(e)
            print("error")
        else:
            print("success")

获取结果集

1. psycopg2

import psycopg2

database = "unvdb"
user = "unvdb"
password = "123456"
host = "192.168.65.129"
port = "5678"

if  name   == " main ": 
    try:
        conn = psycopg2.connect(
            "dbname={} user={} password={} host={} port={}"
            .format(database, user, password, host, port))

        cur = conn.cursor()

        cur.execute('drop table if exists test_ksy')
        cur.execute('create table test_ksy(id integer, name TEXT)')
        cur.execute("insert into test_ksy values(%s, %s)", (1, "jiuyou"))
        cur.execute("insert into test_ksy values(%s, %s)", (2, '九有数据库'))
        cur.execute("insert into test_ksy values(%s, %s)", (3, '%$#@!……'))
        conn.commit()
        cur.execute("select * from test_ksy") 
        rows = cur.fetchall()
        for row in rows:
            print(cell)

        cur.close()
        conn.commit()
        conn.close()

    except Exception, e:
        print(e) print("error")
    else:
        print("success")

2. psycopg3

import psycopg

database = "unvdb"
user = "unvdb"
password = "123"
host = "localhost"
port = "6666"

if __name__== "__main__":
        try:
            conn = psycopg.connect(
                "dbname={} user={} password={} host={} port={}"
                .format(database, user, password, host, port))
            cur = conn.cursor()

            cur.execute('drop table if exists test_ksy')
            cur.execute('create table test_ksy(id integer, name TEXT)')

            cur.execute("insert into test_ksy values(%s, %s)", (1, "jiuyou"))
            cur.execute("insert into test_ksy values(%s, %s)", (2, '九有数据库'))
            cur.execute("insert into test_ksy values(%s, %s)", (3, '%$#@!……'))
            conn.commit()
            cur.execute("select * from test_ksy")
            rows = cur.fetchall()
            for row in rows:
                print(row)

        except Exception as e:
            print(e)
            print("error")
        else:
            print("success")

存储过程

/* 创建存储过程 */
CREATE FUNCTION reffunc(refcursor) RETURNS refcursor AS 
$$ BEGIN
    OPEN $1 FOR SELECT col FROM test; 
    RETURN $1;
END;
$$ LANGUAGE plsql；

/* 调用示例 */
curl = conn.cursor()
curl.callproc('reffunc', ['curname'])

cur2 = conn.cursor('curname') 
for record in cur2:
    print(record) pass