SQLAlchemy的使用指南

xiaoyu_01 · 2025 年1 月 16 日 10:45

SQLAlchemy使用指南

参考帖子：SQLAlchemy使用教程-CSDN博客

简介

ORM

ORM（obect relational mapping ：对象关系映射）可以绕过SQL语句，把数据库的table映射为编程语言的class类，可以直接使用编程语言的对象模型操作数据库，而不使用SQL语句。

ORM把表映射成类，把行作为实例 ，把字段作为属性 ，在执行对象操作时最终会把对象的操作转为数据库原生语句
SQL Alchemy的优点

易用性：减少SQL语句的使用，使代码，模型更加直观，清晰

性能损耗小：设计灵活，可移植性强
SQL Alchemy分为两部分

ORM对象映射和核心的SQL expression

安装和使用

安装SQLAlchemy

使用SQLAlchemy时需要先给python安装MYSQL驱动，由于MYSQL不支持Python3，因此需要使用PyMySQL和SQLAlchemy交互
```
pip install pymysql
pip install sqlalchemy
```
使用SQLAlchemy链接MySQL数据库
```
from sqlalchemy import crreate_engine
engine = create_engine("mysql+pymysql://user:password@localhost:3306/database",echo=True)
#参数格式
engine = create_engine("数据库类型+数据库驱动://数据库用户名:数据库密码@IP地址:端口号/数据库?编码...", 其它参数)
```
*create_engine()*返回的是Engine的一个实例，代表了操作数据库的核心接口，处理数据库和数据库的API ，初次调用create_engine()并不会真正链接数据库，只有在真正执行一条命令时才会尝试建立连接，目的是为了节省资源

映射声明：

当使用ORM时，其配置过程主要分为两个部分：
- 描述处理的数据库表的信息
- 将python类映射到这些表上，他们在SQLAlchemy中一起完成，被称为Declarative
使用Declarative参与的ORM映射的类需要被定义为一个指定基类的子类，这个基类含有ORM映射中相关类和表的信息，这样的基类称为delarative base class，这个基类可以通过declarative_base来创建
```
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
```

`SQLAlchemy`常用数据类型

Interger：整型，映射到数据库中是Int类型

Float：浮点类型，float

Double：String，Boolean

Decimal：定点类型，专门解决浮点类型精度丢失的问题而设定，Deimal需要传入两个参数，第一个参数标记该字段能存储多少位数，第二个参数表示小数点后有多少小数位

Enum：枚举类型

Date：日期类型，年月日

Date Time：时间类型，年月日时分毫秒

Time：时间类型，时分秒

Text：长字符串，可以存储6W多个字符 text

LongText：长文本类型，LONGTEXT

创建类

前面已经介绍了如何创建一个基类及常用的数据类型，可以基于这个基类创建自定义的类

解读

首先创建基类，然后继承，这个其实是为了让系统知道你这个类是要准备与数据库交互
在类里指定类变量：__tablename__的值，用来指定表，因为ORM操作本质上是为了让你以操作类的方式来完成数据库操作，而该库则实现 类操作->数据库SQL语句操作 所以在创建类 ->（映射）表时，会要求提供足够的信息来完成操作的转换
- 表名：作为类变量的__tablename__
- 该表中各个属性：
  - id = Column(Integer, primary_key=True)
  - 左侧变量名需要与表中字段名保持一致，右侧则是数据类型，特征等也需保持一致
类返回字符串

使用__repr__函数定义该类返回的字符串内容

from sqlalchemy import Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base  # ORM(对象关系映射)的基类
Base = declarative_base()

# 继承Base基类
class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String(20), default=None, nullable=False, comment="用户姓名")
    phone = Column(String(20), default=None, nullable=False, comment="电话")
    country = Column(Integer, default=0, nullable=False, comment="国家")

    def __repr__(self):
        Name = self.course_name
        Phone = self.teacher_name
        Country = self.class_times
        return f"User: name: {Name}, phone: {Phone}, country: {Country}"

常用属性
primary_key 是否为主键
unique 是否唯一
index 如果为True，为该列创建索引，提高查询效率
nullable 是否允许为空
default 默认值
name 在数据表中的字段映射
autoincrement 是否自动增长
onupdate 更新时执行的函数
comment 字段描述

创建模式

查看表信息

uSER.__table__

创建表

Table对象是一个更大家庭：metadata，metadata是与数据库打交道的一个接口，创建表需要使用metadata发出CREATE TABLE的命令

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

Base = declarative_base()
engine = create_engine("mysql+pymysql://root:root@localhost:3306/apptest", echo=True)


