网站备案信息批量查询,app开发培训课程,沈阳工务轨道建设网站,搭建个网站需要多少钱上次发布《多种方式访问mysql的对比分析》一文后#xff0c;有读者留言#xff0c;说SQLAlchemy的使用方法没讲清楚#xff0c;只有一段简短的介绍#xff0c;演示代码也比较模糊#xff0c;SQLAlchemy在实际项目运用非常广泛#xff0c;由于其支持 ORM 模型#xff0c;…上次发布《多种方式访问mysql的对比分析》一文后有读者留言说SQLAlchemy的使用方法没讲清楚只有一段简短的介绍演示代码也比较模糊SQLAlchemy在实际项目运用非常广泛由于其支持 ORM 模型能够将表映射为类让你用Python类的方式来操作数据库而不需要直接写SQL语句是面向对象访问数据库不可或缺的一个库根据需求设计数据库结构和操作逻辑可以大大提高开发效率和代码可维护性。今天风云将该库的详细使用方法重新整理了发出来供大家参考欢迎留言讨论。
为什么还要用SQLAlchemy呢原因有几个 1.简洁SQLAlchemy让你用Python代码来操作数据库代码更简洁也更容易理解。 2.安全它能帮助你避免SQL注入等安全问题让你的数据库更安全。 3.灵活SQLAlchemy支持多种数据库比如MySQL、PostgreSQL、SQLite等你只需要改一下配置就能在不同的数据库之间切换。
SQLAlchemy 包括两个核心组件
SQLAlchemy Core提供底层 SQL 构造和数据库连接。SQLAlchemy ORM实现面向对象的 ORM 映射。
1. 安装 SQLAlchemy
使用 pip 安装
pip install sqlalchemy
2. 数据库连接
使用 create_engine 创建数据库连接
from sqlalchemy import create_engine
# 替换为实际的数据库 URL
engine create_engine(sqlite:///example.db) # SQLite 示例
# MySQL 示例: mysqlpymysql://user:passwordhost/dbname
3. 使用 SQLAlchemy Core
3.1 创建表
使用 MetaData 和 Table 定义表结构
from sqlalchemy import Table, Column, Integer, String, MetaData,create_engineengine create_engine(sqlite:///example.db) # SQLite 示例数据库
metadata MetaData() # 元数据对象users Table( # 创建表users, metadata, # 表名Column(id, Integer, primary_keyTrue), # id字段Column(name, String(50)), # name字段Column(age, Integer) # age字段
)metadata.create_all(engine) # 创建表
3.2 插入数据
使用 insert() 插入数据
from sqlalchemy import insertwith engine.connect() as conn: # 连接数据库stmt insert(users).values(nameAlice, age30) # 插入数据conn.execute(stmt) # 执行插入语句
3.3 查询数据
使用 select() 查询
from sqlalchemy import selectwith engine.connect() as conn: # 连接数据库stmt select(users) # 查询数据result conn.execute(stmt) # 执行查询语句for row in result: # 遍历结果print(row)
4. 使用 SQLAlchemy ORM
4.1 定义模型
使用 ORM 的 declarative_base 定义模型
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, StringBase declarative_base() # 创建基类class User(Base): # 创建User类__tablename__ users # 表名id Column(Integer, primary_keyTrue) # id字段name Column(String(50)) # name字段age Column(Integer) # age字段
4.2 创建表
创建表结构
Base.metadata.create_all(engine)
4.3 创建会话
使用 sessionmaker 创建会话
from sqlalchemy.orm import sessionmakerSession sessionmaker(bindengine) # 创建Session会话类
session Session()
4.4 增删改查
插入数据
new_user User(nameBob, age25) # 创建新用户session.add(new_user)# 添加新用户到sessionsession.commit() # 提交事务
查询数据
users session.query(User).all() # 查询所有用户
for user in users: # 遍历查询结果print(user.name, user.age)
更新数据
user session.query(User).filter_by(nameBob).first() # 查询用户
user.age 26session.commit() # 提交事务
删除数据
user session.query(User).filter_by(nameBob).first() # 查询具体用户定位到第一条记录session.delete(user) # 删除用户
session.commit() # 提交事务
5. 使用关系映射
定义一对多关系
from sqlalchemy import ForeignKeyfrom sqlalchemy.orm import relationshipclass Post(Base):__tablename__ posts # 表名id Column(Integer, primary_keyTrue) # id字段title Column(String(100)) # title字段user_id Column(Integer, ForeignKey(users.id)) # 定义外键user relationship(User, back_populatesposts) # 关联关系User.posts relationship(Post, order_byPost.id, back_populatesuser) # 关联关系# 插入关联数据
user User(nameCharlie)post1 Post(titlePost 1, useruser)post2 Post(titlePost 2, useruser)session.add(user) # 添加关联数据session.add_all([post1, post2]) # 添加所有关联session.commit() # 提交事务#查询关联数据
user session.query(User).filter_by(nameCharlie).first() # 查询关联数据for post in user.posts: # 遍历关联数据print(post.title)
6. 使用事务
手动管理事务
from sqlalchemy.exc import SQLAlchemyErrortry:with session.begin(): # 开启事务user User(nameDave, age40) # 创建新用户session.add(user) # 添加新用户到session
except SQLAlchemyError as e: # 如果发生异常回滚事务print(fTransaction failed: {e})session.rollback() # 回滚事务
7. 完整的封装类
最后风云按惯例将此使用封装为一个完整的类大家有需要有的自取
from sqlalchemy import create_engine, Column, Integer, String, exc
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmakerBase declarative_base() # 创建基类# 定义表模型
class User(Base):__tablename__ users # 表名id Column(Integer, primary_keyTrue, autoincrementTrue) # id字段name Column(String(50), nullableFalse) # name字段age Column(Integer, nullableFalse) # age字段class MySQLHandler:def __init__(self, user, password, host, database):初始化数据库连接和会话try:url fmysqlpymysql://{user}:{password}{host}/{database} # 创建数据库连接self.engine create_engine(url, echoFalse) # 数据库引擎Base.metadata.create_all(self.engine) # 创建表self.Session sessionmaker(bindself.engine) # 创建会话except exc.SQLAlchemyError as e:print(fError initializing database: {e})def add_user(self, name, age):添加用户session self.Session() try:new_user User(namename, ageage) # 创建新用户session.add(new_user) # 添加新用户到sessionsession.commit() # 提交事务print(fUser {name} added successfully!)except exc.SQLAlchemyError as e: # 如果发生异常回滚事务session.rollback() # 执行回滚print(fError adding user: {e})finally:session.close() # 关闭会话def get_users(self):查询所有用户session self.Session()try:users session.query(User).all() # 查询所有用户return [{id: user.id, name: user.name, age: user.age} for user in users]except exc.SQLAlchemyError as e: # 如果发生异常返回空print(fError fetching users: {e})return []finally:session.close()def update_user(self, user_id, nameNone, ageNone):更新用户信息session self.Session()try:user session.query(User).filter_by(iduser_id).first() # 查询用户if not user: # 如果用户不存在返回print(fUser with ID {user_id} not found.)returnif name:user.name nameif age:user.age agesession.commit() # 提交事务print(fUser {user_id} updated successfully!)except exc.SQLAlchemyError as e: # 如果发生异常回滚事务session.rollback()print(fError updating user: {e})finally:session.close()def delete_user(self, user_id):删除用户session self.Session()try:user session.query(User).filter_by(iduser_id).first() # 查询用户if not user: # 如果用户不存在返回print(fUser with ID {user_id} not found.)returnsession.delete(user) # 删除用户session.commit() # 提交事务print(fUser {user_id} deleted successfully!)except exc.SQLAlchemyError as e: # 如果发生异常回滚事务session.rollback()print(fError deleting user: {e})finally:session.close()def transaction_example(self):事务操作示例session self.Session()try:user1 User(nameAlice, age30) # 创建2个新用户user2 User(nameBob, age25) # 添加多个用户session.add(user1) # 添加新用户到sessionsession.add(user2)# 模拟事务中的错误# Uncomment the line below to raise an exception and trigger a rollback# raise ValueError(Simulated error!)session.commit() # 提交事务print(Transaction completed successfully!)except (exc.SQLAlchemyError, ValueError) as e: # 如果发生异常回滚事务session.rollback()print(fTransaction failed: {e})finally:session.close()# 使用示例
if __name__ __main__:# 初始化数据库连接db_handler MySQLHandler(userroot, passwordpassword123, hostlocalhost, databasetest_db)# 添加用户db_handler.add_user(nameJohn Doe, age28)# 查询用户users db_handler.get_users()print(Users:, users)# 更新用户db_handler.update_user(user_id1, nameJane Doe, age32)# 删除用户db_handler.delete_user(user_id1)# 事务示例db_handler.transaction_example()
