python：sqlalchemy-快速开始

#

一、声明模型

首先我们定义了模块级结构，这些结构将形成我们将从数据库中查询的结构。这种结构称为 声明性映射，讲白了其实就是SQL表中存在的元数据。

from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import relationship


Base = declarative_base()


class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String(30))
    fullname = Column(String)
    addresses = relationship(
        "Address", back_populates="user", cascade="all, delete-orphan"
    )

    def __repr__(self):
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"


class Address(Base):
    __tablename__ = "address"
    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(Integer, ForeignKey("user.id"), nullable=False)
    user = relationship("User", back_populates="addresses")

    def __repr__(self):
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"

信息注释：

• Column表示数据库表中的列。
• primary_key设置字段为主键
• ForeignKey设置字段为外键
• nullable可为空，意思是可为空值
• relationship提供两个映射类之间的关系
• back_populates是指定和哪个类作映射关系
• cascade级联，可理解意思就是主表数据被删除，关联的子表跟随操作

二、建立连接

from sqlalchemy import create_engine


_SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@192.168.0.110:3306/sqlalchemy_db"

engine = create_engine(
    # echo=True参数表示连接发出的 SQL 将被记录到标准输出
    # future=True是为了方便便我们充分利用sqlalchemy2.0样式用法
    _SQLALCHEMY_DATABASE_URL, echo=True, future=True
)

信息注释：

• 这里就是定义一个连接数据库的变量
• 调用creat_engine方法来连接指定的数据库

三、创建数据表

Base.metadata.create_all(engine)

信息注释：

• 创建存储在此元数据中的所有表。
• 默认情况下有条件，不会尝试重新创建目标数据库中已经存在的表。

注意点：

• 如果你使用的是sqlite数据库，则直接运行代码就会自动根据定义的模型自动创建数据；如果你是mysql数据库，则需要自己先建好表，不然运行时会抛出数据表不存在的问题。

上述代码运行完成，在控制台就会产生一些日志，没有报错就表示成功了：

2022-05-01 13:35:53,251 INFO sqlalchemy.engine.Engine SELECT DATABASE()
2022-05-01 13:35:53,251 INFO sqlalchemy.engine.Engine [raw sql] {}
2022-05-01 13:35:53,255 INFO sqlalchemy.engine.Engine SELECT @@sql_mode
2022-05-01 13:35:53,255 INFO sqlalchemy.engine.Engine [raw sql] {}
2022-05-01 13:35:53,256 INFO sqlalchemy.engine.Engine SELECT @@lower_case_table_names
2022-05-01 13:35:53,257 INFO sqlalchemy.engine.Engine [raw sql] {}
2022-05-01 13:35:53,260 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2022-05-01 13:35:53,260 INFO sqlalchemy.engine.Engine SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = %(table_schema)s AND table_name = %(table_name)s
2022-05-01 13:35:53,260 INFO sqlalchemy.engine.Engine [generated in 0.00014s] {'table_schema': 'sqlalchemy_db', 'table_name': 'user'}
2022-05-01 13:35:53,262 INFO sqlalchemy.engine.Engine SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = %(table_schema)s AND table_name = %(table_name)s
2022-05-01 13:35:53,263 INFO sqlalchemy.engine.Engine [cached since 0.002399s ago] {'table_schema': 'sqlalchemy_db', 'table_name': 'address'}
2022-05-01 13:35:53,265 INFO sqlalchemy.engine.Engine COMMIT

Process finished with exit code 0

#

一、创建对象并持久化

现在准备往数据库中插入数据

• 通过创建User和Address对象的实例来实现这一点，这些对象具有init()声明性映射过程自动建立的方法。
• 使用一个名为Session的对象将它们传递给数据库，该对象利用SessionEngine与数据库进行交互。
• 使用Session.add_all()方法一次添加多个对象，该Session.commit()方法将用于刷新对数据库的任何未决更改，然后提交当前数据库事务，只要使用 ，该事务始终在进行中Session

