SQLALchemy操作MySQL關(guān)系型數(shù)據(jù)庫(kù)

1.SQLALchemy使用

• 安裝

pip install sqlalchemy；

• SQLAlchemy是Python編程語(yǔ)言下的一款開源軟件，是PythonSQL工具包和對(duì)象關(guān)系映射器，它為應(yīng)用程序開發(fā)人員提供了SQL的全部功能和靈活性

2.什么是ORM

• ORM（Object Relational Mapper）就是把數(shù)據(jù)庫(kù)表的行與相應(yīng)的對(duì)象建立關(guān)聯(lián)，互相轉(zhuǎn)換；

• 目前，最知名的Python ORM是SQLAlchemy和SQLobject；

3.數(shù)據(jù)庫(kù)操作

• 創(chuàng)建表 使用 create_all() 方法， 刪除表 使用 drop_all() 方法，我們一起來(lái)看一下如何創(chuàng)建和刪除表

# 導(dǎo)入引擎模塊
from sqlalchemy import create_engine
# 導(dǎo)入基類模塊
from sqlalchemy.ext.declarative import declarative_base
# 導(dǎo)入字段類
from sqlalchemy import Column, Integer, String
# 導(dǎo)入會(huì)話模塊
from sqlalchemy.orm import sessionmaker
# 實(shí)體類的基類
Base = declarative_base()
# 實(shí)體類
class Teacher(Base):
    '''
    實(shí)體類的創(chuàng)建有兩個(gè)方面的用處：
    1：如果數(shù)據(jù)庫(kù)中沒有表，那么可以使用實(shí)體類創(chuàng)建
    2：如果數(shù)據(jù)庫(kù)中有表，實(shí)體類可以映射表的結(jié)構(gòu)，對(duì)表的CRUD操作
    '''
    # 表名必須指定
    __tablename__ = 'teacher'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255))
    age = Column(Integer)
    # 打印實(shí)例的使用可以發(fā)現(xiàn)，數(shù)據(jù)也是保存在實(shí)體類實(shí)例的__dict__中
    def __repr__(self):
        # print(self.__dict__)
        return "id='%s', name='%s', age='%s" % (self.id, self.name, self.age)
    __str__ = __repr__
# 創(chuàng)建連接引擎
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)
# 創(chuàng)建表
Base.metadata.create_all(engine)
# 刪除表
# Base.metadata.drop_all(engine)

我們現(xiàn)在命令工具中查看一下數(shù)據(jù)庫(kù)中有沒有我們想要的teacher表，上一篇文章中講過(guò)，先登錄mysql： mysql -uzengzeng -p123456 ，然后進(jìn)入我們要使用的數(shù)據(jù)庫(kù)： use XKD_Python_Course ，我們先通過(guò)： show tables; 查看一下數(shù)據(jù)庫(kù)中的表，發(fā)現(xiàn)沒有teacher表
那現(xiàn)在就可以執(zhí)行代碼，創(chuàng)建數(shù)據(jù)庫(kù)了，創(chuàng)建好后我們可以在命令行查看一下： show tables; ，發(fā)現(xiàn)teacher表已經(jīng)存在了，創(chuàng)建表成功yes!

想要?jiǎng)h除剛剛創(chuàng)建的表，可以使用 drop_all() 方法，執(zhí)行代碼，再查看數(shù)據(jù)庫(kù)表，就可以看teacher表已經(jīng)被刪除了

Base.metadata.drop_all(engine)

• 插入數(shù)據(jù) 使用 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)
# 創(chuàng)建會(huì)話，用于提交數(shù)據(jù)
Session = sessionmaker(bind=engine)
session = Session()
# 創(chuàng)建多行行實(shí)例，給表添加數(shù)據(jù)
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) 可以添加一行記錄，也可以添加多行記錄
    # 注意：這里將行記錄實(shí)例添加到session，不會(huì)提交，需要手動(dòng)提交
    session.add_all(lst)
except Exception as e:
    print('~~~~~~~~~~~'*200)
    session.rollback()
    print(e)
finally:
    session.commit()

我們執(zhí)行代碼，然后去命令工具查看一下表是否插入數(shù)據(jù)： 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) # 返回結(jié)果：{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x1063125f8>, 'age': 20, 'name': 'nihao0', 'id': 1}
print('*'*300)
# 直接返回實(shí)例對(duì)象
teacher = session.query(Teacher).get(4)
print(teacher) # 返回結(jié)果：{'_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)
# 返回的是可迭代對(duì)象
teacher_results = session.query(Teacher).filter(Teacher.id == 5)
for teacher in teacher_results:  # 拿到student實(shí)例對(duì)象
    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__
# 創(chuàng)建連接引擎
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)
# 創(chuàng)建會(huì)話，用于提交數(shù)據(jù)
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'
# 實(shí)體類
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__
# 創(chuàng)建連接引擎
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)
# 創(chuàng)建表
Base.metadata.create_all(engine)
# 創(chuàng)建會(huì)話，用于提交數(shù)據(jù)
Session = sessionmaker(bind=engine)
session = Session()
# # 簡(jiǎn)單的where條件查詢過(guò)濾， 返回可迭代對(duì)象, 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可以忽略大小寫進(jìn)行模式匹配
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())

• 多字段排序：在語(yǔ)句里面進(jìn)行多次排序；

聚合與分組

• list() ：轉(zhuǎn)化為列表；

• count() ：聚合count(*)查詢；

• all() ： 轉(zhuǎn)化為列表；

• limit().one() ：查詢首行；

  
  emploees = session.query(Emploee)
  print(list(emploees))         # 轉(zhuǎn)化為列表
  print(emploees.count())     # 聚合count(*)查詢
  print(emploees.all())        # 轉(zhuǎn)化為列表
  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