class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String(20), default=None, nullable=False, comment="用户姓名")
    phone = Column(String(20), default=None, nullable=False, comment="电话")
    country = Column(Integer, default=0, nullable=False, comment="国家")

    def __repr__(self):
        Name = self.course_name
        Phone = self.teacher_name
        Country = self.class_times
        return f"User: name: {Name}, phone: {Phone}, country: {Country}"
   

Base.metadata.create_all(engine)	# 通过此语句创建表

创建实例

NewUser = User(name="Jason", phone="12345678910", country="China")
创建会话

要真正使用 类对象操作数据表，还需要一个Session对象，ORM对数据库的入口即Session

session创建实例
```
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("mysql+pymysql://root:root@localhost:3306/apptest", echo=True)
Session = sessionmaker(bind=engine)
session = Session()
```
或者
```
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
Session = sessionmaker()
engine = create_engine("mysql+pymysql://root:root@localhost:3306/apptest", echo=True)
Session.configure(bind=engine)
session = Session()
```
经过上述操作，Session已经与Mysql数据库的Engine关联了。当一个Session被首次使用时，它会从Engine所维护的连接池中取出一个连接来操作数据库。这个连接在应用有所更改或关闭Session时会被释放

SQLAlchemy的Session是用于管理数据库操作的一个像容器一样工厂对象。Session工厂对象中提供query()，add(),add_all(),commit(),delete(),flush(),rollback(),close()等方法

`SQLAlchemy`对`Mysql`数据库的基本操作

添加对象

SQLAlchemy可以通过Session提供的add()方法，将数据存入数据库

from sqlalchemy import Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
engine = create_engine("mysql+pymysql://root:root@localhost:3306/apptest", echo=True)
Session = sessionmaker(bind=engine)
session = Session()


class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String(20), default=None, nullable=False, comment="用户姓名")
    phone = Column(String(20), default=None, nullable=False, comment="电话")
    country = Column(Integer, default=0, nullable=False, comment="国家")

    def __repr__(self):
        Name = self.course_name
        Phone = self.teacher_name
        Country = self.class_times
        return f"User: name: {Name}, phone: {Phone}, country: {Country}"
   

Base.metadata.create_all(engine)
NewUser = User(name="python", phone="12345678910", country="China")
session.add(NewUser)
session.commit()	# 需要调用commit()方法提交事务。

也可以通过add_all()方法添加多个对象，传入参数为一个包含多个对象的列表

查询对象

通过Session的query()方法查询数据，返回对象中的__repr()函数的返回值

query_result = session.query(User).all()
for result in query_result:
    print(f"查询结果为: {result}")

# [out]查询结果为: User: name: python, phone: 12345678910, country: China

添加查询条件

使用query()方法中的filter()方法实现与WHERE等同的效果，示例代码如下：

results = session.query(User).filter(User.name=='python')
for result in results:
    print(f"查询结果为: {result}")

返回对象

可以通过在查询语句后添加first(),all()等返回对应的查询对象

更新对象

在SQLAlchemy中，更改值得方式是，找到目标得类对象，然后通过字段映射到类对象对应得属性字段，对属性字段重新赋值。再执行Session得add(),commit()方法实现更新
```
result = session.query(User).filter(User.name=="python").first()
result.name = "Cython"
session.add(result)
session.commit()
```

删除对象

SQLAlchemy中得删除对象方法比较简单

result = session.query(User).filter(User.name=="python").first()
session.delete(result)
session.commit()

高级查询

在SQLAlchemy中通过filter()方法中得操作符实现高级查询
- like操作符
  
  query_result = session.query(User).filter(User.name.like("%py%"))
  
  该方法与MySQL中得like模糊查询用法基本一致
- and操作符
  
  方法一：使用and()
  
  query_result = session.query.filter(and_(User.name == 'python', User.id > 1))
  
  方法二：在filter()中设置多个表达式
  
  query_result = session.query.filter(User.name == 'python', User.id > 1)
  
  方法三：使用多个filter()
  
  query_result = session.query.filter(User.name == 'python').filter(User.id > 1)
- or操作符
  
  query_result = session.query.filter(or_(User.name == 'python', User.id > 1))
- in操作符
  
  query_result = session.query.filter(User.name.in_(["python", "C"]))
- not in 操作符
  
  query_result = session.query.filter(~User.name.in_(["python", "C"]))

嵌入使用SQL语句

在SQLAlchemy中导出text方法，可以通过text(SQL语句)嵌入使用SQL语句

在查询中嵌入SQL语句

from sqlalchemy import text
...
...
...
query_result = session.query(User).filter(text("name='python'")).all()

