• SQLAchemy 常用操作

    文章目录

    SQLAchemy

    1、 ORM框架

    什么是ORM？

    • 关系对象映射

      类   -> 表
对象 -> 记录（一行数据）

      • 1
      • 2

    • 当有了对应关系之后，不再需要填写SQL语句，取而代之的是操作：类、对象

    • python下常见的ORM有django orm、SQLAlchemy和peewee

    概念：

    • db first：根据数据库的表生成类
    • code first：根据类创建数据库表

    2、 快速认识

    2.1 运行流程

    2.2 基本使用

    在models.py文件中

    创建表和删除表

    #!/usr/bin/env python
# -*- coding: UTF-8 -*-
# @author: A.L.Kun
# @file : models.py
# @time : 2022/6/8 0:00
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, INTEGER, String

Base = declarative_base()  # 创建一个基类
# 数据库连接
engine = create_engine(
    "mysql+pymysql://root:qwe123@127.0.0.1:3306/flask1?charset=utf8",  # 数据库url
    max_overflow=0,  # 超过连接池大小外最多创建的连接
    pool_size=5,  # 连接池大小
    pool_timeout=30,  # 池中没有线程最多等待时间，否则报错
    pool_recycle=-1  # 多久之后线程池中的线程进行一次连接重置
)  # 默认带了连接池

# 创建表
class Users(Base):
    __tablename__ = "users1"  # 设置表名
    id = Column(INTEGER, primary_key=True, autoincrement=True)
    name = Column(String(32))
    extra = Column(String(32))


# # 创建表，如果表已经存在，则不会再创建
# Base.metadata.create_all(engine)
# # 删除表
# Base.metadata.drop_all(engine)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30

    给表添加信息

    在其他py文件中

    #!/usr/bin/env python
# -*- coding: UTF-8 -*-
# @author: A.L.Kun
# @file : test.py
# @time : 2022/6/8 0:13
from sqlalchemy.orm import sessionmaker
import models
session = sessionmaker(bind=models.engine)()  # 创建连接

obj = models.Users(name="kun", extra="hello")
session.add(obj)  # 将数据添加到表中

session.commit()  # 提交事务

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13

    2.3 连接数据库

    SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：

    MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
   
pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
   
MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
   
cx_Oracle
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
   
更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13

    2.4 执行原生的SQL语句

    # 数据库连接
engine = create_engine("mysql+pymysql://root:qwe123@127.0.0.1:3306/flask1?charset=utf8")  # 默认带了连接池
cur = engine.execute("SELECT * FROM user1")
print(cur.fetchall())

    • 1
    • 2
    • 3
    • 4

    3、 创建多表

    #!/usr/bin/env python
# -*- coding: UTF-8 -*-
# @author: A.L.Kun
# @file : models.py
# @time : 2022/6/8 0:00
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import (
    					create_engine, Column, INTEGER, String,
                        DATETIME,  # 创建存储时间的列
                        ForeignKey,  # 外键约束
                        Index,  # 创建索引
                        UniqueConstraint,  # 创建联合唯一索引
                        )
import datetime

Base = declarative_base()  # 创建一个基类

# 数据库连接
engine = create_engine(
    "mysql+pymysql://root:qwe123@127.0.0.1:3306/flask1?charset=utf8",  # 数据库url
)  # 默认带了连接池


# 创建班级表
class Classes(Base):
    __tablename__ = "classes"  # 设置表名
    id = Column(INTEGER, primary_key=True, autoincrement=True)
    name = Column(String(32), nullable=False, unique=True)


# 学生表，其和班级表是一对多的关系
class Student(Base):
    __tablename__ = "student"
    id = Column(INTEGER, primary_key=True, autoincrement=True)
    username = Column(String(32), nullable=False, unique=True)
    password = Column(String(64), unique=False)
    ctime = Column(DATETIME, default=datetime.datetime.now)  # 创建时间，注意now后面不需要加括号，不然只会记录初始化时间，而不是添加数据的时间
    class_id = Column(INTEGER, ForeignKey("classes.id"))  # 外键约束


# 学生的爱好
class Hobby(Base):
    __tablename__ = "hobby"
    id = Column(INTEGER, primary_key=True)
    caption = Column(String(50), default="篮球")
    from sqlalchemy.orm import relationship
    cls = relationship("Classes", secondary="S_H", backref="stus")  # 创建多表关联，通过cls键进行关联，secondary其为将两张表关联起来的表，backref，反向生成


# 多对多的表，即把学生和爱好连接起来
class S_H(Base):
    __tablename__ = "s2h"
    id = Column(INTEGER, primary_key=True, autoincrement=True)
    stu_id = Column(INTEGER, ForeignKey("student.id"))
    hob_id = Column(INTEGER, ForeignKey("hobby.id"))
    __table_args__ = (
        UniqueConstraint("stu_id", "hob_id", name="uin_stu_hob"),  # 给两列创建联合唯一索引
        # Index("in_stu_hob", "stu_id", "extra")  # 创建一个索引
    )