from sqlalchemy.orm import Session


with Session(engine) as session:
    spongebob = User(
        name="zhangsansan",
        fullname="sansan",
        addresses=[Address(email_address="sansan@sqlalchemy.com")],
    )
    sandy = User(
        name="lixiaoxiao",
        fullname="xiaoxiao",
        addresses=[
            Address(email_address="xiaoxiao@sqlalchemy.com"),
            Address(email_address="dongdong@squirrelpower.com"),
        ],
    )
    patrick = User(name="lidongdong", fullname="dongdong")
    session.add_all([spongebob, sandy, patrick])
    session.commit()

信息注释：

• 这里使用的是上下文管理器的方式来插入数据，数据执行完成后，会自动执行关闭连接操作

上述代码执行完成数据插入后，在控制台就会产生一些日志，没有报错就表示成功了：

2022-05-01 14:43:26,282 INFO sqlalchemy.engine.Engine INSERT INTO user (name, fullname) VALUES (%(name)s, %(fullname)s)
2022-05-01 14:43:26,282 INFO sqlalchemy.engine.Engine [generated in 0.00015s] {'name': 'zhangsansan', 'fullname': 'sansan'}
2022-05-01 14:43:26,285 INFO sqlalchemy.engine.Engine INSERT INTO user (name, fullname) VALUES (%(name)s, %(fullname)s)
2022-05-01 14:43:26,286 INFO sqlalchemy.engine.Engine [cached since 0.003192s ago] {'name': 'lixiaoxiao', 'fullname': 'xiaoxiao'}
2022-05-01 14:43:26,289 INFO sqlalchemy.engine.Engine INSERT INTO user (name, fullname) VALUES (%(name)s, %(fullname)s)
2022-05-01 14:43:26,289 INFO sqlalchemy.engine.Engine [cached since 0.006635s ago] {'name': 'lidongdong', 'fullname': 'dongdong'}
2022-05-01 14:43:26,293 INFO sqlalchemy.engine.Engine INSERT INTO address (email_address, user_id) VALUES (%(email_address)s, %(user_id)s)
2022-05-01 14:43:26,293 INFO sqlalchemy.engine.Engine [generated in 0.00025s] {'email_address': 'sansan@sqlalchemy.com', 'user_id': 1}
2022-05-01 14:43:26,300 INFO sqlalchemy.engine.Engine INSERT INTO address (email_address, user_id) VALUES (%(email_address)s, %(user_id)s)
2022-05-01 14:43:26,300 INFO sqlalchemy.engine.Engine [cached since 0.007286s ago] {'email_address': 'xiaoxiao@sqlalchemy.com', 'user_id': 2}
2022-05-01 14:43:26,303 INFO sqlalchemy.engine.Engine INSERT INTO address (email_address, user_id) VALUES (%(email_address)s, %(user_id)s)
2022-05-01 14:43:26,303 INFO sqlalchemy.engine.Engine [cached since 0.009968s ago] {'email_address': 'dongdong@squirrelpower.com', 'user_id': 2}
2022-05-01 14:43:26,310 INFO sqlalchemy.engine.Engine COMMIT

Process finished with exit code 0

#

一、数据查询-简单的选择

from sqlalchemy.orm import Session
from sqlalchemy import select


session = Session(engine)

stmt = select(User).where(User.name.in_(["zhangsansan", "lidongdong"]))
for user in session.scalars(stmt):
    print(f"运行结果：{user}")

信息注释：

• scalars是将标量筛选应用于结果上

运行结果：

2022-05-01 15:05:28,425 INFO sqlalchemy.engine.Engine SELECT user.id, user.name, user.fullname 
FROM user 
WHERE user.name IN (%(name_1_1)s, %(name_1_2)s)
2022-05-01 15:05:28,425 INFO sqlalchemy.engine.Engine [generated in 0.00068s] {'name_1_1': 'zhangsansan', 'name_1_2': 'lidongdong'}
运行结果：User(id=1, name='zhangsansan', fullname='sansan')
运行结果：User(id=3, name='lidongdong', fullname='dongdong')

