数据库访问

sqlalchemy

• 官方文档：docs.sqlalchemy.org.cn

定义数据模型

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)

连接数据库

from sqlalchemy import create_engine

# SQLite （生成一个本地文件 example.db）
engine = create_engine("sqlite:///example.db")

# MySQL （需安装 pymysql）
engine = create_engine("mysql+pymysql://user:password@localhost/mydatabase")

# PostgreSQL （需安装 psycopg2）
engine = create_engine("postgresql+psycopg2://user:password@localhost/mydatabase")


# 创建表（如果表不存在）
Base.metadata.create_all(engine)

CRUD

会话

from sqlalchemy.orm import sessionmaker

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 关闭会话
session.close()

# 上下文管理器：可以自动管理会话生命周期
with Session() as session:
    user = User(name="Bob", age=25)
    session.add(user)
    session.commit() 

新增数据

new_user = User(name="Alice", age=23)
session.add(new_user)
session.commit()

查询数据

# 查询所有用户
users = session.query(User).all()

# 条件查询
user = session.query(User).filter_by(name="Alice").first()

# 条件查询
user = session.query(User).filter(User.age > 25).all()

更新数据

user = session.query(User).filter_by(name="Alice").first()
user.age = 31
session.commit()

删除数据

user = session.query(User).filter_by(name="Alice").first()
session.delete(user)
session.commit()

回滚事务

try:
    # 执行操作
    session.commit()
except:
    session.rollback()  # 出错时回滚
    raise