python3 数据库操作 orm sqlalchemy 简介

妖狐艹你老母 2021-07-28 17:11 436阅读 0赞

ORM 全称 Object Relational Mapping, 翻译过来叫对象关系映射。简单的说，ORM 将数据库中的表与面向对象语言中的类建立了一种对应关系。这样，我们要操作数据库，数据库中的表或者表中的一条记录就可以直接通过操作类或者类实例来完成。

SQLAlchemy 是Python 社区最知名的 ORM 工具之一，为高效和高性能的数据库访问设计，实现了完整的企业级持久模型。

1.创建数据表

# ORM中的数据表是什么呢?
    # Object Relation Mapping
    # Object - Table 通过 Object 去操纵数据表
    # 从而引出了我们的第一步创建数据表 - 创建Object
    # 1. 创建Object
    # class User(object):
    #     pass
    
    # 2. 让Object与数据表产生某种关系 也就是让Object与数据表格式极度相似
    # 导入官宣基础模型（SQLAlchemy）
    from sqlalchemy.ext.declarative import declarative_base
    # 实例化官宣模型 - Base 就是 ORM 模型
    Base = declarative_base()
    # 当前的这个Object继承了Base也就是代表了Object继承了ORM的模型
    class User(Base):  # 相当于 Django Models中的 Model
        # 为Table创建名称
        __tablename__ = "user"
        # 创建ID数据字段 , 那么ID是不是一个数据列呢? 也就是说创建ID字段 == 创建ID数据列
        from sqlalchemy import Column,Integer,String
        # id = Column(数据类型,索引,主键,外键,等等)
        # int == Integer
        id = Column(Integer, primary_key=True, autoincrement=True)
        # str == char(长度) == String(长度)
        name = Column(String(32),index=True)
    
    # 3.去数据库中创建数据表? or 先连接数据库?
    # 3.去连接数据库 创建数据引擎
    from sqlalchemy import create_engine
    # 创建的数据库引擎
    engine = create_engine("mysql+pymysql://root:pwd@127.0.0.1:3306/dragon(表名）?charset=utf8")
    
    # Base 自动检索所有继承Base的ORM 对象 并且创建所有的数据表
    Base.metadata.create_all(engine)

2.增删改查

增加数据

#insert 为数据表增加数据
    # insert One 增加一行数据
    # insert into user(name) values ("wang")
    # 在ORM中的操作:
    # 1.首先导入之间做好的ORM 对象 User
    from my_create_table import User
    # 2.使用Users ORM模型创建一条数据
    user1 = User(name="wang")
    # 数据已经创建完了,但是需要写入到数据库中啊,怎么写入呢?
    # 3.写入数据库:
    # 首先打开数据库会话 , 说白了就是创建了一个操纵数据库的窗口
    # 导入 sqlalchemy.orm 中的 sessionmaker
    from sqlalchemy.orm import sessionmaker
    # 导入之前创建好的 create_engine
    from my_create_table import engine
    # 创建 sessionmaker 会话对象,将数据库引擎 engine 交给 sessionmaker
    Session = sessionmaker(engine)
    # 打开会话对象 Session
    db_session = Session()
    # 在db_session会话中添加一条 UserORM模型创建的数据
    db_session.add(user1)
    # 使用 db_session 会话提交 , 这里的提交是指将db_session中的所有指令一次性提交
    db_session.commit()
    
    # 当然也你也可很任性的提交多条数据
    # 方法一:
    user2 = User(name="张三")
    user3 = User(name="王五")
    db_session.add(user2)
    db_session.add(user3)
    db_session.commit()
    # 之前说过commit是将db_session中的所有指令一次性提交,现在的db_session中至少有两条指令user2和user3
    db_session.close()
    #关闭会话
    
    # 方法二:
    user_list = [
        User(name="wang1"),
        User(name="wang2"),
        User(name="wang3")
    ]
    db_session.add_all(user_list)
    db_session.commit()
    
    db_session.close()
    
    orm_insert

查询数据

