sqlalchemy的基本使用

依赖

pip install sqlalchemy

基本使用

from sqlalchemy.orm import Session, sessionmaker, declarative_base
from sqlalchemy import create_engine, Column, Integer, String

# 1.创建引擎
engine = create_engine('mysql+pymysql://root:[email protected]/sqlalchemy_test')

# 2.创建会话
# 2.1 基本创建方式
# s = Session(bind=engine)

# 2.2 通过sessionmaker创建，本质上还是Session，只是可自定义配置强
# Session = sessionmaker(bind=engine)
# s = Session()

# 2.3 通过scoped_session创建, 传入一个session工厂
Session =scoped_session(sessionmaker(bind=engine))
s = Session()


# 3.创建ORM映射基类
Base = declarative_base()


# 4.创建数据表
class User(Base):
    __tablename__ = 'user_account'
    id = Column(Integer, primary_key=True)
    name = Column(String(25))

    def __repr__(self):
        return f'User(id={
      self.id}, name={
      self.name})'


# 5.不存在则创建，存在则不创建
Base.metadata.create_all(engine)

# 6.实例化一个Use对象
# user = User(name='chatu')

# 7.将User对象添加到Session中
s.add(user)
# 8.提交到数据库
s.commit()
# 9.关闭session对象
s.close()

更多数据库操作

from add import s, User, engine  # 这边导入的是上方定义的
from sqlalchemy import select, table, update, insert, delete, create_engine, column

# result = s.query(User).all()
# print(result)

# s.scalars()执行语句，并返回一个个实例对象
# 可以通过select这种原生的，然后去执行
# 也可以通过session.query去筛选
# s.execute()可以执行增删改查


# ##### 接近原生的sqlalchemy查询
print(type(select(User)))
# 查找
sel = select(User).where(User.name == 'chatu1')
print(type(select(User).where(User.name == 'chatu1')))
result = s.execute(sel)
print(result, '=============', type(result))
for i in result:
    print(i)

print('---------------------',s.scalars(sel))
for i in s.scalars(sel):
    print(i)

# 更新
upd = (update(User).
where(User.id == 4).
values(name='chst1122')
)

# 直接通过engine创建的连接则在执行execute的时候就会提交到数据库
# conn = engine.connect()
# conn.execute(upd)

# 通过session执行execute
a = s.execute(upd)  # 执行更新
s.commit()  # 只有提交之后数据库才会更新

# 插入
ins = insert(User).values(name='hello world')
s.execute((ins))
s.commit()

# 删除
delete_data = delete(User).where(column('id') >= 100)
s.execute(delete_data)
s.commit()


###### 使用会话机制查询

# 查询
filter_query = s.query(User).filter(User.name=='chatu1').all()  # 返回一个列表，包含所有筛选对象
# filter_query = s.query(User).filter(User.name.in_(['chatu2'])).all() # 返回一个列表，包含所有筛选对象
print(filter_query)

# 修改
update_filter = s.query(User).filter(User.name=='chatu1').update({
    'name':'chatu11'})
s.commit()

# 添加
user = User(name='hello chatu')
s.add(user)
s.commit()

# 删除
s.query(User).filter(User.name=='chst1122').delete()
s.commit()

附录

1.链接：Session、sessionmaker、scoped_session相关资料
2.接近原生的sqlalchemy操作
链接-查询：sqlalchemy-查询
链接-更新，插入，删除：sqlalchemy-更新，插入，删除
3.会话通过query操作数据库的操作
会话的query查询：query操作API