python 連接 SQL Server 數據庫（使用 pyodbc 庫）

import pyodbc

# SERVER 是服務器名稱，DATABASE 是數據庫名稱，UID 是用戶名，PWD 是密碼
odbc = 'DRIVER={SQL Server};SERVER=DESKTOP-Q2JCMJD;DATABASE=WMS;UID=sa;PWD=******'


# 數據庫連接
def db_connect():
    # 打開數據庫連接
    db = pyodbc.connect(odbc)
    # 使用 cursor（）方法創建一個游標對象 cursor
    cursor = db.cursor()
    # 使用 execute（）方法執行 SQL 查詢
    cursor.execute('select @@version')
    # 使用 fetchone（）方法獲取單條數據
    data = cursor.fetchone()
    print(f'Database version: {data[0]}')
    # 關閉數據庫連接，此時自動關閉多有游標對象
    db.close()


# 創建數據庫表
def create_table():
    # 打開數據庫連接
    db = pyodbc.connect(odbc)
    # 使用 cursor（）方法創建一個游標對象 cursor
    cursor = db.cursor()
    # 使用 execute（）方法執行 SQL，如果表存在就刪除
    cursor.execute('drop table if exists teacher')
    # 使用預處理語句創建表
    sql = '''create table teacher(
        id int primary key,
        name varchar(20) not null,
        sex varchar(2) not null,
        age int not null)'''
    try:
        cursor.execute(sql)
        # 提交到數據庫執行
        db.commit()
        print('CREATE TABLE SUCCESS.')
    # 捕獲與數據庫相關的錯誤
    except pyodbc.Error as err:
        print(f'CREATE TABLE FAILED, CASE:{err}')
        # 如果發生錯誤就回滾
        db.rollback()
    finally:
        # 關閉數據庫連接
        db.close()


# 刪除數據庫表
def drop_table():
    # 打開數據庫連接
    db = pyodbc.connect(odbc)
    # 使用 cursor（）方法創建一個游標對象 cursor
    cursor = db.cursor()
    # 使用預處理語句刪除表
    sql = 'drop table teacher'
    try:
        cursor.execute(sql)
        # 提交到數據庫執行
        db.commit()
        print('DROP TABLE SUCCESS.')
    # 捕獲與數據庫相關的錯誤
    except pyodbc.Error as err:
        print(f'DROP TABLE FAILED, CASE:{err}')
        # 如果發生錯誤就回滾
        db.rollback()
    finally:
        # 關閉數據庫連接
        db.close()


# 數據庫插入
def insert_record():
    # 打開數據庫連接
    db = pyodbc.connect(odbc)
    # 使用 cursor（）方法創建一個游標對象 cursor
    cursor = db.cursor()
    # SQL 插入語句
    sql = "insert information values(1001, '小明', '男', 18)"
    try:
        # 執行 SQL 語句
        cursor.execute(sql)
        # 提交到數據庫執行
        db.commit()
        print('INSERT SUCCESS.')
    # 捕獲與數據庫相關的錯誤
    except pyodbc.Error as err:
        print(f'INSERT FAILED, CASE:{err}')
        # 如果發生錯誤就回滾
        db.rollback()
    finally:
        # 關閉數據庫連接
        db.close()


# 數據庫刪除
def delete_record():
    # 打開數據庫連接
    db = pyodbc.connect(odbc)
    # 使用 cursor（）方法創建一個游標對象 cursor
    cursor = db.cursor()
    # SQL 刪除語句
    sql = "delete from information where sex='女'"
    try:
        # 執行 SQL 語句
        cursor.execute(sql)
        # 提交到數據庫執行
        db.commit()
        print('DELETE SUCCESS.')
    # 捕獲與數據庫相關的錯誤
    except pyodbc.Error as err:
        print(f'DELETE FAILED, CASE:{err}')
        # 如果發生錯誤就回滾
        db.rollback()
    finally:
        # 關閉數據庫連接
        db.close()


# 數據庫更新
def update_record():
    # 打開數據庫連接
    db = pyodbc.connect(odbc)
    # 使用 cursor（）方法創建一個游標對象 cursor
    cursor = db.cursor()
    # SQL 更新語句
    sql = "update information set age=25 where name='小明'"
    try:
        # 執行 SQL 語句
        cursor.execute(sql)
        # 提交到數據庫執行
        db.commit()
        print('UPDATE SUCCESS.')
    # 捕獲與數據庫相關的錯誤
    except pyodbc.Error as err:
        print(f'UPDATE FAILED, CASE:{err}')
        # 如果發生錯誤就回滾
        db.rollback()
    finally:
        # 關閉數據庫連接
        db.close()


# 數據庫查詢
def query_record():
    # 打開數據庫連接
    db = pyodbc.connect(odbc)
    # 使用 cursor（）方法創建一個游標對象 cursor
    cursor = db.cursor()
    # SQL 查詢語句
    sql = 'select * from information'
    try:
        # 執行 SQL 語句
        cursor.execute(sql)
        '''
        fetchone()：該方法獲取下一個查詢結果集，結果集是一個對象。
        fetchall()：接收全部返回結果行。
        rowcount：返回執行 execute（）方法后影響的行數
        '''
        # 獲取所有記錄列表
        results = cursor.fetchall()
        for row in results:
            num = row[0]
            name = row[1]
            sex = row[2]
            age = row[3]
            # 輸出結果
            print(f'id:{num}, name:{name}, sex:{sex}, age:{age}')
    # 捕獲與數據庫相關的錯誤
    except pyodbc.Error as err:
        print(f'QUERY MySQL table FAILED, CASE:{err}')
    finally:
        # 關閉數據庫連接
        db.close()


if __name__ == '__main__':
    db_connect()