文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

SQLALchemy操作MySQL关系型数据库

2023-06-02 07:02

关注
1.SQLALchemy使用
  • 安装
pip install sqlalchemy;
  • SQLAlchemy是Python编程语言下的一款开源软件,是PythonSQL工具包和对象关系映射器,它为应用程序开发人员提供了SQL的全部功能和灵活性
2.什么是ORM
  • ORM(Object Relational Mapper)就是把数据库表的行与相应的对象建立关联,互相转换;

  • 目前,最知名的Python ORM是SQLAlchemy和SQLobject;

3.数据库操作
  • 创建表使用create_all()方法,删除表使用drop_all()方法,我们一起来看一下如何创建和删除表
# 导入引擎模块from sqlalchemy import create_engine# 导入基类模块from sqlalchemy.ext.declarative import declarative_base# 导入字段类from sqlalchemy import Column, Integer, String# 导入会话模块from sqlalchemy.orm import sessionmaker# 实体类的基类Base = declarative_base()# 实体类class Teacher(Base):    '''    实体类的创建有两个方面的用处:    1:如果数据库中没有表,那么可以使用实体类创建    2:如果数据库中有表,实体类可以映射表的结构,对表的CRUD操作    '''    # 表名必须指定    __tablename__ = 'teacher'    id = Column(Integer, primary_key=True, autoincrement=True)    name = Column(String(255))    age = Column(Integer)    # 打印实例的使用可以发现,数据也是保存在实体类实例的__dict__中    def __repr__(self):        # print(self.__dict__)        return "id='%s', name='%s', age='%s" % (self.id, self.name, self.age)    __str__ = __repr__# 创建连接引擎host = 'localhost'port = 3306username = 'zengzeng'password = '123456'db = 'XKD_Python_Course'connect_str = 'mysql+mysqldb://{}:{}@{}:{}/{}'.format(username, password, host, port,db)engine = create_engine(connect_str, echo=True)# 创建表Base.metadata.create_all(engine)# 删除表# Base.metadata.drop_all(engine)

我们现在命令工具中查看一下数据库中有没有我们想要的teacher表,上一篇文章中讲过,先登录mysql:mysql -uzengzeng -p123456,然后进入我们要使用的数据库:use XKD_Python_Course,我们先通过:show tables;查看一下数据库中的表,发现没有teacher表
那现在就可以执行代码,创建数据库了,创建好后我们可以在命令行查看一下:show tables;,发现teacher表已经存在了,创建表成功yes!

SQLALchemy操作MySQL关系型数据库

想要删除刚刚创建的表,可以使用drop_all()方法,执行代码,再查看数据库表,就可以看teacher表已经被删除了

Base.metadata.drop_all(engine)
  • 插入数据使用session.add_all()方法;
from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import sessionmakerBase = declarative_base()class Teacher(Base):    __tablename__ = 'teacher'    id = Column(Integer, primary_key=True, autoincrement=True)    name = Column(String(255))    age = Column(Integer)    def __repr__(self):        # print(self.__dict__)        return "id='%s', name='%s', age='%s" % (self.id, self.name, self.age)    __str__ = __repr__host = 'localhost'port = 3306username = 'zengzeng'password = '123456'db = 'XKD_Python_Course'connect_str = 'mysql+mysqldb://{}:{}@{}:{}/{}'.format(username, password, host, port,db)engine = create_engine(connect_str, echo=True)# 创建会话,用于提交数据Session = sessionmaker(bind=engine)session = Session()# 创建多行行实例,给表添加数据try:    lst = []    for i in range(10):        teacher = Teacher()        teacher.name = 'zengzeng' + str(i)        teacher.age = 20 + i        lst.append(teacher)        print(teacher)    # session.add(student) 可以添加一行记录,也可以添加多行记录    # 注意:这里将行记录实例添加到session,不会提交,需要手动提交    session.add_all(lst)except Exception as e:    print('~~~~~~~~~~~'*200)    session.rollback()    print(e)finally:    session.commit()

我们执行代码,然后去命令工具查看一下表是否插入数据:select * from teacher;

SQLALchemy操作MySQL关系型数据库

  • 查询操作使用session.query()方法,迭代查询;
from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import sessionmakerBase = declarative_base()class Teacher(Base):    __tablename__ = 'teacher'    id = Column(Integer, primary_key=True, autoincrement=True)    name = Column(String(255))    age = Column(Integer)    def __repr__(self):        # print(self.__dict__)        return "id='%s', name='%s', age='%s" % (self.id, self.name, self.age)    __str__ = __repr__host = 'localhost'port = 3306username = 'zengzeng'password = '123456'db = 'XKD_Python_Course'conn_str = 'mysql+mysqldb://{}:{}@{}:{}/{}'.format(username, password, host, port,db)engine = create_engine(conn_str, echo=False)Session = sessionmaker(bind=engine)session = Session()teacher_obj = session.query(Teacher)for teacher in  teacher_obj:    print(teacher) # 返回结果:{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x1063125f8>, 'age': 20, 'name': 'nihao0', 'id': 1}print('*'*300)# 直接返回实例对象teacher = session.query(Teacher).get(4)print(teacher) # 返回结果:{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x1075fd400>, 'age': 23, 'name': 'nihao3', 'id': 4}print(teacher.id)  # 返回 aobamaprint(teacher.name)print(teacher.age)print('*'*300)# 返回的是可迭代对象teacher_results = session.query(Teacher).filter(Teacher.id == 5)for teacher in teacher_results:  # 拿到student实例对象    print(teacher)
  • 修改操作:session.update()
