SQLALchemy操作MySQL关系型数据库

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 = 3306 username = '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!

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

  Base.metadata.drop_all(engine)  

• 插入数据 使用 session.add_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): __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 = 3306 username = '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;

• 查询操作 使用 session.query() 方法，迭代查询；

  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): __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 = 3306 username = '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) # 返回 aobama print(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_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): __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 = 3306 username = '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_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, DATE, Enum, ForeignKey import enum from sqlalchemy.orm import sessionmaker Base = 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 = 3306 username = '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)) # in emploees = session.query(Emploee).filter(Emploee.emp_no.in_([10010, 10011, 10012])) # not in emploees = 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