使用SQLALchemy操作MySql数据库

刺骨的言语ヽ痛彻心扉 2023-02-22 10:40 16阅读 0赞

**连接数据库并进行操作**

\[ORM概念\]

是Object-Relational Mapping即对象关系映射，就是将关系型数据库和对象之间做一个映射，这样就可以不用使用SQL语句，直接用python语句就可以处理数据库了，其中，python中最成熟的ORM库就是SQLALchemy～

使用SQLALchemy进行数据库操作，需要三步，定义表（对应着数据库的表），与数据库进行连接，对数据库进行操作。

**安装库**

pip3 install sqlalchemy
    # 如果pymysql驱动失败，需要安装
    pip install mysql-connector
    pip3 install mysql-connector-python

**代码**

# -*- coding:utf-8 -*-
    # ---------------定义表需要的类-----------------
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String, ForeignKey, text, and_, or_, func
    # --------------连接数据库需要的类---------------
    from sqlalchemy import create_engine  # 建立数据库引擎
    from sqlalchemy.orm import sessionmaker, relationship  # 建立会话session
    
    # ---------第一部分：定义表-----------
    from sqlalchemy.testing import in_
    
    Base = declarative_base()  # 实例,创建基类
    
    
    # 所有的表必须继承于Base
    class Enhancer(Base):
        __tablename__ = 'my_test'  # 定义该表在mysql数据库中的实际名称
        # 定义表的内容
        id = Column(Integer, primary_key=True)
        name = Column(String(32), nullable=False)
        age = Column(Integer, nullable=False)
        gender = Column(Integer, nullable=False)
    
    
    # ---------第二部分：与数据库连接--------
    db_connect_string = 'mysql+mysqlconnector://root:123456@localhost:3306/my_db?charset=utf8'
    # 以mysql数据库为例：mysql+数据库驱动：//用户名：密码@localhost:3306/数据库
    engine = create_engine(db_connect_string)  # 创建引擎
    Sesssion = sessionmaker(bind=engine)  # 产生会话
    session = Sesssion()  # 创建Session实例
    # ----------第三部分：进行数据操作--------
    # 提交新数据
    # session.add(Enhancer(name="测试", age=30, gender=1)) # 只能加一条数据
    # session.add_all([Enhancer(name="Bob", age=29, gender=1), Enhancer(name="Huni", age=16, gender=1)])
    # 使用add_all可以一次传入多条数据，以列表的形式。
    # session.commit() # 提交数据
    
    # 获取数据
    res = session.query(Enhancer).all()
    print(res)
    print(type(res))
    for item in res:
        pass
        # print(item.name)
    
    # 模糊查询like()
    rs = session.query(Enhancer).filter(Enhancer.name.like('R%')).all()
    # 通配符%, %ｃ代表以ｃ结尾，ｃ%代表以ｃ开头，%c%代表包含ｃ
    for i in rs:
        print(i.id, i.name)
    
    # 删除记录
    session.query(Enhancer).filter(Enhancer.name == "测试").delete()
    session.commit()
    
    # 第一种更新记录方法
    rs = session.query(Enhancer).filter(Enhancer.name.like('%i')).first()
    # print(rs) # 结果：<__main__.Enhancer object at 0x7fd7117d9a50>
    # print(type(rs)) # 结果：<class '__main__.Enhancer'>
    # rs.name = "Huya"
    # 第二种更新记录方法
    # synchronize_session=False,意思是不对session进行同步，直接更新数据库。
    session.query(Enhancer).filter(Enhancer.name == "Huya").update({ 'name': 'Huy', 'age': 14}, synchronize_session=False)
    session.commit()
    
    
    # 多张表操作
    # 定义两个表
    class Class(Base):
        __tablename__ = 'my_class'
        id = Column(Integer, primary_key=True)
        name = Column(String(32))
        lever = Column(Integer)
        address = Column(String(64))
    
        # 这里需要注意，并不是有一个studens列～，这里的students可以理解为
        # 关联的下一个表中满足条件的所有内容
        # backref用于反向查询
        students = relationship("Student", backref="my_class")
    
    
    class Student(Base):
        __tablename__ = "student"
        id = Column(Integer, primary_key=True)
        name = Column(String(32))
        age = Column(Integer)
        class_id = Column(Integer, ForeignKey('my_class.id'))
    
    
    # 班级表添加数据
    # session.add_all([Class(name="一班", lever=1, address="二楼"), Class(name="二班", lever=2, address="三楼"),
    # Class(name="三班", lever=3, address="八楼")])
    
    # 学生表添加数据
    # session.add_all([Student(name="Ryan", age=19, class_id=1), Student(name="Lucy", age=18, class_id=2),
    # Student(name="Baby", age=19, class_id=3), Student(name="Huy", age=17, class_id=1),
    # Student(name="Jack", age=22, class_id=2), Student(name="Niu", age=19, class_id=1)])
    # session.commit()
    
    # stu = session.query(Student).all()
    # for i in stu:
    # print(i.name)
    
    # 查找一班的所有学生
    class_ = session.query(Class).filter(Class.name == "一班").first()
    for i in class_.students:
        print(i.name)
    print("---------------------------------------------------------------------")
    
    # 补充查询
    r1 = session.query(Student).all()
    # filter传的是表达式，filter_by传的是参数
    r3 = session.query(Student).filter(Student.name == "Ryan").all()
    r4 = session.query(Student).filter_by(name='Ryan').all()
    r5 = session.query(Student).filter_by(name='Ryan').first()
    print(r5.age)
    # :value 和:name 相当于占位符，用params传参数
    r6 = session.query(Student).filter(text("id>:value and name=:name")).params(value=2, name='Ryan').order_by(
        Student.id).all()
    print("r6:", r6)
    # 自定义查询sql
    r7 = session.query(Student).from_statement(text("SELECT * FROM student where name=:name")).params(name='Baby').first()
    print("r7:", r7.name)
    
    print("---------------------------------------------------------------------")
    
    # 常用操作
    # in_ 在范围内
    ret = session.query(Student).filter(Student.id.in_([1, 2, 3])).all()
    print([i.name for i in ret])
    # ~ 非，除..之外
    ret2 = session.query(Student).filter(~Student.id.in_([1, 2, 3])).all()
    print([i.name for i in ret2])
    # and_ 和 or_
    ret3 = session.query(Student).filter(and_(Student.name == "Ryan", Student.age == 18)).all()
    print([i.name for i in ret3])
    ret4 = session.query(Student).filter(or_(Student.name == "Ryan", Student.age == 18)).all()
    print([i.name for i in ret4])
    # % 通配符,以R开头
    ret5 = session.query(Student).filter(Student.name.like("R%")).all()
    print([i.name for i in ret5])
    # 排序，根据id降序排列（从大到小）
    ret6 = session.query(Student).order_by(Student.id.desc()).all()
    print([i.name for i in ret6])
    # 第一个条件重复后，再按第二个条件升序排
    ret7 = session.query(Student).order_by(Student.name.desc(), Student.id.asc()).all()
    print([i.name for i in ret7])
    
    print("---------------------------------------------------------------------")
    
    # 分组
    # 根据班级分组
    ret8 = session.query(Class).group_by(Class.name).all()
    print([i.name for i in ret8])
    # 分组之后取最大id，id之和，最小id
    ret9 = session.query(
        func.max(Class.id),
        func.sum(Class.id),
        func.min(Class.id)).group_by(Class.name).all()
    
    # 连表（默认用forinkey关联）
    res = session.query(Student, Class).filter(Student.class_id == Class.id).all()
    for i in res:
        print(i.Student.name, i.Class.name)
    
    # join表， 默认是inner join
    res2 = session.query(Student).join(Class).all()
    print([i.class_id for i in res2])
    # isouter=True 外连，Student left join Class，没有右连接，反过来即可
    res3 = session.query(Student).join(Class, isouter=True).all()
    # 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上
    res4 = session.query(Student).join(Class, Student.class_id == Class.id, isouter=True).all()
    for i in res4:
        print(i.name)
    res5 = session.query(Student).join(Class, Student.class_id == Class.id, isouter=True)
    print(res5)
    
    # 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集
    # union和union all的区别？
    q1 = session.query(Student.name).filter(Student.id > 2)
    q2 = session.query(Class.name).filter(Class.id < 2)
    res6 = q1.union(q2).all()
    print([i.name for i in res6])
    
    q1 = session.query(Student.name).filter(Student.id > 2)
    q2 = session.query(Class.name).filter(Class.id < 2)
    res7 = q1.union_all(q2).all()
    print([i.name for i in res7])
    
    print("---------------------------------------------------------------------")
    
    # 查询
    cursor = session.execute('select * from my_class')
    result = cursor.fetchall()
    print(result)
    
    # # 添加
    # cursor = session.execute('insert into student(name,age,class_id) values(:value,:age,:class_id)',
    # params={"value": 'lqz', "age": 18,
    # "class_id": 1})
    # session.commit()
    # print(cursor.lastrowid)
    
    # 一对多查询
    stu = session.query(Class).first()
    print([i.name for i in stu.students])
    
    session.close()
    
    stu2 = session.query(Student, Class).join(Class, isouter=True).all()
    for i in stu2:
        print(i[0].name, i[1].name)

