利用Python查询postgresql数据库数据

今天药忘吃喽~ 2023-07-04 08:35 7阅读 0赞

这里提供两种查询方式：

首先配置连接信息：  
出于安全性考虑，涉及到的配置信息都存放在外部文档里。

import os, sys
    import pandas as pd
    from sqlalchemy import create_engine
    import psycopg2
    from tabulate import tabulate
    
    CurrentFile = os.path.abspath(__file__)
    CurrentPath = os.path.dirname(CurrentFile)
    FatherPath = os.path.dirname(CurrentPath)
    sys.path.insert(0, FatherPath)
    import logo
    
    logger = logo.create_logo(os.getpid(), __name__, os.path.join(FatherPath, 'log/tableau_update_info.log'))
    
    import configparser
    
    cf = configparser.ConfigParser()
    cf.read(os.path.join(FatherPath, 'doc/config.ini'))
    postgre_dict = dict(cf.items('tableau-postgre'))
    
    host = postgre_dict['host']
    user = postgre_dict['user']
    pwd = postgre_dict['pwd']
    db = postgre_dict['db']
    port = postgre_dict['port']

# 1. 通过psycopg2库 #

def query_by_psycopg2():
        # 建立连接
        connect = psycopg2.connect(host=host, port=int(port), user=user, password=pwd, database=db)
        try:
            with connect.cursor() as cursor:
                # 获取表头
                sqlstr_cols = """ select column_name FROM information_schema.columns WHERE table_schema = 'public' and table_name='workbooks' """
                cursor.execute(sqlstr_cols)
                cols = [x[0] for x in cursor.fetchall()]
                # 获取数据
                sqlstr = 'select * from public.workbooks'
                cursor.execute(sqlstr)
                res = cursor.fetchall()
                df = pd.DataFrame(res, columns=cols)
                return df
        except Exception as e:
            logger.info(F'查询失败，详情:{e}')
        finally:
            connect.close()

# 2. 通过sqlalchemy #

def query_by_sqlalchemy():
        connect = create_engine(F"postgresql://{user}:{pwd}@{host}:{port}/{db}")
        sqlstr = 'select * from public.workbooks'
        df = pd.read_sql(sqlstr, connect)
        connect.dispose()
        return df

# 单元测试 #

if __name__ == '__main__':
        df = query_by_psycopg2()
        # df = query_by_sqlalchemy()
        print(df.columns)
        print(tabulate(df, headers=df.columns))

![image.png][]

[image.png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9jZG4ubmxhcmsuY29tL3l1cXVlLzAvMjAyMC9wbmcvNjcxOTcwLzE1ODEyMjU0MTc4MTktYmQyNWFkYWUtMTU5Mi00YjRmLThmNjUtODg2MmRiODNjZDFhLnBuZw?x-oss-process=image/format,png#align=left&display=inline&height=380&name=image.png&originHeight=760&originWidth=2680&size=162587&status=done&style=none&width=1340