# ORM操作查询数据
    # 有了刚才Insert增加数据的经验,那么查询之前的准备工作,就不用再重复了吧
    # 回想一下刚才Insert时我们的操作
    from my_create_table import User, engine
    from sqlalchemy.orm import sessionmaker
    
    Session = sessionmaker(engine)
    db_session = Session()
    
    # 1. select * from user 查询user表中的所有数据
    # 语法是这样的 使用 db_session 会话 执行User表 query(User) 取出全部数据 all()
    user_all_list = db_session.query(User).all()
    print(user_all_list)  # [<my_create_table.User object at 0x0000016D7C4BCDD8>]
    # 如何查看user_all_list其中的数据呢? 循环呗
    for i in user_all_list:
        print(i.id, i.name)  # ORM对象 直接使用调用属性的方法 拿出对应字段的值
    
    db_session.close()
    #关闭会话
    
    # 2. select * from user where id >= 20
    # 语法是这样的 使用 db_session 会话 执行User表 query(User) 筛选内容User.id >=20 的数据全部取出 all()
    user_all_list = db_session.query(User).filter(User.id >= 20).all()
    print(user_all_list)
    for i in user_all_list:
        print(i.id, i.name)
    
    db_session.close()
    #关闭会话
    
    # 3. 除了取出全部还可以只取出一条
    user = db_session.query(User).filter(User.id >= 20).first()
    print(user.id, user.name)
    db_session.close()
    #关闭会话
    
    # 4. 查看sql语句
    wulong1 = db_session.query(User).filter(User.id >= 20)
    print(wulong1)
    #SELECT user.id AS user_id, user.name AS user_name
    #FROM user
    #WHERE user.id >= %(id_1)s
    wulong2 = db_session.query(User)
    print(wulong2)
    #SELECT user.id AS user_id, user.name AS user_name
    #FROM user
    db_session.close()
    #关闭会话

修改数据

# ORM更新数据
    # 根据之前原有的经验,接下来是不是要导入ORM对象了,是不是要创建db_session会话了
    from my_create_table import User,engine
    from sqlalchemy.orm import sessionmaker
    Session = sessionmaker(engine)
    db_session = Session()
    
    # UPDATE user SET name="NBDragon" WHERE id=20 更新一条数据
    # 语法是这样的 :
    # 使用 db_session 执行User表 query(User) 筛选 User.id = 20 的数据 filter(User.id == 20)
    # 将name字段的值改为NBDragon update({"name":"NBDragon"})
    res = db_session.query(User).filter(User.id == 20).update({"name":"NBDragon"})
    print(res) # 1 res就是我们当前这句更新语句所更新的行数
    # 注意注意注意
    # 这里一定要将db_session中的执行语句进行提交,因为你这是要对数据中的数据进行操作
    # 数据库中 增 改 删 都是操作,也就是说执行以上三种操作的时候一定要commit
    db_session.commit()
    db_session.close()
    #关闭会话
    
    # 更新多条
    res = db_session.query(User).filter(User.id <= 20).update({"name":"NBDragon"})
    print(res) # 6 res就是我们当前这句更新语句所更新的行数
    db_session.commit()
    db_session.close()
    #关闭会话

删除数据

# ORM 删除一条多条数据
    # 导入 ORM 创建会话
    from my_create_table import User,engine
    from sqlalchemy.orm import sessionmaker
    Session = sessionmaker(engine)
    db_session = Session()
    
    # DELETE FROM `user` WHERE id=20
    res = db_session.query(User).filter(User.id==20).delete()
    print(res)
    db_session.commit()
    
    db_session.close()
    #关闭会话

高级版查询操作