**多对多关系表**

class Boy2Girl(Base):
        __tablename__ = 'boy2girl'
        id = Column(Integer, primary_key=True, autoincrement=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(64), unique=True, nullable=False)
    
    
    class Boy(Base):
        __tablename__ = 'boy'
    
        id = Column(Integer, primary_key=True, autoincrement=True)
        hostname = Column(String(64), unique=True, nullable=False)
        
        # 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以
        servers = relationship('Girl', secondary='boy2girl', backref='boys')

**常用操作**

#　条件
    ret = session.query(Users).filter_by(name='lqz').all()
    #表达式，and条件连接
    ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
    ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
    #注意下划线
    ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
    #~非，除。。外
    ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
    #二次筛选
    ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
    from sqlalchemy import and_, or_
    #or_包裹的都是or条件，and_包裹的都是and条件
    ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
    ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
    ret = session.query(Users).filter(
        or_(
            Users.id < 2,
            and_(Users.name == 'eric', Users.id > 3),
            Users.extra != ""
        )).all()
    
    
    # 通配符，以e开头，不以e开头
    ret = session.query(Users).filter(Users.name.like('e%')).all()
    ret = session.query(Users).filter(~Users.name.like('e%')).all()
    
    # 限制，用于分页，区间
    ret = session.query(Users)[1:2]
    
    # 排序，根据name降序排列（从大到小）
    ret = session.query(Users).order_by(Users.name.desc()).all()
    #第一个条件重复后，再按第二个条件升序排
    ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
    
    # 分组
    from sqlalchemy.sql import func
    
    ret = session.query(Users).group_by(Users.extra).all()
    #分组之后取最大id，id之和，最小id
    ret = session.query(
        func.max(Users.id),
        func.sum(Users.id),
        func.min(Users.id)).group_by(Users.name).all()
    #haviing筛选
    ret = session.query(
        func.max(Users.id),
        func.sum(Users.id),
        func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()
    
    # 连表（默认用forinkey关联）
    
    ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
    #join表，默认是inner join
    ret = session.query(Person).join(Favor).all()
    #isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可
    ret = session.query(Person).join(Favor, isouter=True).all()
    #打印原生sql
    aa=session.query(Person).join(Favor, isouter=True)
    print(aa)
    # 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上
    ret = session.query(Person).join(Favor,Person.id==Favor.id, isouter=True).all()
    # 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集
    #union和union all的区别？
    q1 = session.query(Users.name).filter(Users.id > 2)
    q2 = session.query(Favor.caption).filter(Favor.nid < 2)
    ret = q1.union(q2).all()
    
    q1 = session.query(Users.name).filter(Users.id > 2)
    q2 = session.query(Favor.caption).filter(Favor.nid < 2)
    ret = q1.union_all(q2).all()

**原生SQL**

**一对多关系**

import time
    import threading
    
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
    from sqlalchemy.orm import sessionmaker, relationship
    from sqlalchemy import create_engine
    from sqlalchemy.sql import text
    from sqlalchemy.engine.result import ResultProxy
    from db import Users, Hosts, Hobby, Person
    
    engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
    Session = sessionmaker(bind=engine)
    session = Session()
    # 添加
    """ session.add_all([ Hobby(caption='乒乓球'), Hobby(caption='羽毛球'), Person(name='张三', hobby_id=3), Person(name='李四', hobby_id=4), ]) person = Person(name='张九', hobby=Hobby(caption='姑娘')) session.add(person) #添加二 hb = Hobby(caption='人妖') hb.pers = [Person(name='文飞'), Person(name='博雅')] session.add(hb) session.commit() """
    
    # 使用relationship正向查询
    """ v = session.query(Person).first() print(v.name) print(v.hobby.caption) """
    
    # 使用relationship反向查询
    """ v = session.query(Hobby).first() print(v.caption) print(v.pers) """
    #方式一，自己链表
    # person_list=session.query(models.Person.name,models.Hobby.caption).join(models.Hobby,isouter=True).all()
    person_list=session.query(models.Person,models.Hobby).join(models.Hobby,isouter=True).all()
    for row in person_list:
        # print(row.name,row.caption)
        print(row[0].name,row[1].caption)
    
    #方式二：通过relationship
    
    person_list=session.query(models.Person).all()
    for row in person_list:
        print(row.name,row.hobby.caption)
    #查询喜欢姑娘的所有人
    obj=session.query(models.Hobby).filter(models.Hobby.id==1).first()
    persons=obj.pers
    print(persons)
    session.close()

**多对多关系**

import time
    import threading
    
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
    from sqlalchemy.orm import sessionmaker, relationship
    from sqlalchemy import create_engine
    from sqlalchemy.sql import text
    from sqlalchemy.engine.result import ResultProxy
    from db import Users, Hosts, Hobby, Person, Group, Server, Server2Group
    
    engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
    Session = sessionmaker(bind=engine)
    session = Session()
    # 添加
    """ session.add_all([ Server(hostname='c1.com'), Server(hostname='c2.com'), Group(name='A组'), Group(name='B组'), ]) session.commit() s2g = Server2Group(server_id=1, group_id=1) session.add(s2g) session.commit() gp = Group(name='C组') gp.servers = [Server(hostname='c3.com'),Server(hostname='c4.com')] session.add(gp) session.commit() ser = Server(hostname='c6.com') ser.groups = [Group(name='F组'),Group(name='G组')] session.add(ser) session.commit() """
    
    
    # 使用relationship正向查询
    """ v = session.query(Group).first() print(v.name) print(v.servers) """
    
    # 使用relationship反向查询
    """ v = session.query(Server).first() print(v.hostname) print(v.groups) """
    
    
    session.close()