python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > SQLAlchemy数据库

详解如何使用SQLAlchemy连接数据库

作者:Hunter后端

这篇文章主要为大家详细介绍了如何使用 SQLAlchemy 连接数据库、建立模型、操作表、以及查询操作表数据等内容,感兴趣的小伙伴可以跟随小编一起学习一下

相对于 Django 来说,Flask 并没有 Django 中自带的那种 ORM 框架,但是我们可以利用第三方的 ORM 框架来进行操作,比如我们这里介绍的 SQLAlchemy。

接下来这一篇笔记将会介绍如何使用 SQLAlchemy 连接数据库、建立模型、操作表、以及查询操作表数据等内容。

0、模块安装

对于 SQLAlchemy 来说,它还需要使用 pymysql 这个库来进行数据库的操作,所以这里我们安装的模块如下:

pip3 install sqlalchemy==2.0.19
pip3 install pymysql==1.1.0

1、数据库的连接与使用

以下是使用 SQLAlchemy 对数据库进行连接以及一个简单的查询示例:

from sqlalchemy import create_engine, text

engine = create_engine("mysql+pymysql://root:123456@192.168.1.5:3306/db_test?charset=utf8")

with engine.connect() as conn:
    sql = "select id, name from users"
    result = conn.execute(text(sql))
    print(result.all())

# [(1, 'admin'), (2, 'user_1')]

在这里,我们预设在 db_test 这个库下有一张表名为 users 的表,然后使用原生的 SQL 语句进行了查询和打印操作,这个操作其实就跟直接使用 pymysql 模块操作数据库没有区别。

如果是插入、更新语句的话,在后面还需要加上 conn.commit() 提交操作。

2、ORM 模型的建立与 ORM Session 的使用

接下来我们定义一个 ORM 模型,其实这里的 ORM 模型就跟 Django 的模型类似了,不过在语法上有一些区别:

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import Column, Integer, String, Text, DateTime, func

class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), unique=True, nullable=False, comment="用户名")
    email = Column(String(120), default="", comment="邮箱")
    remark = Column(Text, default="", comment="备注信息")
    created_time = Column(DateTime, server_default=func.now(), comment="创建时间")
    updated_time = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment="修改时间")

在这里,我们定义了一个 Base 类,继承于 DeclarativeBaseDeclarativeBase 是用于定义 ORM 模型的基类,提供了一些方便的功能,使得使用 ORM 进行数据库操作更加简单和直观,比如提供了一些 query 的方法,这个我们后面再介绍。

对于 User 这个 class,一个类似于 Django 的 ORM 的 SQLAlchemy 的 ORM 模型就搭建完成了。

其中,IntegerStringTextDateTime 这些都是各自对应到数据库的字段

对于定义的 Column(),有很多字段的属性,比如 primary_key 主键,auto_increment 自增,default 默认值,nullable 是否允许为 Null 等

在这里,created_time 和 updated_time 实现的是 Django 的日期时间字段里 auto_now_add 和 auto_now 属性,即为创建或者修改的时候值为当前时间

接下来介绍一下 ORM Session 的定义,以下是一个示例:

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

engine = create_engine("mysql+pymysql://root:123456@192.168.1.5:3306/db_test?charset=utf8")

db_session = scoped_session(
    sessionmaker(
        autoflush=False,
        bind=engine
    )
)

上面的操作中,sessionmaker() 通过 engine 创建了一个会话工厂对象,然后通过 scoped_session() 函数创建了一个 scoped_session 对象

我们可以直接使用上面定义出来的 db_session 对数据库进行操作,也可以对其进行实例化后操作:

db_session.do_something()
// 直接用 db_session 操作数据库

session = db_session()
session.db_something()
// 用 db_session 实例化结果操作数据库

他们的使用对象虽然不同,但在功能上都提供了线程本地的会话对象,可以在多线程应用程序中安全地使用。

3、表操作

下面介绍一下 SQLAlchemy 如何操作表。

1. 创建表

