SqlAlchemy使用教程(二) 入门示例及编程步骤-编程知识

SqlAlchemy使用教程(二) 入门示例及编程步骤

在这里插入图片描述

SqlAlchemy使用教程(一) 原理与环境搭建
SqlAlchemy使用教程(三) CoreAPI访问与操作数据库详解

二、入门示例与基本编程步骤

在第一章中提到，Sqlalchemy提供了两套方法来访问数据库，由于Sqlalchemy 官方文档结构有些乱，对于ORM的使用步骤的描述散布于各个章节，再加上SqlAlchemy2.x 与j1.x版本差异较大，很多介绍SqlAlchemy的文章上来就讲ORM，但示例时又使用CoreAPI，常令初学者困惑。本人建议先使用Core API来访问数据库，使用上更接近于 Sqlite3, Mysql-connector 等的方式，入门容易，而且也可以实现1套代码支持各类数据库。等熟悉CoreAPI的 MetaData， Sql Express Language以后，再学习使用ORM就会容易很多。
因此，本教程开头3章均以Core API方式为主, 文中示例均在Python3.10 + SQLAlchemy 2.0.23 版本上通过测试。

1、DB API访问数据库入门示例

示例功能：

建立数据库连接
通过Core API访问数据库（创建表，插入数据，查询数据）

1.1建立数据库连接

Step-1: 创建数据库引擎对象
DB Engine 是个全局变量，允许在其上建立多个connection访问数据库。

创建 DB Engine 实例的方法：
create_engine( db_url )
db_url参数在后面章节中详解介绍。本例使用sqlite3 内存数据库。

from sqlalchemy import create_engine
engine = create_engine("sqlite:///:memory:", echo=True)

Step-2 创建connect对象
connection 对象用于数据库操作。其支持context with语法

from sqlalchemy import text
with engine.connect() as conn:result = conn.execute(text("select 'hello world'"))print(result.all())

output

[('hello world',)]

Step-3 执行SQL Express 语句
text() 是SQL express 的最简单使用形式, 方便传值

创建1张表

conn.execute(text("CREATE TABLE some_table (x int, y int)"))

插入数据,

conn.execute(text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),[ { "x": 1, "y": 1}, {"x": 2, "y": 4 } ],)

SQL express传参语法：

参数占位使用　:x, :y ，参数名前加:分号
实际值用 [ dict, … ] 方式给出。

提交事务, 即将操作保存至数据库

conn.commit()

Step4 执行查询并获取结果

with engine.connect() as conn:result = conn.execute(text("SELECT x, y FROM some_table"))for row in result:print(f"x: {row.x}  y: {row.y}")

本例中，select x,y from some_table 将返回所有行

返回结果类型为 sqlalchemy.engine.cursor.CursorResult，是1个由 object 组成的可迭代对象。提供了多种方法访问结果数据：

fetchall(), fetchone(), fetchmany() ，使用tuple方式读取全部、单条、多条数据
all() 获取所有数据，返回列表
mappings(), 返回列表，元素为dict类型，
keys() 获取对象属性名（字段名）

还可以向查询语句传参：

result = conn.execute(text("SELECT x, y FROM some_table WHERE y > :y"), {"y": 2})

2、SqlAlchemy 异常处理

编写代码时1个好习惯：先写出异常与错误处理语句框架，再写正常流程部分，这样的习惯可以让代码更健壮，避免程序运行中断或出错。

虽然看似麻烦，但最终代码测试中遇到的问题更少，而且错误日志也更精准，问题定位效率更高，所以这样做将更省时间。

2.1 异常处理代码结构建议

from sqlalchemy import create_engine
from sqlalchemy.exc import IntegrityError, ProgrammingErrorengine = create_engine('mysql://username:password@localhost/mydatabase')try:# 执行数据库操作connection = engine.connect()# ...# 这里是可能引发异常的代码# ...connection.close()
except IntegrityError as e:# 处理唯一性约束违反等完整性错误print(f'Integrity error occurred: {str(e)}')
except ProgrammingError as e:# 处理SQL语法或参数错误print(f'Programming error occurred: {str(e)}')
except SQLAlchemyError as e:# 处理其他SQLAlchemy异常print(f'An error occurred: {str(e)}')