数据存储 | MySQL —— ORM框架的使用（python：sqlalchemy库）

r囧r小猫 2022-11-26 08:52 230阅读 0赞

# **目录** #

1.参考博客

2.实战代码

--------------------

# 1.参考博客 #

> [https://www.cnblogs.com/wupeiqi/articles/5713330.html][https_www.cnblogs.com_wupeiqi_articles_5713330.html]

# 2.实战代码 #

>     # ----------------------------------------- ORM框架测试
>     from sqlalchemy.ext.declarative import declarative_base
>     from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
>     from sqlalchemy import create_engine
>     from sqlalchemy.orm import sessionmaker, relationship
>     from sqlalchemy import and_, or_
>     from sqlalchemy.sql import func
>     
>     
>     engine = create_engine("mysql+pymysql://maxin:111@192.168.116.133:3306/t1", max_overflow=5)
>     Base = declarative_base()
>     
>     
>     # - 定义表结构
>     class User(Base):
>         __tablename__ = 'users'
>         id = Column(Integer, primary_key=True, nullable=False, autoincrement=True)
>         name = Column(String(32), nullable=False, default='')
>         extra = Column(String(16), nullable=True, default='')
>         number = Column(Integer, nullable=False, default=0)
>     
>         __table_args__ = (
>             UniqueConstraint('id', 'name', name='uiq_id_name'),
>             Index('idx_name_extra', 'name', 'extra')
>         )
>     
>     
>     class Favor(Base):
>         __tablename__ = 'favors'
>         id = Column(Integer, primary_key=True, nullable=False, autoincrement=True)
>         name = Column(String(32), nullable=False, default='')
>         extra = Column(String(16), nullable=True, default='')
>         number = Column(Integer, nullable=False, default=0)
>     
>         __table_args__ = (
>             UniqueConstraint('id', 'name', name='uiq_id_name'),
>             Index('idx_name_extra', 'name', 'extra'),
>         )
>     
>     
>     def init_db():
>         Base.metadata.create_all(engine)
>     
>     
>     def drop_db():
>         Base.metadata.drop_all(engine)
>     
>     
>     # - 连接数据库
>     Session = sessionmaker(bind=engine)
>     sess = Session()
>     
>     
>     # - [建表]
>     drop_db()
>     init_db()
>     
>     
>     # ============================================== 增
>     obj = User(name="maxin0", extra="male", number=0)
>     
>     sess.add(obj)
>     sess.add_all([
>         User(name="maxin1", extra='male', number=1),
>         User(name="maxin2", extra='male', number=2),
>     ])
>     sess.commit()
>     
>     # ============================================== 查
>     # filter()：内部为表达式
>     # filter_by():内部传入的是等式类型的字典
>     # filter(,): 内部默认为and操作
>     res = sess.query(User).all()
>     res = sess.query(User.name, User.extra).all()
>     # - 条件查询
>     res = sess.query(User).filter_by(number=0).all()
>     res = sess.query(User.id, User.name, User.extra).filter(User.id > 2).all()
>     res = sess.query(User.id, User.name, User.extra).filter(User.id.between(2, 3), User.name == 'maxin1').all()
>     res = sess.query(User.id, User.name, User.extra).filter(User.id.in_([2, 3, 4]), User.name == 'maxin1').all()
>     res = sess.query(User.id, User.name, User.extra).filter(User.id.notin_([2, 3, 4]), User.name == 'maxin1').all()
>     res = sess.query(User.id, User.name, User.extra).filter(User.id.in_(sess.query(User.id).filter_by(name='maxin1'))).all()
>     res = sess.query(User.id, User.name, User.extra).filter(and_(User.id > 2, User.number < 2)).all()
>     res = sess.query(User.id, User.name, User.extra).filter(or_(User.id > 2, User.number < 2)).all()
>     res = sess.query(User.id, User.name, User.extra).filter(or_(User.id > 2, and_(User.id > 3, User.name == 'maxin1'))).all()
>     
>     # - 通配符查询
>     res = sess.query(User).filter(User.name.like('%maxin%')).all()
>     res = sess.query(User.name).filter(User.name.like('%maxin%')).all()
>     res = sess.query(User.name).filter(~User.name.like('%maxin1%')).all()
>     
>     # - 限制查询
>     res = sess.query(User)[0:2]
>     res = sess.query(User.id, User.name)[0:2]
>     res = sess.query(User).filter(User.id > 2).all()[0:1]
>     res = sess.query(User.id, User.name).filter(User.id > 2).all()[0:1]
>     
>     # - 排序查询
>     res = sess.query(User).order_by(User.number.desc()).all()
>     res = sess.query(User.id, User.name, User.number).order_by(User.number.desc()).all()
>     res = sess.query(User.id, User.name, User.number).filter(User.id > 1).order_by(User.number.desc()).all()
>     res = sess.query(User.id, User.name, User.number).order_by(User.number.desc(), User.id.asc()).all()
>     
>     # - 分组/聚合查询
>     res = sess.query(User.number).group_by(User.number).all()
>     res = sess.query(
>         User.number,
>         func.count(1),
>         func.sum(User.id),
>         func.max(User.id),
>         func.min(User.id),
>     ).group_by(User.number).all()
>     res = sess.query(
>         User.number,
>         func.count(1),
>         func.sum(User.id),
>         func.max(User.id),
>         func.min(User.id),
>     ).group_by(User.number).having(func.min(User.id) > 2).all()
>     
>     # - 连表查询
>     res = sess.query(User, Favor).all()  # 笛卡尔积
>     res = sess.query(User, Favor).filter(User.id == Favor.id).all()  # 内连接
>     res = sess.query(User).join(Favor).all()  # 内连接
>     res = sess.query(User).join(Favor, isouter=True).all()  # 左连接
>     res = sess.query(Favor).join(User, isouter=True).all()  # 右连接
>     
>     # - union组合
>     res1 = sess.query(User.name).filter(User.id < 4)
>     res2 = sess.query(User.name).filter(User.id < 4)
>     res3 = res1.union(res2).all()  # 去重组合
>     res4 = res1.union_all(res2).all()  # 不去重组合
>     print(res3)
>     print(res4)
>     
>     # - 嵌套查询
>     # - select * from (select * from A) as B;
>     res1 = sess.query(User).filter(User.id < 4).subquery()
>     res = sess.query(res1).all()
>     print(res)
>     # - select id, (select number from A where A.id=B.id) from B
>     res = sess.query(User.id, User.name, sess.query(Favor.number).filter(Favor.id==User.id).as_scalar()).all()
>     print(res)
>     # - select id from A where A.id in (select id from B)
>     res = sess.query(Favor.id).filter(Favor.id.in_(sess.query(User.id).subquery())).all()
>     print(res)
>     
>     # ============================================== 删[先查再删]
>     sess.query(User).filter(User.id < 2).delete()
>     sess.commit()
>     res = sess.query(User.id).all()
>     print(res)
>     
>     
>     # =============================================== 改[先查再改]
>     # - 统一更新
>     sess.query(User).filter(User.id < 4).update({'name': '你好'})
>     sess.query(User).filter(User.id < 4).update({User.name: 'haha'})
>     
>     # - 分别更新
>     # synchronize_session: 调整更新的类型
>     # synchronize_session=False：字符类型
>     # synchronize_session='evalute'：数字类型
>     sess.query(User).filter(User.id < 4).update({"name": User.name + 'x'}, synchronize_session=False)
>     sess.query(User).filter(User.id < 4).update({User.name: User.name + 'x'}, synchronize_session=False)
>     sess.query(User).filter(User.id < 4).update({"number": User.number + 1}, synchronize_session="evaluate")
>     sess.commit()
>     res = sess.query(User.id, User.name, User.number).all()
>     print(res)
>     
>     
>     # =============================================== 关系
>     # 1.导入必要的库文件
>     from sqlalchemy.orm import relationship
>     
>     # 2.在外键所在列中添加关系
>     class Person(Base):
>         __tablename__ = 'person'
>         nid = Column(Integer, primary_key=True)
>         nm = Column(String(32), index=True, nullable=True)
>         favor_id = Column(Integer, ForeignKey("favor.nid"))
>     
>         # relationship(): 创建一个关系
>         # 'Favor': 关系对应的类(表)
>         # 'backref=': 给外键参照的类(表)用的关系名，相当于Favor_content
>         Favor_content = relationship('Favor', backref='xxoo')
>     
>     # 3.实战代码
>     # 3.1.查询每个人的姓名及其对应的爱好名称(需要left join 连表)
>     # # 3.1.1.传统方法
>     # person_list = sess.query(Person.nm, Favor.name).join(Favor, isouter=True).all()
>     # for row in person_list:
>     #     print(row.nm, row.name)
>     # 3.1.1.关系方法
>     person_list = sess.query(Person)
>     for row in person_list:
>         print(person_list.nm, person_list.Favor_content.name)
>     # 3.2.查询每个爱好对应的人名
>     # # 3.2.1.传统方法
>     # favor_list = sess.query(Favor)
>     # for row in favor_list:
>     #     print(row.id,(sess.query(Favor).filter(Favor.id==row.id).all()))
>     # 3.2.2
>     favor_list = sess.query(Favor)
>     for row in favor_list:
>         print(row.id, row.xxoo.name)

[https_www.cnblogs.com_wupeiqi_articles_5713330.html]: https://www.cnblogs.com/wupeiqi/articles/5713330.html