SQLalchemy

安装

---

• pip install sqlalchemy
• pip install pymysql

---

常用数据类型

---

数据类型	对应 Python 类型	说明
Integer	int	整形
String	str	字符串
Float	float	浮点型
DECIMAL	decimal.Decimal	定点型(高精度类型)
Boolean	bool	布尔型
Date	datetime.date	日期
DateTime	datetime.datetime	日期和时间
Time	datetime.time	时间
Enum	str	枚举类型
Text	str	文本类型
LongText	str	长文本类型

---

常用 Column 参数

---

| Column | 重要参数 |
| ———– | ————————————————- | — |
| name | 字段名（可选，该参数优先级高于变量名） |
| type_ | 字段类型，注意：MySQL 中，String 类型必须指定长度 |
| primary_key | 如该参数为 True，那么该字段为主键（False） |
| unique | 如该参数如果为 True，相应字段具有唯一性（False） |
| index | 如该参数为 True，相应字段为索引字段（False） |
| nullable | 如该参数为 False，该字段不允为空（True） | R |
| ForeignKey | 通过该参数指定外键 |
| default | 指定该字段的默认值 |
| onupdate | 更新时赋值 |

• AddTime = Column(type_=DateTime(timezone=True), default=datetime.datetime.now())
• UpdateTime = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)

---

engine 特性及重要参数

---

engine	特性及重要参数
Engine	是数据库核心接口
pool_size = 5	连接池的大小，默认为 5
max_overflow = 10	连接池可溢出最大数量 默认为 10 [不放入连接池]
echo = False	打印 SQL 语句，调用 logging 模块,默认为 False
encoding = ‘utf-8’	编码方式，默认为 utf8
pool_recycle = -1	连接回收时间 -1，永不回收（推荐设置 3600 即 1h）
pool_timeout=30	尝试从连接池中获取连接的超时时间

---

连接数据库

---

• 连接之前配置以下信息

# 配置数据库信息
HOST = 'localhost'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DB = 'test'

# 创建数据库链接URL
DB_URI = f'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DB}'

---

创建连接,执行 sql 语句

---

from sqlalchemy import text

# 自动提交
   with engine.connect() as conn:
       result = conn.execute(text("select 'hello world'"))
   print(result.all())

   # 主动提交
with engine.connect() as conn:
       conn.execute(text("CREATE TABLE some_table (x int, y int)"))
       conn.commit()
       # 最好异常捕获一下 在发生错误的时候 回滚事务
       # PersonSession.rollback()

   # Session会话方式
   stmt = text("SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y").bindparams(y=6)
with Session(engine) as session:
       result = session.execute(stmt)

---

创建表

• 创建方式只有在表不存在的时候会创建
• 已经存在则不会修改现有表的结构

创建方式 1

from sqlalchemy import MetaData
from Settings import USERNAME, PASSWORD, HOST, PORT, DB

from sqlalchemy import create_engine, Column, Integer, String, Enum, Table

DB_URI = f'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DB}'

engine = create_engine(DB_URI, echo=True, future=True)
metadata_obj = MetaData()

Person = Table(
    'Person',
    metadata_obj,
    Column("id", type_=Integer, primary_key=True, autoincrement=True, index=True),
    Column("name", type_=String(length=20), nullable=True),
)

metadata_obj.create_all(bind=engine)

创建方式 2

from sqlalchemy.orm import Session
from Settings import USERNAME, PASSWORD, HOST, PORT, DB

from sqlalchemy import create_engine, Column, Integer, String, Enum, Table

from sqlalchemy.orm import registry

DB_URI = f'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DB}'

engine = create_engine(DB_URI, echo=True, future=True)

mapper_registry = registry()
Base = mapper_registry.generate_base()


class Person(Base):
    # 数据库中的表名
    __tablename__ = 'Person'
    id = Column(type_=Integer, primary_key=True, autoincrement=True, index=True)
    name = Column(type_=String(length=20), nullable=True)
    sex = Column(type_=Enum('男', '女', '未知'), nullable=True)
    card = Column(type_=String(length=20), nullable=True, unique=True)
    phone = Column(type_=String(length=20), nullable=True)
    codeNo = Column(type_=String(length=40), nullable=True)
    address = Column(type_=String(length=60), nullable=True)

    def __repr__(self):
        # 重写此方法 在打印时 将输出此返回值信息
        return f"id={self.id}, name={self.name}, fullname={self.sex}"


mapper_registry.metadata.create_all(bind=engine)

操作已经存在的数据库

from sqlalchemy.orm import Session
from Settings import USERNAME, PASSWORD, HOST, PORT, DB

from sqlalchemy import create_engine, Table, MetaData


DB_URI = f'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DB}'

engine = create_engine(DB_URI, echo=True, future=True)

metadata = MetaData()
session = Session(engine)

# 获取指定数据库表对象
ex_table = Table('dataName', metadata, autoload=True, autoload_with=engine)

# 插入数据
ret = session.execute(ex_table.insert(), {"字段名": "值"})
session.commit()

session = Session(engine)

通过 ORM 增加数据到数据库