from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import sessionmakerBase = declarative_base()class Teacher(Base):    __tablename__ = 'teacher'    id = Column(Integer, primary_key=True, autoincrement=True)    name = Column(String(255))    age = Column(Integer)    def __repr__(self):        # print(self.__dict__)        return "id='%s', name='%s', age='%s" % (self.id, self.name, self.age)    __str__ = __repr__# 创建连接引擎host = 'localhost'port = 3306username = 'zengzeng'password = '123456'db = 'XKD_Python_Course'conn_str = 'mysql+mysqldb://{}:{}@{}:{}/{}'.format(username, password, host, port,db)engine = create_engine(conn_str, echo=False)# 创建会话,用于提交数据Session = sessionmaker(bind=engine)session = Session()teacher = session.query(Teacher).get(4)teacher.name = 'Robby'teacher.age = '99'session.commit()
where条件查询
from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, DATE, Enum, ForeignKeyimport enumfrom sqlalchemy.orm import sessionmakerBase = declarative_base()class My_Enum(enum.Enum):    M = 'M'    F = 'F'# 实体类class Emploee(Base):    '''    +------------+---------------+------+-----+---------+-------+    | Field      | Type          | Null | Key | Default | Extra |    +------------+---------------+------+-----+---------+-------+    | emp_no     | int(11)       | NO   | PRI | NULL    |       |    | birth_date | date          | NO   |     | NULL    |       |    | first_name | varchar(14)   | NO   |     | NULL    |       |    | last_name  | varchar(16)   | NO   |     | NULL    |       |    | gender     | enum('M','F') | NO   |     | NULL    |       |    | hire_date  | date          | NO   |     | NULL    |       |    +------------+---------------+------+-----+---------+-------+    '''    __tablename__ = 'employees'    emp_no = Column(Integer, primary_key=True, nullable=False)    birth_date = Column(DATE, nullable=False)    first_name = Column(String(14), nullable=False)    last_name = Column(String(16),nullable=False)    gender = Column(Enum(My_Enum), nullable=False)    hire_date = Column(DATE, nullable=False)    def __repr__(self):        return "emp_no='%s', birth_date='%s', first_name='%s', last_name='%s', gender='%s', hire_date='%s'" % (self.emp_no, self.birth_date, self.first_name, self.last_name, self.gender, self.hire_date)    __str__ = __repr__# 创建连接引擎host = 'localhost'port = 3306username = 'zengzeng'password = '123456'db = 'XKD_Python_Course'conn_str = 'mysql+mysqldb://{}:{}@{}:{}/{}'.format(username, password, host, port,db)engine = create_engine(conn_str, echo=False)# 创建表Base.metadata.create_all(engine)# 创建会话,用于提交数据Session = sessionmaker(bind=engine)session = Session()# # 简单的where条件查询过滤, 返回可迭代对象, AND 取与emploees = session.query(Emploee).filter(Emploee.emp_no < 10010).filter(Emploee.gender == 'M')emploees = session.query(Emploee).filter((Emploee.emp_no > 10010) & (Emploee.gender == 'F'))# OR 取并emploees = session.query(Emploee).filter((Emploee.emp_no > 10010) | (Emploee.gender == 'F'))# NOT 取反emploees = session.query(Emploee).filter(~(Emploee.emp_no > 10010))# inemploees = session.query(Emploee).filter(Emploee.emp_no.in_([10010, 10011, 10012]))# not inemploees = session.query(Emploee).filter(~Emploee.emp_no.in_([10010, 10011, 10012]))# like ,like可以忽略大小写进行模式匹配emploees = session.query(Emploee).filter(Emploee.last_name.like('B%'))for emploee in emploees:    print(emploee)
order排序
  • asc():升序;
例如:emploees = session.query(Emploee).filter(Emploee.last_name.like('B%')).order_by(Emploee.emp_no.asc())
  • desc():降序;
例如:emploees=session.query(Emploee).filter(Emploee.last_name.like('B%')).order_by(Emploee.emp_no.desc())
  • 多字段排序:在语句里面进行多次排序;
聚合与分组
  • list():转化为列表;

  • count():聚合count(*)查询;

  • all(): 转化为列表;

  • limit().one():查询首行;

    emploees = session.query(Emploee)print(list(emploees))         # 转化为列表print(emploees.count())     # 聚合count(*)查询print(emploees.all())        # 转化为列表print(emploees.limit(1).one())    # 查询首行
  • max():返回最大值;

  • min():返回最小值;

  • avg():返回平均值;

    emploees = session.query(func.max(Emploee.emp_no))emploees = session.query(func.min(Emploee.emp_no))emploees = session.query(func.avg(Emploee.emp_no))
  • group_by():分组查询;

emploees = session.query(func.count(Emploee.emp_no)).group_by(Emploee.gender)print(emploees)for emploee in emploees:    print(emploee)

参考:https://www.9xkd.com/user/plan-view.html?id=2415909403

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 资料下载
  • 历年真题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