Python数据库编程（以pyodbc和SQLAlchemy为例）

Posted 2022-06-08 詩和遠方

概述

本文以pyodbc和sqlalchemy操作SQL Server为例说明python的数据库编程。

pyodbc

pyodbc只是众多python database driver（适配器）之一，其它诸如turbodbc，以及适用于PostgreSQL的psycopg2等也有大量的使用。
但不管哪种driver均在一定程度上遵循python的DB-API标准，所以编程模式还是挺一致的。

连接

import pyodbc
# 显式指定连接属性
conn = pyodbc.connect('DRIVER=ODBC Driver 13 for SQL Server;SERVER=localhost;DATABASE=TestDB;UID=sa;PWD=password')
# 用 ODBC的 DSN (data source name)
conn = pyodbc.connect('DSN=DSN_Name;UID=sa;PWD=password',autocommit=True)
# Windows认证方式
conn = pyodbc.connect('DSN=DSN_Name;Trusted_Connection=yes',autocommit=True)

查询与数据获取

• 建表、插入数据

# 声明游标
cur = conn.cursor()
# 建表
cur.execute('DROP TABLE IF EXISTS dbo.Product')
cur.execute('CREATE TABLE dbo.Product(SKU INT,ProductName NVARCHAR(200))')
# 插入数据
cur.execute("""
    INSERT INTO dbo.Product(SKU,ProductName) VALUES
        (1,'React-270'),
        (2,'Element-55'),
        (3,'BRA')
"""
)

cur.execute('Select Top 1 SKU,ProductName From dbo.Product')
result = cur.fetchone()
print(result)

(1, 'React-270')

• 获取字段数据

# 下标形式
print(result[0],result[1])
# 字段名形式
print(result.SKU,result.ProductName)

1 React-270
1 React-270

• 一次获取一行或多行记录

for row in cur.execute('Select SKU,ProductName From dbo.Product'):
    print(row)

(1, 'React 270')
(2, 'Element 55')
(3, 'BRA')

rows = cur.execute('Select SKU,ProductName From dbo.Product').fetchall()
for row in rows:
    print(row)

(1, 'React-270')
(2, 'Element-55')
(3, 'BRA')

• 参数化查询

result = cur.execute('Select ProductName From dbo.Product Where SKU = ?',1).fetchone()[0]
print(result)

React-270

• with风格的上下文管理器

with conn.cursor() as cur:
    cur.execute("Update dbo.Product Set ProductName = 'ELE 56' Where SKU = 3")
    cur.execute("Delete dbo.Product Where SKU = 2")

以上with语句内操作，执行完毕后会自动commit，但若有异常不会rollback
更多用法与提示，参考官方github主页

sqlalchemy

sqlalchemy功能非常强大，其在数据库驱动的基础上完成了更高级的抽象，支持对象化操作，相当于将数据库里的一行当做对象的一个实例。

但本文仅关注更加SQL化的操作。

连接字符串

模板：dialect+driver://username:password@host:port/database

dialect指具体的SQL方言，比如mssql、mysql、postgresql等

from sqlalchemy import create_engine
# DSN
con_string_dsn = "mssql+pyodbc://sa:password@DSN_Name"
# 显式指定连接属性
con_string_1 = "mssql+pyodbc://sa:password@localhost/TestDB?driver=ODBC+Driver+13+for+SQL+Server"
# 忽略用户名和密码，则为 windows 验证
con_string_2 = "mssql+pyodbc://localhost/TestDB?driver=ODBC+Driver+13+for+SQL+Server"

查询与客户端获取

# 引擎，可直接运行SQL
engine = create_engine(con_string_2)
result = engine.execute('Select Top 1 ProductName From dbo.Product').fetchall()
# 连接
conn = engine.connect()
result = conn.execute('Select Top 1 ProductName From dbo.Product').fetchall()

with语法

with engine.begin() as conn:
    conn.execute("Update dbo.Product Set ProductName = 'ELE 98' Where SKU = 3")
    conn.execute("Delete dbo.Product Where SKU = 2")

以上操作将自动commit，若遇到异常则会自动rollback，非常方便执行DML操作

使用原生driver

若高层抽象方法不包括某些数据库特有的方法（如并不是所有驱动都支持的存储过程调用方法callproc），则可使用原生driver

conn = engine.raw_connection()
cur = conn.cursor()
cur.callproc('ProcName',(parm1,parm2))

更多详情详见官网