通过Engine对象执行SQL语句

engine = create_engine("mysql+pymysql://root:root@localhost:3306/apptest", echo=True)
        with engine.connect() as conn:
            conn.execute(text(sql))

保存或更新用户信息示例

在WEB开发中，通常涉及数据得保存和更新，下面得例子通过嵌入使用SQL语句，实现数据（字典类型）得保存，当primary_key或uniqu字段重复时，会自动实现数据得更新

def SaveData(table, data: dict) -> (bool, str):
    sqlSet = ','.join([key+'='+(str(value) if type(value) == int else f"'{value}'")
                       for key, value in data.__iter__()])
    sqlInsert = f"insert into {table} set {sqlSet} on duplicate key update {sqlSet}"
    try:
        engine = get_engine()	# 在这里我将创建Engine的方法封装成get_engine()方法
        with engine.connect() as conn:
            conn.execute(text(sqlInsert))
        return True, "数据保存/更新成功！"
    except Exception as e:
        return False, f"数据保存/更新失败！err:{e}"

总结

SQLAlchemy本意就是使编程里的对象操作等效为SQL操作，其中的转换由pip install sqlalchemy库完成，而SQLAlchemy库的使用虽步骤繁多，却也是为了让底层转换对象操作为SQL语句所需最基本的信息。

第一步：注册SQL库的核心【链接核心】engine，创建和数据库的连接：

from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://user:password@localhost:3306/database",echo=True)
#参数格式
engine = create_engine("数据库类型+数据库驱动://数据库用户名:数据库密码@IP地址:端口号/数据库?编码...", 其它参数)

第二步，创建一个对象类和具体的表映射，要做到这一步，有两个要求：第一个是要让系统知道这个类是映射类

指定映射类：创建一个SQLAlchemy的基类，然后继承拓展，实际上这个映射类本身就是SQLAlchemy提供的一个类：

from sqlalchemy.ext.declarative import declarative_base  # ORM(对象关系映射)的基类
Base = declarative_base()

然后继承这个类，然后对类的属性进行填充，必要填充：类变量__tablename__ 用来指定表名，然后是这个表的字段信息

class User_model(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String(20), default=None, nullable=False, comment="用户姓名")
    phone = Column(String(20), default=None, nullable=False, comment="电话")
    country = Column(Integer, default=0, nullable=False, comment="国家")

第三步：要真正通过类对象操作数据库，需要创建Session对象，ORM对数据库的入口即为Session，可以理解为前面的engine啥的都是静态的配置，只有创建Session才是真正驱动，驱动需要使用之前得engine：

Session = sessionmaker(bind=engine)
session = Session()
或
Session.configure(bind=engine)
session = Session()

第四步：

然后是使用session进行操作：

#往某个表里添加数据，参数是某个映射类的对象
NewUser = User(name="python", phone="12345678910", country="China")
NewUser1 = User(name="python1", phone="12345678911", country="China1")
session.add(NewUser)
#或者使用add_all()方法一次添加多个数据，将这些数据写出列表传入
session.add_all([NewUser.NewUser1])
session.commit()	# 需要调用commit()方法提交事务

#查询对象:参数是某个映射类，即为对某张表进行查询，.all()表示获取全部查询结果
session.query(User).all()
#添加查询条件：filter()  和 WHERE等价,其实就是在查询条件里，如果存在多个条件则用逗号，隔开，或者使用 and_(条件一,条件二)  如果存在多个条件集，则用逗号，隔开， 或逻辑：使用or_()：or_(User.name=='aa',User.age==10)   in 操作符： User.name.in_(['a','b'])
session.query(User).filter(User.name=='python')
# .all() 和 .first() 来控制返回得对象是全部，还是第一条
#更新对象：这里得做法是先找到这个对象，然后修改该对象某些值，然后再进行add() 操作，这时会因为主键已存在而转为更新操作，返回得是某个类得对象或者对象列表，取决于用的是.all()还是.first()
result = session.query(User).filter(User.name="Python").first()
result.name = "Cythton"
session.add(result)
session.commit()
#删除对象，实际上无论是添加，删除，更新，丢进去得都是一个映射类对象，这个映射类对象得来源可以是你前面搜索获取得，也可以是自己构建得映射类对象，他主要是靠数据主键来进行识别，操作
session.delete(result)
session.commit()
#高级查询
#模糊查询LIKE 模糊查询语句使用 User.name 来指定模糊的字段，然后使用关键字like  然后括号里填写正则表达式就行
query_result = session.query(User).filter(User.name.like("%py%"))