from sqlalchemy.orm import Session

session = Session(engine)

session.add(Person(name='anni', sex='女'))

session.add_all([
    Person(name='anni', sex='女'),
    Person(name='anni', sex='女')
])

session.commit()
# 最好异常捕获一下 在发生错误的时候 回滚事务
# PersonSession.rollback()

增删改查

查询一个对象

# 注意两个的区别 如果下面指定了 query(TongChengUserORM) 那么代表查询响应是所有参数
obj = TcORMSession.query(TongChengUserORM).filter(TongChengUserORM.phone == data.username).first()
# 如果这里指定了 TongChengUserORM.phone 那么返回的值 只有一个 phone
obj = TcORMSession.query(TongChengUserORM.phone).filter(TongChengUserORM.phone == data.username).first()

查询所有数据

# 查询所有数据(可以得到一个包含所有数据的对象列表,列表内的每一项通过.属性进行取值)
item_list = session.query(Test).all()

# 循环取值,如果重写了repr那直接输出列表即可
for item in item_list:
    print(item.name, item.age)

按某一字段查询

# 返回的也是一个列表
item_list = session.query(Test.age).all()
print(item_list)

通过 filter 条件查询

# 查询所有数据(可以得到一个包含所有数据的对象列表,列表内的每一项通过.属性进行取值)
item_list = session.query(Test.age).filter(Test.age == 18).all()
print(item_list)

# 只取出一条
item_list = session.query(Test.age).filter(Test.age == 18).first()

通过 order_by 排序

# 通过order_by进行排序,通过test的age进行排序,desc表示倒序排序
item_list = session.query(Test.name,Test.age).order_by(Test.age.desc()).all()
print(item_list)

and 与 or 的搭配查询

# 通过and和or可以进行多语句过滤查询
item_list = session.query(Test.name, Test.age).filter(
    Test.name == '1232' and Test.age == 18
).all()
print(item_list)

like 通配符匹配

# 通过.like进行通配符匹配
item_list = session.query(Test.name, Test.age).filter(
    Test.name.like("%2%")
).all()
print(item_list)

通过.in 进行匹配

# 通过.like进行通配符匹配(匹配到包含123)的
item_list = session.query(Test.name, Test.age).filter(
    Test.name.in_([123])
).all()
print(item_list)

count 计数

# 通过.count()计数
count = session.query(Test.name, Test.age).filter(
    Test.name.in_([123])
).count()
print(count)

修改现有数据

# 通过session查询到name为***的字段,通过ipdate传入一个字典（字典内部将name修改为1235）
session.query(Test).filter(Test.name == '***').update({'name': '1235'})

# 记得要提交
session.commit()

# 在现有的数据上进行修改
session.query(User).filter(User.id > 0).update({User.name: User.name + "_admin"}, synchronize_session=False)
dbsession.query(User).filter(User.id > 0).update({"age": User.age + 1}, synchronize_session="evaluate")

删除数据

# 通过session查询到name为123的字段,调用delete进行删除
session.query(Test).filter(Test.name == '123').delete()
# 记得要提交
session.commit()

# 第二种方法通过session进行删除
session.delete(对象)

高级查询

• and_ 与 or_

from sqlalchemy.sql import and_ , or_
session.query(User).filter(and_(User.id > 3, User.name == 'wang')).all()
session.query(User).filter(or_(User.id < 2, User.name == 'wang')).all()

• text

from sqlalchemy.sql import text
session.query(User).filter(text("id<:value and name=:name"))

表与表的关系

表外键约束

约束	对应操作
RESTRICT	修改或者删除之前去检查从表中是否有对应的数据，如果有，拒绝操作
NO ACTIION	同 RESTRICT 一样
CASCADE	父表数据删除、从表数据也会跟着删除
SET NULL	父表数据删除，从表外键字段设为 NULL

---

外键

• 在创建表与表之间的关系的时候总要创建外键
  • 并且都是创建在 多 表中
• 下面代码中创建了一个外键 uid,指向了 test.id,并且指定 CASCADE (父表数据删除、从表数据也会跟着删除)

from sqlalchemy import ForeignKey
uid = Column(Integer, ForeignKey(Test.id, ondelete="CASCADE"))

一对多

• 举个栗子：一个作者可以出多本书,但一本书只能有一个作者
  • 那就可以将一对多外键设置在书上,并且将他指向作者,当作者从数据库中被删除
  • 数据库将根据外键约束进行操作（比如删除所有关于这个作者的书）

# 创建一个类继承Base后增加属性,(表名,字段等信息)
class User(Base):
   # 数据库中的表名
   __tablename__ = 'user'

   # 创建id主键,自增
   id = Column(type_=Integer, primary_key=True, autoincrement=True)

   # 创建名字,并且长度限制66
   name = Column(String(66))

   # 通过orm的relationship绑定外键Article,注意这里的Article是类名
   article = relationship('Article', back_populates='user')



   class Article(Base):
       __tablename__ = 'article'

       # 创建id主键,自增
       id = Column(type_=Integer, primary_key=True, autoincrement=True)

       # 创建标题,并且长度限制66
       title = Column(String(66))

       # 绑定用户外键（这个外键在一对多中是创建在多表中）
       uid = Column(Integer, ForeignKey(User.id, ondelete="CASCADE"))

       # 通过orm的relationship绑定外键user,注意这里的User是类名
       # 第二个参数,为反向查找的时候用到的名字
       user = relationship('User', back_populates='article')