if __name__ == '__main__':
    # 创建表，如果表已经存在，则不会再创建
    Base.metadata.create_all(engine)
    # # 删除表
    # Base.metadata.drop_all(engine)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66

    4、 增删改查

    #!/usr/bin/env python
# -*- coding: UTF-8 -*-
# @author: A.L.Kun
# @file : test.py
# @time : 2022/6/8 0:13
from sqlalchemy.orm import sessionmaker
import models
from sqlalchemy import text

session = sessionmaker(bind=models.engine)()  # 连接数据库


def add_data():
    """添加数据"""
    # 单条增加
    cls = models.Classes(name='1班')
    session.add(cls)
    # 多条增加
    objs = [
        models.Classes(name='2班'),
        models.Classes(name='3班')
    ]
    session.add_all(objs)


def del_data():
    """删除数据"""
    session.query(models.Classes).filter(models.Classes.id > 2).delete()  # 指定条件删除数据


def find_data():
    """查询数据"""
    ret = session.query(models.Classes).all()  # 获取全部数据
    ret1 = session.query(models.Classes.name).all()  # 获取名字数据
    ret2 = session.query(models.Classes).filter(models.Classes.name == "2班009009").all()  # 获取2班的数据
    ret3 = session.query(models.Classes).filter_by(name="2班009009").first()  # 获取二班的数据
    ret4 = session.query(models.Classes).filter(text("id<:value and name=:name")).params(
        # :value 和 :name 为一个占位符，使用order_by进行排序作用
        value=224,
        name="field"
    ).order_by(models.Classes.id)

    ret5 = session.query(models.Classes).from_statement(text("SELECT * FROM classes WHERE name=:name")).params(
        name="ed"  # 构造SQL语句
    )
    print(ret5)
    # 使用子查询
    ret6 = session.query(models.Classes).filter(models.Classes.id.in_(
        session.query(models.Classes.id).filter_by(name="eee")  # 其为子查询
    )).all()
    ret7 = session.query()
    print(ret)
    
    """联表操作"""
    # 获取学生信息，以及其班级信息
    # 方法一
    objs = session.query(models.Student.id, models.Student.username, models.Classes.name).join(models.Classes, isouter=True).all()
    # 方法二
    objs1 = session.query(models.Student).all()
    """
    # 在models.Student 末尾添加这两行代码，进行内部自动根据外键关联数据库
    from sqlalchemy.orm import relationship
    cls = relationship("Classes", backref="stus")  # backref 其为反向生成，即等于是在 Classes 中添加 stus = relationship("Student")
    """
    for item in objs1:
        print(
            item.id,
            item.username,
            item.cls,  # 这个为关联的数据库，可以访问其内部的内容
        )


def up_data():
    """修改数据"""
    session.query(models.Classes).filter(models.Classes.id >= 1).update({
        models.Classes.name: models.Classes.name + "009",
        # 也可以 "name": models.Classes.name + "009",
    },
        synchronize_session=False  # 在内部不进行运算，而是直接拼接，如果值为 "evaluate" 的话，就会进行运算
    )


# add_data()
# del_data()
# find_data()
# up_data()
session.commit()  # 提交事务
session.close()

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88

    5、 常用操作

    #　条件
# 通过一个条件来查询
ret = session.query(Users).filter_by(name='alex').all() 
# id > 1 && name == "eric"
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
# 1 <= id <= 3 && name == "eric" 
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
# id in (1, 3, 4)
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
# id not in (1, 3, 4)
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
# SELECT * FROM USERS WHERE id IN (SELECT id FROM USERS WHERE name = "eric")
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
from sqlalchemy import and_, or_
# id > 3 && name == "eric"
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
# id < 2 || name == "eric"
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
# id < 2 || (name == "eric" && id > 3) || extra
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'eric', Users.id > 3),
        Users.extra != ""
    )).all()


# 通配符
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()

# 限制
ret = session.query(Users)[1:2]

# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(
    Users.name.desc(),
    Users.id.asc()  # 
).all()

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()

ret = session.query(Person).join(Favor).all()

ret = session.query(Person).join(Favor, isouter=True).all()


# 组合
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
  • 相关阅读:
    C++实验1：图书管理系统1.0——组合依赖
    Unix后记&寻找Shen Lin
    跨境电商平台自养号测评防关联技巧！
    git push 出错error: failed to push some refs to ....
    损失函数总结（五）：PoissonNLLLoss、GaussianNLLLoss
    空间变形网络——STN
    回溯算法中排列、组合问题中的used数组怎么使用
    软件工程师参加技术峰会，带来的价值，你想象不到！
    【Redis】主从复制
    Android 远程调用服务之 AIDL
  • 原文地址：https://blog.csdn.net/qq_62789540/article/details/125552046