from my_create_table import User,engine
    from sqlalchemy.orm import sessionmaker
    
    Session = sessionmaker(engine)
    db_session = Session()
    
    # 查询数据表操作
    # and or
    from sqlalchemy.sql import and_ , or_
    ret = db_session.query(User).filter(and_(User.id > 3, User.name == 'wang')).all()
    ret = db_session.query(User).filter(or_(User.id < 2, User.name == 'wang')).all()
    
    # 查询所有数据
    r1 = db_session.query(User).all()
    
    # 查询数据 指定查询数据列 加入别名
    r2 = db_session.query(User.name.label('username'), User.id).first()
    print(r2.id,r2.username) # 15 NBDragon
    
    # 表达式筛选条件
    r3 = db_session.query(User).filter(User.name == "wang").all()
    
    # 原生SQL筛选条件
    r4 = db_session.query(User).filter_by(name='wang').all()
    r5 = db_session.query(User).filter_by(name='wang').first()
    
    # 字符串匹配方式筛选条件 并使用 order_by进行排序
    r6 = db_session.query(User).filter(text("id<:value and name=:name")).params(value=224, name='wang').order_by(User.id).all()
    
    #原生SQL查询
    r7 = db_session.query(User).from_statement(text("SELECT * FROM User where name=:name")).params(name='wang').all()
    
    # 筛选查询列
    # query的时候我们不在使用User ORM对象,而是使用User.name来对内容进行选取
    user_list = db_session.query(User.name).all()
    print(user_list)
    for row in user_list:
        print(row.name)
    
    # 别名映射  name as nick
    user_list = db_session.query(User.name.label("nick")).all()
    print(user_list)
    for row in user_list:
        print(row.nick) # 这里要写别名了
    
    # 筛选条件格式
    user_list = db_session.query(User).filter(User.name == "wang").all()
    user_list = db_session.query(User).filter(User.name == "wang").first()
    user_list = db_session.query(User).filter_by(name="wang").first()
    for row in user_list:
        print(row.nick)
    
    # 复杂查询
    from sqlalchemy.sql import text
    user_list = db_session.query(User).filter(text("id<:value and name=:name")).params(value=3,name="wang")
    
    # 查询语句
    from sqlalchemy.sql import text
    user_list = db_session.query(User).filter(text("select * from User id<:value and name=:name")).params(value=3,name="wang")
    
    # 排序 :
    user_list = db_session.query(User).order_by(User.id).all()
    user_list = db_session.query(User).order_by(User.id.desc()).all()
    for row in user_list:
        print(row.name,row.id)
    
    #其他查询条件
    """
    ret = session.query(User).filter_by(name='wang').all()
    ret = session.query(User).filter(User.id > 1, User.name == 'wang').all()
    ret = session.query(User).filter(User.id.between(1, 3), User.name == 'wang').all() # between 大于1小于3的
    ret = session.query(User).filter(User.id.in_([1,3,4])).all() # in_([1,3,4]) 只查询id等于1,3,4的
    ret = session.query(User).filter(~User.id.in_([1,3,4])).all() # ~xxxx.in_([1,3,4]) 查询不等于1,3,4的
    ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='wang'))).all() 子查询
    from sqlalchemy import and_, or_
    ret = session.query(User).filter(and_(User.id > 3, User.name == 'wang')).all()
    ret = session.query(User).filter(or_(User.id < 2, User.name == 'wang')).all()
    ret = session.query(User).filter(
        or_(
            User.id < 2,
            and_(User.name == 'eric', User.id > 3),
            User.extra != ""
        )).all()
    # select * from User where id<2 or (name="eric" and id>3) or extra != "" 
    
    # 通配符
    ret = db_session.query(User).filter(User.name.like('e%')).all()
    ret = db_session.query(User).filter(~User.name.like('e%')).all()
    
    # 限制
    ret = db_session.query(User)[1:2]
    
    # 排序
    ret = db_session.query(User).order_by(User.name.desc()).all()
    ret = db_session.query(User).order_by(User.name.desc(), User.id.asc()).all()
    
    # 分组
    from sqlalchemy.sql import func
    
    ret = db_session.query(User).group_by(User.extra).all()
    ret = db_session.query(
        func.max(User.id),
        func.sum(User.id),
        func.min(User.id)).group_by(User.name).all()
    
    ret = db_session.query(
        func.max(User.id),
        func.sum(User.id),
        func.min(User.id)).group_by(User.name).having(func.min(User.id) >2).all()
    """
    
    # 关闭连接
    db_session.close()

