EuanSu's Blog

外观

SQLAlchemy 数据库操作

1205字约4分钟

PythonSQLAlchemy

2024-03-06

SQLAlchemy 数据库操作

SQLAlchemy 基础操作

连接数据库

from sqlalchemy import create_engine
# 连接数据库
engine = create_engine("mysql://euansu:euansu123456%24%23%40%21@127.0.0.1:3306/testdb")
conn = engine.connect()

查询数据库

from sqlalchemy import text
# 查询数据库，这里需要使用sqlalchemy的text()封装原生SQL
query = text("show tables")
result_set = conn.execute(query)
print(result_set.all())

执行如上内容会报如下错误，这是因为没有安装连接 mysql 的数据库驱动。

image-20241010110347358

使用 pip 直接安装 mysqlclient 第三方包，这是 sqlalchemy 连接 mysql 的默认驱动。

image-20241010110418163

安装完成后，再执行代码，就能够正常获取到SQL语句的执行结果，如下所示。

image-20241010110438042

程序结束后，需要关闭连接并销毁引擎。

# 关闭连接
conn.close()
# 销毁引擎
engine.dispose()

完整代码如下所示：

from sqlalchemy import create_engine, text


# 连接数据库
engine = create_engine("mysql://euansu:euansu123456%24%23%40%21@127.0.0.1:3306/testdb")
conn = engine.connect()

# 查询数据库
query = text("show tables")
result_set = conn.execute(query)
print(result_set.all())

# 关闭连接
conn.close()
# 销毁引擎
engine.dispose()

SQLAlchemy 创建表与增加记录

如下代码演示了 SQLAlchemy 创建表的操作：

  • 连接数据库

    engine = sqlalchemy.create_engine("mysql://euansu:euansu123456%24%23%40%21@127.0.0.1:3306/testdb")

    如上代码创建了一个数据库引擎，连接到位于 127.0.0.1 主机上 3306 端口的 testdb 的 mysql 数据库。

  • 定义元数据

    meta = sqlalchemy.MetaData()

    MetaData 是一个容器，用于存储数据库结构的信息，如表、列等。

  • 定义 ORM 表结构

    person = sqlalchemy.Table(
    "sqlalchemy_person", meta,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("name", sqlalchemy.String(255), nullable=False),
    sqlalchemy.Column("birthday", sqlalchemy.Date, nullable=False),
)

    这里定义了一个名为 sqlalchemy_person 的表，该表包含三个列：

    • id：整数类型，主键。
    • name：字符串类型，最大长度为255，不能为空。
    • birthday：日期类型，不能为空。

  • 创建表

    meta.create_all(engine)

    这行代码在数据库中创建表 sqlalchemy_person，如果表已存在，则不会重复创建。

完整代码如下所示：

import sqlalchemy

# 连接数据库
engine = sqlalchemy.create_engine("mysql://euansu:euansu123456%24%23%40%21@127.0.0.1:3306/testdb")


meta = sqlalchemy.MetaData()

# ORM
person = sqlalchemy.Table(
    "sqlalchemy_person", meta,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("name", sqlalchemy.String(255), nullable=False),
    sqlalchemy.Column("birthday", sqlalchemy.Date, nullable=False),
)

# 创建表
meta.create_all(engine)

image-20241010174638009

import sqlalchemy
from sqlalchemy_create_table import person
from datetime import date

# 连接数据库
engine = create_engine("mysql://euansu:euansu123456%24%23%40%21@127.0.0.1:3306/testdb")
# 插入一条记录
insert_tome = person.insert().values(name="南歌EuanSu", birthday=date(2024,10,2))
# 执行插入操作
with engine.connect() as connection:
    result = connection.execute(insert_tome)
    print(result.inserted_primary_key)
    connection.commit()

执行如上代码后，使用数据库连接工具进行查询，能够获取到最新插入的数据记录。

image-20241010181353962

import sqlalchemy
from sqlalchemy_create_table import person
from datetime import date

# 连接数据库
engine = create_engine("mysql://euansu:euansu123456%24%23%40%21@127.0.0.1:3306/testdb")
# 创建一个插入对象
person_insert = person.insert()
# 执行插入操作
with engine.connect() as connection:
    connection.execute(person_insert, [
        {"name":"Jack", "birthday": date(2024,10,2)},
        {"name": "Marry", "birthday": date(2024, 10, 3)},
        {"name": "Ben", "birthday": date(2024, 10, 4)},
    ])
    connection.commit()

执行如上代码后，使用数据库连接工具进行查询，能够获取到插入的多条记录。

image-20241010181851035

SQLAlchemy 查询表

原生 sql 查询

from sqlalchemy import create_engine, text


# 连接数据库
engine = create_engine("mysql://euansu:euansu123456%24%23%40%21@127.0.0.1:3306/testdb")
conn = engine.connect()

# 查询数据库
query = text("select * from sqlalchemy_person")
result_set = conn.execute(query)
print(result_set.all())

# 关闭连接
conn.close()
# 销毁引擎
engine.dispose()

执行如上代码后，结果如下：

image-20241031104523504

使用工具连接数据库，查询返回的结果相同。

image-20241031102814264

SQLAlchemy ORM 查询

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

# 定义基础类
Base = declarative_base()


# 定义模型类
class Person(Base):
    __tablename__ = 'sqlalchemy_person'

    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
    name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
    birthday = sqlalchemy.Column(sqlalchemy.Date, nullable=False)


# 创建数据库引擎
engine = create_engine("mysql://euansu:euansu123456%24%23%40%21@127.0.0.1:3306/testdb")

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 查询所有记录
print("查询所有记录")
people = session.query(Person).all()
for person in people:
    print(f"ID: {person.id}, Name: {person.name}, Birthday: {person.birthday}")

print("查询特定条件的记录")
# 查询特定条件的记录
specific_people = session.query(Person).filter(Person.name == "Jack").all()
for person in specific_people:
    print(f"ID: {person.id}, Name: {person.name}, Birthday: {person.birthday}")

# 关闭会话
session.close()

执行如上代码，结果如下所示：

image-20241031104738359

SQLAlchemy 报错记录

TLS/SSL error

报错如下所示，这是因为一些版本的 MySQLdb 对 SSL 证书的验证更为严格，特别是在 Windows 环境下，如果服务器证书不在本地信任链中，MySQLdb 会直接报错，pymysql 对 SSL 证书的处理更为宽松。

image-20241031104041070

使用 pip，安装 pymysql。

pip install pymysql

修改连接串为如下样式。

engine = create_engine("mysql+pymysql://euansu:euansu123456%24%23%40%21@127.0.0.1:3306/testdb")

再次执行对应的代码，就能够正常进行连接请求和查询。

image-20241031104332261