Python从SQL Server中查询数据

清疚 2023-07-05 10:05 16阅读 0赞

该项目开发主要使用SQL Server数据库，但有时候需要用到Python与数据库交互：查询数据并导出、往数据库里插入数据等操作。以下脚本为从数据库里通过查询获取数据，进而导出或存储等操作。  
操作流程如下：

1.  首先通过ip，用户名、密码、要连接到的数据库名建立连接。这里还有一个charset参数需要格外说一下，我之前建立连接的时候并没有指定该参数，查询的结果也是没问题的。但有一次查询到的结果为乱码，发现是编码方式的问题，故后来就把该参数也添加了进来。该参数一般指定为GBK即可。
2.  建立连接后，就可以开始通过游标cursor执行sql语句达到我们想要的结果。
3.  sql语句执行完后，再通过cursor.fetchall获取数据。

import os, sys
    import pymssql
    import pandas as pd
    import warnings
    
    warnings.filterwarnings('ignore')
    
    CurrentFile = os.path.abspath(__file__)
    CurrentPath = os.path.dirname(CurrentFile)
    FatherPath = os.path.dirname(CurrentPath)
    import configparser
    cf = configparser.ConfigParser()
    # 出于安全性考虑，这些配置信息都存放到doc下的配置文件里。
    cf.read(os.path.join(FatherPath, 'doc/config.ini'))
    mscon_dict = dict(cf.items('mssql-config'))
    
    class MssqlConnect(object):
        def __init__(self):
            self.host = mscon_dict['host']
            self.user = mscon_dict['user']
            self.pwd = mscon_dict['pwd']
            self.db = mscon_dict['db']
            self.charset = mscon_dict['charset']
    
        def query(self, tablename):
            logger.info('>>> 开始连接数据库')
            try:
                connect = pymssql.connect(host = self.host,
                                          user = self.user,
                                          password = self.pwd,
                                          database = self.db,
                                          charset=self.charset
                                          )
                logger.info('>>> 数据库连接成功')
                try:
                    with connect.cursor() as cursor:
                        sqlstr_cols = F"SELECT COLUMN_NAME,DATA_TYPE FROM INFORMATION_SCHEMA.columns WHERE TABLE_NAME= '{tablename}'"
                        cursor.execute(sqlstr_cols)
                        cols = [col[0] for col in cursor.fetchall()]
                        cursor.execute(f'select * from {tablename}')
                        res = cursor.fetchall()
                        df = pd.DataFrame(res,columns=cols)
                except Exception as e:
                    df = pd.DataFrame()
                    logger.info(F'>>> 查询数据失败，查询语句为：{sqlstr}；\n错误信息为：{e}')
                finally:
                    connect.close()
            except Exception as e:
                df = pd.DataFrame()
                logger.info('>>> 数据库连接失败，请检查配置信息是否准确。错误信息为：{}'.format(e))
            return df
        
    if __name__ == '__main__':
        from tabulate import tabulate
        tablename = 'STG_END_MAIL_CONF'
        t0 = time.time()
        df = MssqlConnect().query(tablename)
        print(tabulate(df,headers=df.columns))