高级版修改数据操作

#高级版更新操作
    from my_create_table import User,engine
    from sqlalchemy.orm import sessionmaker
    
    Session = sessionmaker(engine)
    db_session = Session()
    
    #直接修改
    db_session.query(User).filter(User.id > 0).update({"name" : "099"})
    
    #在原有值基础上添加 - 1
    db_session.query(User).filter(User.id > 0).update({User.name: User.name + "099"}, synchronize_session=False)
    
    #在原有值基础上添加 - 2
    db_session.query(User).filter(User.id > 0).update({"age": User.age + 1}, synchronize_session="evaluate")
    db_session.commit()

3.一对多的操作：ForeignKey

创建数据表及关系：relationship

from sqlalchemy.ext.declarative import declarative_base
    
    Base = declarative_base()
    
    # 这次我们要多导入一个 ForeignKey 字段了,外键关联对了
    from sqlalchemy import Column,Integer,String,ForeignKey
    # 还要从orm 中导入一个 relationship 关系映射
    from sqlalchemy.orm import relationship
    
    class ClassTable(Base):
        __tablename__="classtable"
        id = Column(Integer,primary_key=True)
        name = Column(String(32),index=True)
    
    class Student(Base):
        __tablename__="student"
        id=Column(Integer,primary_key=True)
        name = Column(String(32),index=True)
    
        # 关联字段,让class_id 与 class 的 id 进行关联,主外键关系(这里的ForeignKey一定要是表名.id不是对象名)
        class_id = Column(Integer,ForeignKey("classtable.id"))
    
        # 将student 与 classtable 创建关系 这个不是字段,只是关系,backref是反向关联的关键字
        to_class = relationship("ClassTable",backref = "stu2class")
    
    from sqlalchemy import create_engine
    engine = create_engine("mysql+pymysql://root:pwd@127.0.0.1:3306/dragon?charset=utf8")
    
    Base.metadata.create_all(engine)

基于relationship增加数据

from my_ForeignKey import Student, ClassTable,engine
    # 创建连接
    from sqlalchemy.orm import sessionmaker
    # 创建数据表操作对象 sessionmaker
    DB_session = sessionmaker(engine)
    db_session = DB_session()
    
    # 增加数据
    # 1.简单增加数据
    # 添加两个班级:
    # db_session.add_all([
    #     ClassTable(name="OldBoyS1"),
    #     ClassTable(name="OldBoyS2")
    # ])
    # db_session.commit()
    # 添加一个学生 DragonFire 班级是 OldBoyS1
    # 查询要添加到的班级
    # class_obj = db_session.query(ClassTable).filter(ClassTable.name == "OldBoyS1").first()
    # 创建学生
    # stu = Student(name="DragonFire",class_id = class_obj.id)
    # db_session.add(stu)
    # db_session.commit()
    
    # 2. relationship版 添加数据
    # 通过关系列 to_class 可以做到两件事
    # 第一件事 在ClassTable表中添加一条数据
    # 第二件事 在Student表中添加一条数据并将刚刚添加的ClassTable的数据id填写在Student的class_id中
    # stu_cla = Student(name="DragonFire",to_class=ClassTable(name="OldBoyS1"))
    # print(stu_cla.name,stu_cla.class_id)
    # db_session.add(stu_cla)
    # db_session.commit()
    
    # 3.relationship版 反向添加数据
    # 首先建立ClassTable数据
    class_obj = ClassTable(name="OldBoyS2")
    # 通过class_obj中的反向关联字段backref - stu2class
    # 向 Student 数据表中添加 2条数据 并将 2条数据的class_id 写成 class_obj的id
    # class_obj.stu2class = [Student(name="BMW"),Student(name="Audi")]
    # db_session.add(class_obj)
    # db_session.commit()
    
    # 关闭连接
    db_session.close()

基于relationship查询数据