---

一对一

• 为什么要把一对一放在一对多后面呢？
• 其实不难发现,上面的一对多其实就是用户表指向了多个文章（列表）
• 那只要不让上面的用户指向多个表不就变成一对一的
• 所以只要在上面的用户表中增加 uselist=False 即可变为一对一
• 查询操作都大同小异

# 通过orm的relationship绑定外键Article,注意这里的Article是类名（指定为一对一）
article = relationship('Article', back_populates='user', uselist=False)

# 在创建的时候不能再使用列表,只能单个单个的创建
mm = User(name='木木')
mm.article = Article(title='新华字典')

---

多对多

• 多对多是需要额外创建一张表的（在 mysql 中也是这样）
• Django 中的多对多不需要是因为 Django 会自动帮我们创建（跑题了）

from sqlalchemy import Table
    # 多对多的插件需要多创建一个三方表进行关联
    # 第一个参数：表在数据库中名字
    # 第二个参数：Base的元类
    # 后面的参数就是绑定多对多的表
    association_table = Table(
        'association_table',
        Base.metadata,
        Column('teacher_id', Integer, ForeignKey('teacher.id'), primary_key=True),
        Column('classes_id', Integer, ForeignKey('classes.id'), primary_key=True),
    )


    class Teacher(Base):
        # 创建老师类
        __tablename__ = 'teacher'

        id = Column(Integer, primary_key=True, autoincrement=True)

        name = Column(String(20))

        classes = relationship('Classes', secondary=association_table, overlaps="classes")


    class Classes(Base):
        # 创建班级类
        __tablename__ = 'classes'

        id = Column(Integer, primary_key=True, autoincrement=True)

        name = Column(String(20))

        teacher = relationship('Teacher', secondary=association_table, overlaps="classes")

• 执行完毕后就创建了多对多关系的表。
• 在多对多的绑定中,只要有一个进行绑定即可,如果要两个都绑定则需要增加overlaps属性

多对多数据的创建

# 创建多对多数据 先创建三个老师
t1 = Teacher(name='王老师')
t2 = Teacher(name='林老师')
t3 = Teacher(name='龚老师')

# 再创建三个班级
c1 = Classes(name='一班')
c2 = Classes(name='二班')
c3 = Classes(name='三班')

# t1教三个班,t2和t3只教两个班
t1.classes = [c1, c2, c3]
t2.classes = [c1, c2]
t3.classes = [c2, c3]

# 批量增加数据到数据库
session.add_all([
   t1, t2, t3
])

# 单条增加数据到数据库
session.add(t2)

# 记得提交数据
session.commit()
# 最好异常捕获一下 在发生错误的时候 回滚事务
# PersonSession.rollback()

---

多表查询

---

# 首先通过用户类查找到id为1的用户,然后通过外键属性反向查找到绑定这个用户的所有文章
mm = session.query(User).filter(User.id == 1).first()
print(mm.article)

反过来也是同理

# 先查到文章,通过外键属性,查到文章绑定的是哪个用户
a = session.query(Article).filter(Article.id == 1).first()
print(a.user)

多对多的查询

# 这里查询到所有老师,再拿到第一个数据.也就是王老师
t1 = session.query(Teacher).first()

# 通过王老师的外键classes查询到其绑定的班级
print(t1.classes)

# 输出为 [一班, 二班, 三班]

数据的更新

# 更新数据,首先找到id为1的老师,通过update更新名字为刘老师
t1 = session.query(Teacher).filter(Teacher.id==1).update({'name':'刘老师'})

修改已有数据（一）

• 增加一个老师也去教一班

# 目的是创建一个新的老师,也去教一班
# 创建一个新的老师
t3 = Teacher(name='3老师')
# 查到一班对象
c1 = session.query(Classes).filter(Classes.id == 1).first()
# 将老师指向一班
t3.classes = [c1]
# 提交数据
session.add(t3)

修改已有数据（二）

• 取消一个老师教的所有班级

# 查找到取消的老师
t1 = session.query(Teacher).filter(Teacher.name == '1老师').first()

# 通过外键属性指向空列表来取消班级
t1.classes = []

修改已有数据（三）

• 为老师增加一个班级

# 找到要为哪个老师增加一个班级
t1 = session.query(Teacher).filter(Teacher.name == '3老师').first()

# 查到要增加的那个班级
c1 = session.query(Classes).filter(Classes.name == '1班').first()

# 为老师增加班级
t1.classes += [c1]

修改已有数据（三）

• 为老师取消一个班级

# 查到要取消的班级
c1 = session.query(Classes).filter(Classes.name == '2班').first()

# 通过外键指定为空进行取消
c1.teacher = []

---