python3连接postgresql工具类

小灰灰 2022-11-05 12:52 163阅读 0赞

**目录**

一、环境准备

1、python3安装

2、MongoDB4安装

二、安装驱动

1、psycopg2驱动安装

2、pymogo驱动测试

三、py工具类

1、工具脚本

--------------------

## 一、环境准备 ##

postgresql官网：[https://www.postgresql.org][https_www.postgresql.org]

postgresql中文社区：[http://www.postgres.cn/index.php/v2/home][http_www.postgres.cn_index.php_v2_home]

postgresql12中文手册：[http://www.postgres.cn/docs/12/index.html][http_www.postgres.cn_docs_12_index.html]

菜鸟教学：[https://www.runoob.com/postgresql/postgresql-syntax.html][https_www.runoob.com_postgresql_postgresql-syntax.html]

### 1、python3安装 ###

windows离线安装python3.6.8环境：[https://blog.csdn.net/qq262593421/article/details/111309116][https_blog.csdn.net_qq262593421_article_details_111309116]

Linux 离线安装 python 3.6.10：[https://blog.csdn.net/qq262593421/article/details/105395101][https_blog.csdn.net_qq262593421_article_details_105395101]

### 2、MongoDB4安装 ###

win10安装postgresql：[https://blog.csdn.net/qq262593421/article/details/114318815][https_blog.csdn.net_qq262593421_article_details_114318815]

linux安装postgresql：[https://blog.csdn.net/qq262593421/article/details/114011921][https_blog.csdn.net_qq262593421_article_details_114011921]

##  ##

## 二、安装驱动 ##

### 1、psycopg2驱动安装 ###

pip install psycopg2

### 2、pymogo驱动测试 ###

python
    import psycopg2

##  ##

## 三、py工具类 ##

### 1、工具脚本 ###

#-*- encoding: utf-8 -*-
    
    import psycopg2
    
    class PGSQLUtil:
        """
        PostgreSQL工具类
        """
        def __init__(self, host="127.0.0.1", user=None, password=None, database=None, port=5432, dsn=None, connection_factory=None, cursor_factory=None, **kwargs):
            """构造函数"""
            self.__database = database
            self.__conn = psycopg2.connect(host=host, user=user, password=password, database=database, port=str(port),
                                           dsn=dsn, connection_factory=connection_factory, cursor_factory=cursor_factory, **kwargs)
            self.__cursor = self.__conn.cursor()
    
        def __del__(self):
            """析构函数"""
            self.__cursor.close()
            self.__conn.close()
    
        def get_conn(self):
            """获取连接"""
            return self.__conn
    
        def get_cursor(self):
            """获取游标"""
            return self.__cursor
    
        def list_databases(self, vars=None):
            """查询所有数据库"""
            self.__cursor.execute("SELECT pg_database.datname FROM pg_database", vars)
            dbs = []
            for db in self.__cursor.fetchall():
                dbs.append(db[0])
            return dbs
    
        def list_user_tables(self, vars=None):
            """查询当前用户所有表"""
            self.__cursor.execute("SELECT tablename FROM pg_tables WHERE schemaname='public'", vars)
            # self.__cursor.execute("SELECT * FROM  information_schema.tables WHERE table_schema='public'", vars)
            tables = []
            for table in self.__cursor.fetchall():
                tables.append(table[0])
            return tables
    
        def list_tables_privileges(self, vars=None):
            """查询所有表的权限"""
            self.__cursor.execute("SELECT * FROM  information_schema.table_privileges", vars)
            return self.__cursor.fetchall()
    
        def execute(self, sql, vars=None):
            """获取SQL执行结果"""
            self.__cursor.execute(sql, vars)
            return self.__cursor.fetchall()
    
        def get_version(self, vars=None):
            """获取MySQL版本"""
            self.__cursor.execute("SELECT VERSION()", vars)
            version = self.__cursor.fetchone()
            print("Postgresql Version : %s" % version)
            return version[0]
    
        def list_table_metadata(self, vars=None):
            """查询所有表的元数据信息"""
            sql = "SELECT * FROM information_schema.TABLES WHERE TABLE_TYPE !='SYSTEM VIEW' AND TABLE_SCHEMA NOT IN ('sys','mysql','information_schema','performance_schema')"
            self.__cursor.execute(sql, vars)
            return self.__cursor.fetchall()
    
        def get_table_fields(self, table, vars=None):
            """获取表字段信息"""
            sql = "SELECT column_name FROM information_schema.COLUMNS WHERE table_schema='public' AND table_name='"+ table +"'"
            self.__cursor.execute(sql, vars)
            fields = []
            for field in self.__cursor.fetchall():
                fields.append(field[0])
            return fields
    
        def table_metadata(self, table, vars=None):
            """获取表字段信息"""
            sql = "SELECT * FROM information_schema.COLUMNS WHERE table_schema='public' AND table_name='"+ table +"'"
            self.__cursor.execute(sql, vars)
            fields = []
            for field in self.__cursor.fetchall():
                fields.append(field)
            return fields
    
    if __name__ == "__main__":
        pgsqlUtil = PGSQLUtil(host="127.0.0.1", user="root", password="123456", database="xl")
        pgsqlUtil.get_version()
        conn = pgsqlUtil.get_conn()
        ## 查询所有数据库
        databases = pgsqlUtil.list_databases()
        print(type(databases), databases)
        ## 查询当前用户的表
        tables = pgsqlUtil.list_user_tables()
        print(type(tables), tables)
        ## 查询所有表的权限信息
        privileges = pgsqlUtil.list_tables_privileges()
        for privilege in privileges:
            print(privilege)
        ## 查询表数据
        result = pgsqlUtil.execute("SELECT * FROM t_user")
        for i in result:
            print(i)
    
        # result = pgsqlUtil.get_table_fields("t_user")
        result = pgsqlUtil.table_metadata("t_user")
        for i in result:
            print(i)

[https_www.postgresql.org]: https://www.postgresql.org
[http_www.postgres.cn_index.php_v2_home]: http://www.postgres.cn/index.php/v2/home
[http_www.postgres.cn_docs_12_index.html]: http://www.postgres.cn/docs/12/index.html
[https_www.runoob.com_postgresql_postgresql-syntax.html]: https://www.runoob.com/postgresql/postgresql-syntax.html
[https_blog.csdn.net_qq262593421_article_details_111309116]: https://blog.csdn.net/qq262593421/article/details/111309116
[https_blog.csdn.net_qq262593421_article_details_105395101]: https://blog.csdn.net/qq262593421/article/details/105395101
[https_blog.csdn.net_qq262593421_article_details_114318815]: https://blog.csdn.net/qq262593421/article/details/114318815
[https_blog.csdn.net_qq262593421_article_details_114011921]: https://blog.csdn.net/qq262593421/article/details/114011921