内蒙古交通建设质监局网站百度推广四川成都地区服务中心
- 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)}')
2.2 SQLAlchemy常用的内置异常类
- sqlalchemy.exc.SQLAlchemyError：所有SQLAlchemy异常的基类
- sqlalchemy.exc.InvalidRequestError：无效的请求异常，包括无效的查询或表达式
- sqlalchemy.exc.StatementError：执行SQL语句时出错的异常
- sqlalchemy.exc.IntegrityError：完整性约束错误，例如唯一性约束或外键约束违反等
- sqlalchemy.exc.OperationalError：操作数据库时出错的异常
- sqlalchemy.exc.ProgrammingError：编程错误，例如错误的SQL语法或参数错误等