在前面我们定义了 engine 和 Base 基类和 User 这个表的 class 之后,我们可以使用下面的方式创建全部定义的表:

Base.metadata.create_all(bind=engine)

执行上面的语句之后,User 这个 class 对应的表就会被创建到数据库中。

可以把上面这个语句添加到项目的启动步骤中,因为这个操作会创建库里没有的表,已有的就不会重复执行了。

2. 修改表结构

SQLAlchemy 不支持通过函数的形式直接修改表结构,但是可以执行原生的 SQL 来进行 ALTER TABLE 的操作。

或者通过下面的删除表操作删除,再进行 create_all() 操作,但是这样的话,原表的数据就不存在了。

3. 删除表

删除表的操作与创建的方式类似,如下:

Base.metadata.drop_all(bind=engine)

这样,所有继承了 Base 基类的表都会被删除

如果要执行删除单张表,可以使用 Table 的 drop() 函数:

from sqlalchemy import Table, MetaData

meta_data = MetaData()
table_name = "users"
user_table = Table(table_name, meta_data, autoload_with=engine)
user_table.drop(bind=engine)

4、创建表数据

1. 创建单条数据

u = User(name="admin", email="120@qq.com")
db_session.add(u)
db_session.commit()

使用 add() 添加,然后进行 commit 操作

2. 创建多条数据

u1 = User(name="user_1", email="user1@qq.com")
u2 = User(name="user_2", email="user2@qq.com")
db_session.add_all([u1, u2])
db_session.commit()

批量创建使用 add_all() 函数。

5、查询表数据

1. 根据主键 id 查询数据

user = db_session.get(User, 1)

返回的 user 就是一个前面我们定义好的 User 对象

2. 条件查询

条件查询,可以有两个操作,一个是 where(),一个是 filter(),这两者在效果上是相同的,都是作用于条件查询。

比如,我们要查询 name 字段的值为 "admin" 以及 id 字段的值为 1 的数据,且返回 id, name, email 字段,可以使用 query() 来进行字段限制,如下操作:

query = db_session.query(User.id, User.name, User.email).filter(User.name == "admin").filter(User.id == 1)

query = db_session.query(User.id, User.name, User.email).filter(User.name == "admin").where(User.id == 1)

query = db_session.query(User.id, User.name, User.email).where(User.name == "admin").filter(User.id == 1)

query = db_session.query(User.id, User.name, User.email).where(User.name == "admin").where(User.id == 1)

如果我们想要一条数据,可以使用 first(),如果想要符合条件的全部数据,可以使用 all()

query.first()
# (1, 'admin', '120@qq.com')

query.all()
# [(1, 'admin', '120@qq.com')]

除此之外,我们还可以将 db_session.query_property() 赋值给 Base.query,后面就可以直接通过 User 进行查询操作

Base.query = db_session.query_property()

User.query.filter(User.name == "admin").filter(User.id == 1)

然后通过 first() 或者 all() 返回的就是定义的 User 的对象

6、更新表数据

对于我们获取到的 ORM 模型实例,比如我们在前面通过主键 id 获取到的数据,或者在 db_session.query() 中不指定字段,直接指定模型获取到的数据,我们可以直接对其字段进行修改,然后 commit

user_1 = db_session.get(User, 1)
user_1.email = "999@qq.com"
db_session.add(user_1)

user_2 = db_session.query(User).filter(User.id == 2).first()
user_2.email = "888@qq.com"
db_session.add(user_2)

db_session.commit()

或者我们前面的 query 属性的方式也可以:

User.query.filter(User.id == 1).update({"email": "19283@qq.com"})
db_session.commit()

7、删除表数据

1. 单条记录删除

user = db_session.get(User, 2)
db_session.delete(user)
db_session.commit()

2. 根据条件批量删除

User.query.filter(User.id == 3).delete()
db_session.commit()

以上就是详解如何使用SQLAlchemy连接数据库的详细内容,更多关于SQLAlchemy数据库的资料请关注脚本之家其它相关文章!

您可能感兴趣的文章:
阅读全文