Python 连接到 Microsoft SQL 数据库

模块

import pyodbc

class SQLServerOperations:
    def __init__(self, server, database, uid, pwd):
        self.server = server
        self.database = database
        self.uid = uid
        self.pwd = pwd
        self.connection = None

    def connect(self):
        try:
            self.connection = pyodbc.connect(f'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={self.server};DATABASE={self.database};UID={self.uid};PWD={self.pwd}')
            print(f"连接数据库成功!")
        except pyodbc.Error as e:
            print(f"连接数据库时出错: {e}")

    def execute_query(self, query):
        if not self.connection:
            self.connect()
        cursor = self.connection.cursor()
        try:
            cursor.execute(query)
            # self.connection.commit()
            return cursor.fetchall()
        except pyodbc.Error as e:
            print(f"执行查询时出错: {e}")
        finally:
            cursor.close()

    def insert_data(self, table_name, data_dict):
        """
        插入数据到指定表

        参数:
        table_name (str): 要插入数据的表名
        data_dict (dict): 包含列名和对应值的字典

        示例:
        data = {'column1': 'value1', 'column2': 'value2'}
        insert_data('your_table', data)
        """
        if not self.connection:
            self.connect()
        columns = ', '.join(data_dict.keys())
        values = ', '.join(['?' for _ in data_dict])
        query = f"INSERT INTO {table_name} ({columns}) VALUES ({values})"
        cursor = self.connection.cursor()
        try:
            cursor.execute(query, list(data_dict.values()))
            self.connection.commit()
        except pyodbc.Error as e:
            print(f"插入数据时出错: {e}")
        finally:
            cursor.close()

    def update_data(self, table_name, condition, data_dict):
        """
        更新指定表中的数据

        参数:
        table_name (str): 要更新数据的表名
        condition (str): 更新的条件
        data_dict (dict): 包含要更新的列名和对应新值的字典

        示例:
        condition = "id = 1"
        data = {'column1': 'new_value1', 'column2': 'new_value2'}
        update_data('your_table', condition, data)
        """
        if not self.connection:
            self.connect()
        set_values = ', '.join([f"{column} =?" for column in data_dict])
        query = f"UPDATE {table_name} SET {set_values} WHERE {condition}"
        cursor = self.connection.cursor()
        try:
            params = list(data_dict.values())
            cursor.execute(query, params)
            self.connection.commit()
        except pyodbc.Error as e:
            print(f"更新数据时出错: {e}")
        finally:
            cursor.close()

    def close_connection(self):
        if self.connection:
            self.connection.close()
​

示例

from sql_server_operations import SQLServerOperations
# import logging

# 创建操作对象
sql_ops = SQLServerOperations('127.0.0.1', 'databasename','user', 'password')

# 连接数据库
sql_ops.connect()

# 执行查询
query = "SELECT * FROM your_table WHERE column_name = value"  # 替换 your_table 为实际表名，column_name 和 value 为实际的列名和值
# logging.info(f"Executing query: {query}")  # 记录正在执行的查询
results = sql_ops.execute_query(query)
if results is None:
    logging.error("Query returned no results or an error occurred.")  # 记录查询没有结果或出错
else:
    for row in results:
        print(row)

# 插入数据示例
data_to_insert = {'column1': 'value1', 'column2': 'value2'}
sql_ops.insert_data('your_table', data_to_insert)

# 更新数据示例
condition = "id = 1"data_to_update = {'column1': 'new_value1', 'column2': 'new_value2'}
sql_ops.update_data('your_table', condition, data_to_update)

# 关闭连接
sql_ops.close_connection()
​

‍