Python:操作PostgreSQL数据库(使用PyGreSQL)

这里()写了使用python写了操作嵌入式数据库SQLite的方法今天用python写了针对个人比较熟悉的开源数据库PostgreSQL的常用操作开发过程简介如下

　　一环境信息 

　　操作系统

　　RedHat Enterprise Linux

　　Windows XP SP

　　数据库

　　PostgreSQL

　　 开发工具:

　　Eclipse+Pydev+python+PyGreSQL(提供pg模块)

　　说明

　　aPostgreSQL数据库运行于RedHat Linux上Windows下也要安装pgAdmin（访问PostgreSQL服务器的客户端）

　　bPyGreSQL(即pg)模块下载路径及API手册

　　二配置

　　将pgAdmin安装路径下以下子目录添加到系统环境变量中:

　　E:\Program Files\PostgreSQL\\lib

　　E:\Program Files\PostgreSQL\\bin

　　将python安装目录C:\Python\Lib\sitepackages\pywin_system下的dll文件拷贝到C:\WINDOWS\system

　　说明

　　a 如果跳过以上两步在import pg时将会报错浪费较长时间才搞定

　　b 如果对PostgreSQL安装和配置不熟悉请参考本博客中以下几篇文章

　　《RedHat Linux上安装PostgreSQL》

　　《PostgreSQL服务端监听设置及客户端连接方法》

　　《PostgreSQL数据库创建删除方法》

　　三程序实现

　　[python] view plaincopyprint?

#!/usr/bin/env python
# * coding: utf *

#导入日志及pg模块
import logging  
import nfig  
import pg  

#日志配置文件名
LOG_FILENAME = nf

#日志语句提示信息
LOG_CONTENT_NAME = pg_log

def log_init(log_config_filename logname):  

    Function:日志模块初始化函数
    Inputlog_config_filename:日志配置文件名
           lognmae:每条日志前的提示语句
    Output: logger
    author: socrates
    date:
    
    nfigfileConfig(log_config_filename)  
    logger = logginggetLogger(logname)  
return logger  

def operate_postgre_tbl_product():  

    Function:操作pg数据库函数
    InputNONE
    Output: NONE
    author: socrates
    date:
    
    pgdb_loggerdebug(operate_postgre_tbl_product enter)   

#连接数据库  
try:  
        pgdb_conn = nnect(dbname = kevin_test host =  user = dyx passwd = )  
except Exception e:  
print eargs[]  
         pgdb_loggererror(conntect postgre database failed ret = %s % eargs[])      
return

    (conntect postgre database(kevin_test) succ)   

#删除表
    sql_desc = DROP TABLE IF EXISTS tbl_product;
try:  
        pgdb_connquery(sql_desc)  
except Exception e:  
printdrop table failed
        pgdb_loggererror(drop table failed ret = %s % eargs[])  
        pgdb_connclose()    
return

    (drop table(tbl_product) succ)   

#创建表
    sql_desc = CREATE TABLE tbl_product(
        i_index INTEGER
        sv_productname VARCHAR()
        );
try:      
        pgdb_connquery(sql_desc)  
except Exception e:  
printcreate table failed
        pgdb_loggererror(create table failed ret = %s % eargs[])  
        pgdb_connclose()    
return

    (create table(tbl_product) succ)   

#插入记录   
    sql_desc = INSERT INTO tbl_product(sv_productname) values(apple)
try:  
        pgdb_connquery(sql_desc)  
except Exception e:  
printinsert record into table failed
        pgdb_loggererror(insert record into table failed ret = %s % eargs[])  
        pgdb_connclose()    
return

    (insert record into table(tbl_product) succ)       

#查询表        
    sql_desc = select * from tbl_product
for row in pgdb_connquery(sql_desc)dictresult():  
print row  
        (%s row)   

#查询表        
    sql_desc = select * from tbl_test_port
for row in pgdb_connquery(sql_desc)dictresult():  
print row   
        (%s row)          

#关闭数据库连接     
    pgdb_connclose()         
    pgdb_loggerdebug(operate_sqlite_tbl_product leaving)   

if __name__ == __main__:   

#初始化日志系统
    pgdb_logger = log_init(LOG_FILENAME LOG_CONTENT_NAME)     

#操作数据库
    operate_postgre_tbl_product()  

四测试

　　运行后命令行打印结果

　　[plain] view plaincopyprint?

{sv_productname: apple i_index: None}  
{i_status:  i_port:  i_index: }  
{i_status:  i_port:  i_index: }  
{i_status:  i_port:  i_index: }  
{i_status:  i_port:  i_index: }  
{i_status:  i_port:  i_index: }  
{i_status:  i_port:  i_index: }  
{i_status:  i_port:  i_index: }  
{i_status:  i_port:  i_index: }  
{i_status:  i_port:  i_index: }  
{i_status:  i_port:  i_index: }  
{i_status:  i_port:  i_index: }  
{i_status:  i_port:  i_index: }  

日志文件内容

　　[plain] view plaincopyprint?

[ ::  pg_log]DEBUG:  operate_postgre_tbl_product enter (test_funcpy:)  
[ ::  pg_log]INFO:  conntect postgre database(kevin_test) succ (test_funcpy:)  
[ ::  pg_log]INFO:  drop table(tbl_product) succ (test_funcpy:)  
[ ::  pg_log]INFO:  create table(tbl_product) succ (test_funcpy:)  
[ ::  pg_log]INFO:  insert record into table(tbl_product) succ (test_funcpy:)  
[ ::  pg_log]INFO:  {sv_productname: apple i_index: None} (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]INFO:  {i_status:  i_port:  i_index: } (test_funcpy:)  
[ ::  pg_log]DEBUG:  operate_sqlite_tbl_product leaving (test_funcpy:)  

psql查看结果

　　[plain] view plaincopyprint?

[root@kevin ~]# su  postgres  
[postgres@kevin ~]$ psql U dyx d kevin_test  
psql ()  
Type help for help  

kevin_test=# \dt  
                List of relations  
 Schema |     Name      | Type  |     Owner        
+++  
 public | tbl_product  | table | dyx  
 public | tbl_test_port | table | pg_test_user_  
( rows)  

kevin_test=# select * from tbl_product;  
 i_index | sv_productname   
+  
         | apple  
( row)  

kevin_test=# select * from tbl_test_port;  
 i_index | i_port | i_status   
++  
        |       |          
        |       |          
        |       |          
        |       |          
        |      |          
        |       |          
        |       |          
       |      |          
       |      |          
       |      |          
        |    |          
        |    |          
( rows)  

kevin_test=# \q  
[postgres@kevin ~]$