sqlalchemy_No1_简介及快速入门

简介

SQLAlchemy包是满足ORM架构的包，用来实现Python和数据之间的交互。

组成部分

SQLAlchemy的结构图如下：

SQLAlchemy主要由SQLAlchemy ORM和SQLAlchemy Core两个部分组成。

• SQLAlchemy Core
  基层结构
  • 连接数据库
  • 数据库进行交互(进行查询和获取结果)
  • 以编程方式构造SQL语句。
• SQLAlchemy ORM
  基于SQLAlchemy Core， 对象映射
  • Python类与数据库的映射
  • 会话对象的持久机制
  • 扩展了SQL Expression Language，允许根据对象编写和调用SQL

快速入门

安装

安装SQLAlchemy

通过pip安装SQLAlchemy。

pip install SQLAlchemy
1

安装DBAPI

DBAPI表示数据库驱动，通常是每种数据库类型专用的包。例如MySQL的有pymysql，Oracle的有cx_Oracle等。此处以pymysql作为DBAPI进行安装。

pip install pymysql
1

检测

根据上面的步骤安装后，需要检测一下安装是否存在问题。
在cmd中输入python，进入python环境。然后输入如下指令查看SQLAlchemy的版本，如果正常显示，则说明已经完成安装，否则需重新安装。

import sqlalchemy
sqlalchemy.__verion__
1
2

代码实现步骤

第一步：声明数据模型类

# 步骤1：声明模型
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_account"
    # 表的列名及相关设定
    id = Column(Integer, primary_key=True)
    name = Column(String(30))
    fullname = Column(String(60))
    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(255), nullable=False)
    user_id = Column(Integer, ForeignKey("user_account.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})"
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

第二步：创建数据库连接Engine

from sqlalchemy import create_engine
# dialect+driver://username:password@host:port/database
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/testdb",encoding="utf8")
1
2
3

第三步：创建数据库表

# 步骤3：创建数据库表
Base.metadata.create_all(engine)
1
2

第四步：增加记录

# 步骤4：利用持久化会话创建记录
from sqlalchemy.orm import Session
with Session(engine) as session:
# 创建要添加的实例对象
    spongebob = User(
        name="spongebob",
        fullname="Spongebob Squarepants",
        addresses=[Address(email_address="spongebob@sqlalchemy.org")],
    )
    sandy = User(
        name="sandy",
        fullname="Sandy Cheeks",
        addresses=[
            Address(email_address="sandy@sqlalchemy.org"),
           Address(email_address="sandy@squirrelpower.org"),
        ],
    )
    patrick = User(name="patrick", fullname="Patrick Star")
    # 添加所有对象至会话
    session.add_all([spongebob, sandy, patrick])
    # 执行添加命令
    session.commit()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

第五步：单表查询

from sqlalchemy import select
session = Session(engine)
stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))
for user in session.scalars(stmt):
    print(user)
1
2
3
4
5

第六步：多表查询

stmt = (
    select(Address)
    .join(Address.user)
    .where(User.name == "sandy")
    .where(Address.email_address == "sandy@sqlalchemy.org")
)
sandy_address = session.scalars(stmt).one()
1
2
3
4
5
6
7

第七步：更新记录

stmt = select(User).where(User.name == "patrick")
patrick = session.scalars(stmt).one()
patrick.addresses.append(Address(email_address="patrickstar@sqlalchemy.org"))
sandy_address.email_address = "sandy_cheeks@sqlalchemy.org"
session.commit()
1
2
3
4
5

第八步：删除记录

sandy = session.get(User, 2)
sandy.addresses.remove(sandy_address)
session.flush()
session.delete(patrick)
session.commit()
1
2
3
4
5