Python操作数据库之sqlalchemy

安装

pip install -i https://pypi.douban.com/simple sqlalchemy

---

ModuleNotFoundError: No module named 'MySQLdb'。由于sqlalchemy建立在 DB API之上，使用关系对象映射进行数据库操作,缺少依赖模块，安装即可。

---

Windows系统

pip install -i https://pypi.douban.com/simple mysqlclient

---

Linux系统

pip install -i https://pypi.douban.com/simple PyMySQL

---

使用

1. 创建数据表

文件名：model.py

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey

# 通过declarative_base()功能创建一个基类
Base = declarative_base()


# 定义三个表，student表，course表和sc表，以及对应的字段
class Student(Base):
    __tablename__ = 'student'
    Sno = Column(String, primary_key=True)
    Sname = Column(String)
    Ssex = Column(String)
    Sage = Column(Integer)
    Sdept = Column(String)
    S_entrance = Column(String)

    def __repr__(self):
        return "Sno {}".format(self.Sno)

    def to_json(self):
        diction = self.__dict__
        if "_sa_instance_state" in diction:
            del diction["_sa_instance_state"]
        return diction


class SC(Base):
    __tablename__ = 'sc'
    Sno = Column(String, ForeignKey('student.Sno'), primary_key=True)
    Cno = Column(String, ForeignKey('course.Cno'), primary_key=True)
    Grade = Column(Integer)

    def __repr__(self):
        return "Sno {}".format(self.Sno)

    def to_json(self):
        diction = self.__dict__
        if "_sa_instance_state" in diction:
            del diction["_sa_instance_state"]
        return diction


class Course(Base):
    __tablename__ = 'course'
    Cno = Column(String, primary_key=True)
    Cname = Column(String)
    Cpno = Column(String, ForeignKey('course.Cno'))
    Ccredit = Column(Integer)

    def __repr__(self):
        return "Cno {}".format(self.Cno)

    def to_json(self):
        diction = self.__dict__
        if "_sa_instance_state" in diction:
            del diction["_sa_instance_state"]
        return diction

2. 数据库连接

文件名：config.py

from sqlalchemy import create_engine
# 通过create_engine()返回一个Engine的实例
engine = create_engine("mysql://root:admin123@127.0.0.1:3306/school")

3. 创建会话session

文件名：config.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("mysql://root:admin123@127.0.0.1:3306/school")
# 这个定制的Session类会创建绑定到数据库的Session对象。
Session = sessionmaker(bind=engine)

4.数据库查询数据

1. 使用all()，返回一个列表
2. 使用first(),返回一个结果
3. 使用one(),返回且仅返回一个查询结果。当结果的数量不足一个或者多于一个时会报错。
4. 使用count(),计数
   文件名：dbsession.py

1. 单表查询

---

查询学生表所有数据

from config import Session
from model import Student


def get_all_student_data():
    # 创建一个实例
    session = Session()
    result_list = session.query(Student).all()
    results = [result.to_json() for result in result_list]
    # 关闭session
    session.close()
    return results

运行该函数，结果输出：

[{'S_entrance': datetime.date(2019, 8, 6), 'Sage': 20, 'Sname': 'liyong', 'Sno': '201915121', 'Sdept': 'CS', 'Ssex': 'male'}, {'S_entrance': datetime.date(2019, 8, 6), 'Sage': 19, 'Sname': 'liuchen', 'Sno': '201915122', 'Sdept': 'CS', 'Ssex': 'female'}, {'S_entrance': datetime.date(2019, 8, 6), 'Sage': 18, 'Sname': 'wangming', 'Sno': '201915123', 'Sdept': 'MA', 'Ssex': 'female'}, {'S_entrance': datetime.date(2019, 8, 6), 'Sage': 19, 'Sname': 'zhangli', 'Sno': '201915125', 'Sdept': 'IS', 'Ssex': 'male'}]

---

使用filter或filter_by进行查询

def get_student_data():
    session = Session()
    result_list = session.query(Student).filter(Student.Sage == 20).all()
    # result_list = session.query(Student).filter_by(Sage=20).all()
    results = [result.to_json() for result in result_list]
    session.close()
    return results

---

order_by

result_list = session.query(Student).order_by(Student.Sage.desc()).all()
result_list = session.query(Student).order_by(Student.Sage.desc(), Student.Sdept.asc()).all()

---

group_by

result_list = session.query(Student).group_by(Student.Ssex).all()

---

通配符

result_list = session.query(Student).filter(Student.Sname.like('li%')).all()

---

in

result_list = session.query(Student).filter(Student.Sdept.in_(['CS', 'IS'])).all()

---

not in

result_list = session.query(Student).filter(~Student.Sdept.in_(['CS', 'IS'])).all()

---

startswith

result_list = session.query(Student).filter(Student.Sno.startswith('1')).all()

---

endswith

result_list = session.query(Student).filter(Student.Sno.endswith('1')).all()

---

contains

result_list = session.query(Student).filter(Student.Sno.contains('1')).all()

---

and

from sqlalchemy import and_
result_list = session.query(Student).filter(and_(Student.Ssex == 'female', Student.Sage == 18)).all()

---

or

from sqlalchemy import or_
result_list = session.query(Student).filter(or_(Student.Ssex == 'female', Student.Sage == 18)).all()

---

between

result_list = session.query(Student).filter(Student.Sage.between(16, 18)).all()

---

2. 联表查询

result_list = session.query(Student, SC).filter(Student.Sno == SC.Sno).filter(Student.Sage == 20).all()

5. 数据库新增数据

import datetime
from config import Session
from model import Student

def add_student_data():
    session = Session()
    student = Student(S_entrance=datetime.datetime.now(), Sno='201915126', Sname='caixukun', Ssex='male', Sage=18,
                      Sdept='CS')
    session.add(student)  
    # 一定要commit，不然不会新增数据
    session.commit()
    session.close()

代码执行前
代码执行后

6. 数据库修改数据

from config import Session
from model import Student

def set_student_data():
    session = Session()
    student = session.query(Student).filter(Student.Sname == 'caixukun').first()
    student.Sage = 19
    session.commit()
    session.close()

代码执行前
代码执行后

7. 数据库删除数据

from config import Session
from model import Student

def del_student_data():
    session = Session()
    student = session.query(Student).filter(Student.Sname == 'caixukun').first()
    session.delete(student)
    session.commit()
    session.close()

代码执行前
代码执行后