从查询结果可以清晰看到符合条件的两条数据显示出来了，而且查询的SQL语句也体现出来了，是不是慢慢的感觉SQL语句的技能也提升了呢

二、数据查询-多表

from sqlalchemy.orm import Session
from sqlalchemy import select


session = Session(engine)

stmt = (
    select(Address).join(
        Address.user
    ).where(
        User.name == "zhangsansan"
    ).where(
        Address.user_id == 1
    )
)
sandy_address = session.scalars(stmt).one()
print(f"运行结果：{sandy_address}")

信息注释：
scalars是将标量筛选应用于结果上，one是拿一条结果

运行结果：

2022-05-01 15:10:13,712 INFO sqlalchemy.engine.Engine SELECT address.id, address.email_address, address.user_id 
FROM address INNER JOIN user ON user.id = address.user_id 
WHERE user.name = %(name_1)s AND address.user_id = %(user_id_1)s
2022-05-01 15:10:13,712 INFO sqlalchemy.engine.Engine [generated in 0.00021s] {'name_1': 'zhangsansan', 'user_id_1': 1}
运行结果：Address(id=1, email_address='sansan@sqlalchemy.com')

从查询结果可以清晰看到符合条件的数据显示出来了，体现的SQL语句中也可以清晰看到，使用的就是JOIN关联查询。

#

一、数据更新

from sqlalchemy.orm import Session
from sqlalchemy import select


session = Session(engine)

stmt = select(User).where(User.name == "zhangsansan")
patrick = session.scalars(stmt).one()

stmt = (
    select(Address).join(
        Address.user
    ).where(
        User.name == "lixiaoxiao"
    ).where(
        Address.user_id == 2
    )
)
sandy_address = session.scalars(stmt).one()


#   进行更改-给用户再增加一个关联的邮箱地址
patrick.addresses.append(
    Address(email_address="sansan@163.com")
)

#   进行更改-对用户邮箱进行修改
sandy_address.email_address = "xiaoxiao@qq.com"


session.commit()

信息注释：

• append就是追加的意思，查询到指定用户进行数据新增
• email_address=是赋值操作，是查询到指定用户，对原有数据进行修改，也就是替换一个新的

#

一、数据删除

from sqlalchemy.orm import Session
from sqlalchemy import select


session = Session(engine)

stmt = select(User).where(User.name == "zhangsansan")
patrick = session.scalars(stmt).one()

stmt = select(User).where(User.name == "zhangsansan")
patrick = session.scalars(stmt).one()

stmt = (
    select(Address).join(
        Address.user
    ).where(
        User.name == "lixiaoxiao"
    ).where(
        Address.user_id == 2
    )
)
sandy_address = session.scalars(stmt).one()

# 一些删除
sandy = session.get(User, 2)
sandy.addresses.remove(sandy_address)
session.flush()

session.delete(patrick)
session.commit()

信息注释：

• session.get(User, 2)是根据映射表获取数据
• sandy.addresses.remove(sandy_address)根据查询到的数据进行删除
• session.flush()是将所有对象更改刷新到数据库
• session.delete(patrick)是根据查询到的进行删除

注意点：

• remove一般来说，是操作对象还存在，只是从一个集合或容器中移除；而delete，则会执行销毁对象的操作，容器都不在了

今天先聊到这里吧，以上总结或许能帮助到你，或许帮助不到你，但还是希望能帮助到你，如有疑问、歧义，直接私信留言会及时修正发布；非常期待你的一键 3 连【 点赞、收藏、分享 】哟，谢谢！

未完成，待续……

一直在努力，希望你也是！

微信搜索公众号：就用python