Python 调用sqlserver

声明：本文为偏方

python3

缘由：

pymssql速度感人（不是说好的 Cython吗），连续操作约8000-9000次就莫名报错（还是从微软网站下载的）
c/c++手生

因此百度了一个c#的DBHelper测试了 一下，效果不错

基本思路：

首先pythonnet可以调用C#的dll文件
其次需要调整DBHelper

注意以下几点基本上可以就调整DBHelper了

• python的dict貌似没法往dll里传
• python的list传过去被识别为数组
• python的None传入C#中为null

python中写了module衔接一下

其它还可能有点风格差异

代码：

多了有点繁杂，贴两个当例子，使用请慎重

DBHelper

namespace MSSQL
{
    /// <summary>
    /// 用于Python调用SqlServer
    /// by gz.18.10.20
    /// </summary>
    public static class DBHelper
    {       
         private static void PrepareCommand(SqlCommand cmd, SqlConnection conn, SqlTransaction trans, string cmdText, SqlParameter[] cmdParms)
         {
            if (conn.State != ConnectionState.Open)
            {
                conn.Open();
            }
            cmd.Connection = conn;
            cmd.CommandText = cmdText;
            if (trans != null)
            {
                cmd.Transaction = trans;
            }
            cmd.CommandType = CommandType.Text;
            if (cmdParms != null)
            {
                foreach (SqlParameter parameter in cmdParms)
                {
                    if (((parameter.Direction == ParameterDirection.InputOutput) || (parameter.Direction == ParameterDirection.Input)) && (parameter.Value == null))
                    {
                        parameter.Value = DBNull.Value;
                    }
                    cmd.Parameters.Add(parameter);
                }
            }
        }
        public static int ExecuteSql(string SQLString, SqlParameter[] cmdParms, string sqlconnection)
        {
            int num = 0;
            using (SqlConnection connection = new SqlConnection(sqlconnection))
            {
                using (SqlCommand command = new SqlCommand(SQLString, connection))
                {
                    try
                    {
                        if (cmdParms != null && cmdParms.Length > 0)
                        {
                            PrepareCommand(command, connection, null, SQLString, cmdParms);
                        }
                        num = command.ExecuteNonQuery();
                        command.Parameters.Clear();
                    }
                    finally
                    {
                        if (connection.State > ConnectionState.Closed)
                        {
                            command.Dispose();
                            connection.Dispose();
                            connection.Close();
                        }
                    }
                    return num;
                }
            }
        }
        public static int ExecuteSqlMany(string[] sqlArr, SqlParameter[][] parameters, string sqlconnection)
        {
            int num = 0;
            using (SqlConnection connection = new SqlConnection(sqlconnection))
            {
                connection.Open();
                using (SqlTransaction transaction = connection.BeginTransaction())
                {
                    using (SqlCommand command = new SqlCommand())
                    {
                        try
                        {
                            command.Connection = connection;
                            for (int i = 0; i < sqlArr.Length; i++)
                            {
                                string cmdText = sqlArr[i].ToString();
                                SqlParameter[] cmdParms = parameters[i];
                                if (cmdParms != null && cmdParms.Length > 0)
                                {
                                    PrepareCommand(command, connection, transaction, cmdText, cmdParms);
                                }
                                num+=command.ExecuteNonQuery();
                                command.Parameters.Clear();
                            }
                            transaction.Commit();
                        }
                        catch
                        {
                            transaction.Rollback();
                            throw;
                        }
                        finally
                        {
                            if (connection.State > ConnectionState.Closed)
                            {
                                command.Dispose();
                                connection.Dispose();
                                connection.Close();
                            }
                        }
                        return num;
                    }
                }
            }
        }
    }
}

把.cs文件打包成dll（打包可以用csc命令或者在vs上做一个项目生成）
我这里得到MSSQL.dll，先拖到db.py同级目录

db.py
用法参照最后的调用示例

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author: gz


import os
import sys
import clr

dll_path = os.path.dirname(__file__)
dll_name = 'MSSQL.dll'
sys.path.append(dll_path)
clr.FindAssembly(os.path.join(dll_path, dll_name))


def _convert_param(sql, param_dict):
    """
    参数为pymssql格式,将转换成c#格式
    若需要判定sql语句,可以正则表达式,写法如下
    # import re
    # all_params = re.findall(r'%\((.*?)\)s', sql, re.M | re.I)
    # d = {item: '@' + item for item in all_params}
    # sql_str = sql % d
    :param sql: 
    :param param_dict: 
    :return: 
    """
    from System.Data.SqlClient import SqlParameter
    sql_str = sql % {item: '@' + item for item in param_dict.keys()}
    param_arr = [SqlParameter("@" + k, str(v) if v else v) for k, v in param_dict.items()]
    return sql_str, param_arr


def exec_sql(sql, param_dict, conn_string):
    """
    执行sql语句
    若无需参数化，param_dict传None
    """
    from MSSQL import DBHelper
    sql_str, param_arr = _convert_param(sql, param_dict) if param_dict else (sql, None)
    rc = DBHelper.ExecuteSql(sql_str, param_arr, conn_string)
    return rc


def exec_sql_many(sql_param_list, conn_string):
    """
    执行多条sql语句
    使用事务
    :param sql_param_list: 
    :param conn_string: 
    :return: 
    """
    sql_arr = []
    param_arr_arr = []
    for sql, param_dict in sql_param_list:
        sql_str, param_arr = _convert_param(sql, param_dict) if param_dict else (sql, None)
        sql_arr.append(sql_str)
        param_arr_arr.append(param_arr)
    
    from MSSQL import DBHelper
    rc = DBHelper.ExecuteSqlMany(sql_arr, param_arr_arr, conn_string)
    return rc

最后的调用示例（真的是示例）

if __name__ == '__main__':
    sql = "INSERT INTO [dbo].[t_1021]([name],[age]) VALUES(%(name)s,%(age)s)"
    params = {'name': '小明', 'age': 20}
    conn_string = "server=localhost;uid=sa;password=123456;database=t-db-18;"
    
    from lib import db
    result = db.exec_sql(sql, params, conn_string)

    print(result)

没错，看起来像pymsql的cursor.execute(sql,params)，这样是为了方便我用原来的代码和sql语句

骚操作容易闪着腰。