当前位置 博文首页 > qq262593421的博客：python3连接postgresql工具类
目录
一、环境准备
1、python3安装
2、MongoDB4安装
二、安装驱动
1、psycopg2驱动安装
2、pymogo驱动测试
三、py工具类
1、工具脚本
postgresql官网：https://www.postgresql.org
postgresql中文社区：http://www.postgres.cn/index.php/v2/home
postgresql12中文手册：http://www.postgres.cn/docs/12/index.html
菜鸟教学：https://www.runoob.com/postgresql/postgresql-syntax.html
windows离线安装python3.6.8环境：https://blog.csdn.net/qq262593421/article/details/111309116
Linux 离线安装 python 3.6.10：https://blog.csdn.net/qq262593421/article/details/105395101
win10安装postgresql：https://blog.csdn.net/qq262593421/article/details/114318815
linux安装postgresql：https://blog.csdn.net/qq262593421/article/details/114011921
pip install psycopg2python
import psycopg2#-*- 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)