from my_ForeignKey import Student, ClassTable,engine
    
    from sqlalchemy.orm import sessionmaker
    DB_session = sessionmaker(engine)
    db_session = DB_session()
    
    # 1.查询所有数据,并显示班级名称,连表查询
    student_list = db_session.query(Student).all()
    for row in student_list:
        # row.to_class.name 通过Student对象中的关系字段relationship to_class 获取关联 ClassTable中的name
        print(row.name,row.to_class.name,row.class_id)
    
    # 2.反向查询
    class_list = db_session.query(ClassTable).all()
    for row in class_list:
        for row2 in row.stu2class:
            print(row.name,row2.name)
    # row.stu2class 通过 backref 中的 stu2class 反向关联到 Student 表中根据ID获取name
    
    
    db_session.close()

更新数据

from my_ForeignKey import Student, ClassTable,engine
    
    from sqlalchemy.orm import sessionmaker
    DB_session = sessionmaker(engine)
    db_session = DB_session()
    
    # 更新
    class_info = db_session.query(ClassTable).filter(ClassTable.name=="OldBoyS1").first()
    db_session.query(Student).filter(Student.class_id == class_info.id).update({"name":"NBDragon"})
    db_session.commit()
    
    db_session.close()

删除数据

from my_ForeignKey import Student, ClassTable,engine
    
    from sqlalchemy.orm import sessionmaker
    DB_session = sessionmaker(engine)
    db_session = DB_session()
    
    # 删除
    class_info = db_session.query(ClassTable).filter(ClassTable.name=="OldBoyS1").first()
    db_session.query(Student).filter(Student.class_id == class_info.id).delete()
    db_session.commit()
    
    db_session.close()

4.多对多：ManyToMany

创建表及关系

from sqlalchemy.ext.declarative import declarative_base
    
    Base = declarative_base()
    
    from sqlalchemy import Column,Integer,String,ForeignKey
    from sqlalchemy.orm import relationship
    
    class Hotel(Base):
        __tablename__="hotel"
        id=Column(Integer,primary_key=True)
        girl_id = Column(Integer,ForeignKey("girl.id"))
        boy_id = Column(Integer,ForeignKey("boy.id"))
    
    class Girl(Base):
        __tablename__="girl"
        id=Column(Integer,primary_key=True)
        name = Column(String(32),index=True)
    
        #创建关系
        boys = relationship("Boy",secondary="hotel",backref="girl2boy")
    
    
    class Boy(Base):
        __tablename__="boy"
        id=Column(Integer,primary_key=True)
        name = Column(String(32),index=True)
    
    
    from sqlalchemy import create_engine
    engine = create_engine("mysql+pymysql://root:pwd@127.0.0.1:3306/dragon?charset=utf8")
    
    Base.metadata.create_all(engine)

基于relationship 增加数据

from my_M2M import Girl,Boy,Hotel,engine
    
    # 创建连接
    from sqlalchemy.orm import sessionmaker
    # 创建数据表操作对象 sessionmaker
    DB_session = sessionmaker(engine)
    db_session = DB_session()
    
    # 1.通过Boy添加Girl和Hotel数据
    boy = Boy(name="wang")
    boy.girl2boy = [Girl(name="赵丽颖"),Girl(name="Angelababy")]
    db_session.add(boy)
    db_session.commit()
    
    # 2.通过Girl添加Boy和Hotel数据
    girl = Girl(name="珊珊")
    girl.boys = [Boy(name="Dragon")]
    db_session.add(girl)
    db_session.commit()

基于relationship 查询数据

from my_M2M import Girl,Boy,Hotel,engine
    
    # 创建连接
    from sqlalchemy.orm import sessionmaker
    # 创建数据表操作对象 sessionmaker
    DB_session = sessionmaker(engine)
    db_session = DB_session()
    
    # 1.通过Boy查询约会过的所有Girl
    hotel = db_session.query(Boy).all()
    for row in hotel:
        for row2 in row.girl2boy:
            print(row.name,row2.name)
    
    # 2.通过Girl查询约会过的所有Boy
    hotel = db_session.query(Girl).all()
    for row in hotel:
        for row2 in row.boys:
            print(row.name,row2.name)