SQLAlchemy

https://docs.sqlalchemy.org.cn/en/20/orm/quickstart.htmlhttps://docs.sqlalchemy.org.cn/en/20/orm/quickstart.html

声明模型¶

在这里，我们定义模块级构造，这些构造将构成我们从数据库中查询的结构。这种结构被称为声明式映射，它同时定义了 Python 对象模型以及数据库元数据，该元数据描述了在特定数据库中存在或将存在的实际 SQL 表。

Base = declarative_base() 是 SQLAlchemy 中的一个关键语句，它用于定义一个基础类（Base），所有 ORM 模型类都需要继承这个基础类。通过这种方式，SQLAlchemy 能够将 Python 类与数据库表结构建立映射关系。

from sqlalchemy.ext.declarative import declarative_base# 创建基础类
Base = declarative_base()

实例代码

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker# 创建基础类
Base = declarative_base()# 定义 ORM 模型类
class User(Base):__tablename__ = 'users'  # 映射到数据库中的表名id = Column(Integer, primary_key=True)  # 定义主键name = Column(String(50))  # 定义列age = Column(Integer)# 创建数据库引擎
engine = create_engine('sqlite:///:memory:')# 创建所有表
Base.metadata.create_all(engine)# 创建会话
Session = sessionmaker(bind=engine)
session = Session()# 插入数据
new_user = User(name='Alice', age=25)
session.add(new_user)
session.commit()# 查询数据
user = session.query(User).filter_by(name='Alice').first()
print(user.name, user.age)  # 输出: Alice 25

SQLAlchemy 2.0

SQLAlchemy 2.0 或更高版本，建议采用这种新形式来定义 ORM 模型类。

这是新版本的

class User(Base):__tablename__ = "user_account"id: Mapped[int] = mapped_column(primary_key=True)name: Mapped[str] = mapped_column(String(30))fullname: Mapped[Optional[str]]addresses: Mapped[List["Address"]] = relationship(back_populates="user", cascade="all, delete-orphan")

这个是旧版本的

Base = declarative_base()# Step 3. 定义可以使用Base类管理的模型类
class Session(Base):"""Session 类表示聊天会话"""__tablename__ = "sessions"id = Column(Integer, primary_key=True)## unique=True表示 session_id 字段的值必须是唯一的，不能有重复值。 不能为空session_id = Column(String, unique=True, nullable=False)# 这不是数据库中的一个具体字段，而是 ORM 层面的逻辑关系，用于简化操作和查询。messages = relationship("Message", back_populates="session")class Message(Base):"""Message 类表示会话中的各个消息"""__tablename__ = "messages"id = Column(Integer, primary_key=True)session_id = Column(Integer, ForeignKey("sessions.id"), nullable=False)role = Column(String, nullable=False)content = Column(Text, nullable=False)session = relationship("Session", back_populates="messages")

安装

连接数据库

映射类

添加记录

然后session.commit()

查询

session = Session()# result = session.query(Person).all()
result = session.query(Person).filter(Person.address == 'aaa')for person in result:print(f'name: {person.name}, birthday: {person.birthday}')

# person = session.query(Person).filter(Person.address == 'aaa').first()
# person = session.query(Person).filter(Person.id == 100).first()
# person = session.query(Person).filter(Person.id == 1).one()
# 返回查询结果的第一条记录的第一列值。
person = session.query(Person).filter(Person.id == 1).scalar()

修改记录

记得提交事务

方式2修改时要传json的表达式

新的映射

class User(Base):__tablename__ = "user_account"id: Mapped[int] = mapped_column(primary_key=True)name: Mapped[str] = mapped_column(String(30))fullname: Mapped[Optional[str]]addresses: Mapped[List["Address"]] = relationship(back_populates="user", cascade="all, delete-orphan")

当一些字段重复是可以单独定义出来

import datetimefrom sqlalchemy import create_engine, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Mapped, mapped_column
from sqlalchemy.sql import func
from typing_extensions import Annotatedengine = create_engine('mysql://root:test@localhost/testdb', echo=True)
Base = declarative_base()# 当很多字段都有统一的要求，就把他给单独定义出来
int_pk = Annotated[int, mapped_column(primary_key=True)]
required_unique_name = Annotated[str, mapped_column(String(128), unique=True, nullable=False)]
timestamp_default_now = Annotated[datetime.datetime, mapped_column(nullable=False, server_default=func.now())]class Customer(Base):__tablename__ = "customers"id: Mapped[int_pk]name: Mapped[required_unique_name]birthday: Mapped[datetime.datetime]city: Mapped[str] = mapped_column(String(128), nullable=True)create_time: Mapped[timestamp_default_now]Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

一对多的映射

一开始没relationship

class Department(Base):__tablename__ = "department"id: Mapped[int_pk]name: Mapped[required_unique_name]def __repr__(self):return f'id: {self.id}, name: {self.name}'class Employee(Base):__tablename__ = "employee"id: Mapped[int_pk]dep_id: Mapped[int] = mapped_column(ForeignKey("department.id"))name: Mapped[required_unique_name]birthday: Mapped[timestamp_not_null]

这样会出错，因为对象还没真正存入数据库，所以主键id就没生成，可以采用 flush() 刷新进去，但是多了之后不知道哪里需要flush(),所以有了relationship

单纯加上relationship这个就好，并且可以直接查看关联的部门信息

class Department(Base):__tablename__ = "department"id: Mapped[int_pk]name: Mapped[required_unique_name]def __repr__(self):return f'id: {self.id}, name: {self.name}'class Employee(Base):__tablename__ = "employee"id: Mapped[int_pk]dep_id: Mapped[int] = mapped_column(ForeignKey("department.id"))name: Mapped[required_unique_name]birthday: Mapped[timestamp_not_null]department: Mapped[Department] = relationship()

只加员工对象就可以，他会直接把部门对象也创建出来，依据有没有id判断是存在还是不存在，

不存在就会创建

并且可以直接查看部门信息

import datetimefrom sqlalchemy import create_engine, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Mapped, mapped_column, relationship
from typing_extensions import Annotated
from typing import Listengine = create_engine('mysql://root:test@localhost/testdb', echo=True)
Base = declarative_base()int_pk = Annotated[int, mapped_column(primary_key=True)]
required_unique_name = Annotated[str, mapped_column(String(128), unique=True, nullable=False)]
timestamp_not_null = Annotated[datetime.datetime, mapped_column(nullable=False)]class Department(Base):__tablename__ = "department"id: Mapped[int_pk]name: Mapped[required_unique_name]# 显式的表示出来employees: Mapped[List["Employee"]] = relationship(back_populates="department")def __repr__(self):return f'id: {self.id}, name: {self.name}'class Employee(Base):__tablename__ = "employee"id: Mapped[int_pk]dep_id: Mapped[int] = mapped_column(ForeignKey("department.id"))name: Mapped[required_unique_name]birthday: Mapped[timestamp_not_null]# relationship 实际不存在表中，他只存在内存中，lazy=False表示取消懒加载 ，查询的时候直接把另一个给查出来# back_populates="employees"  自动给department表增加一个映射employees，而且他是一个集合类型的 这个employees当前对应的就是这个表# 这个映射在department你看不见，但是为了可读性直观一些，我们会显式的表示出来# 这样定义完之后，我们就可以直接查看部分的信息了department: Mapped[Department] = relationship(lazy=False, back_populates="employees")def __repr__(self):return f'id: {self.id}, dep_id: {self.dep_id}, name: {self.name}, birthday: {self.birthday}'Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

多对多的映射

并不需要专门定义一个类，如果专门定义一个类，那么就成为了多个一对多的映射

这是联合主键

import datetimefrom sqlalchemy import create_engine, String, ForeignKey, Table, Column
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Mapped, mapped_column, relationship
from typing_extensions import Annotated
from typing import List, Setengine = create_engine('mysql://root:test@localhost/testdb', echo=True)
Base = declarative_base()int_pk = Annotated[int, mapped_column(primary_key=True)]
required_unique_name = Annotated[str, mapped_column(String(128), unique=True, nullable=False)]
required_string = Annotated[str, mapped_column(String(128), nullable=False)]# 多对多的中间表，不用声明一个类
# 这是联合主键
association_table = Table("user_role",Base.metadata,Column("user_id", ForeignKey("users.id"), primary_key=True),Column("role_id", ForeignKey("roles.id"), primary_key=True)
)class User(Base):__tablename__ = "users"id: Mapped[int_pk]name: Mapped[required_unique_name]password: Mapped[required_string]# secondary=association_table 指明中间表roles: Mapped[List["Role"]] = relationship(secondary=association_table, lazy=False, back_populates="users")def __repr__(self):return f'id: {self.id}, name: {self.name}'class Role(Base):__tablename__ = "roles"id: Mapped[int_pk]name: Mapped[required_unique_name]users: Mapped[List["User"]] = relationship(secondary=association_table, lazy=True, back_populates="roles")def __repr__(self):return f'id: {self.id}, name: {self.name}'Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

from db_init import Session, User, Roledef insert_records(s):role1 = Role(name="Admin")role2 = Role(name="Operator")user1 = User(name="Jack", password="111")user2 = User(name="Tom", password="222")user3 = User(name="Mary", password="333")# 用户中添加角色user1.roles.append(role1)user1.roles.append(role2)user2.roles.append(role1)user3.roles.append(role2)s.add_all([user1, user2, user3])s.commit()def select_user(s):u = s.query(User).filter(User.id == 1).one()print(u)print(u.roles)def select_role(s):r = s.query(Role).filter(Role.id == 2).one()print(r)print(r.users)session = Session()
# insert_records(session)
# select_user(session)
select_role(session)

一对一的映射

import datetimefrom sqlalchemy import create_engine, String, ForeignKey, Table, Column
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Mapped, mapped_column, relationship
from typing_extensions import Annotated
from typing import List, Setengine = create_engine('mysql://root:test@localhost/testdb', echo=True)
Base = declarative_base()int_pk = Annotated[int, mapped_column(primary_key=True)]
required_unique_string = Annotated[str, mapped_column(String(128), unique=True, nullable=False)]
required_string = Annotated[str, mapped_column(String(128), nullable=False)]class Employee(Base):__tablename__ = "employee"id: Mapped[int_pk]name: Mapped[required_unique_string]computer_id: Mapped[int] = mapped_column(ForeignKey("computer.id"), nullable=True)computer = relationship("Computer", lazy=False, back_populates="employee")def __repr__(self):return f'id: {self.id}, name: {self.name}'class Computer(Base):__tablename__ = "computer"id: Mapped[int_pk]model: Mapped[required_string]number: Mapped[required_unique_string]employee = relationship("Employee", lazy=True, back_populates="computer")def __repr__(self):return f'id: {self.id}, model: {self.model}, number: {self.number}'Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

from db_init import Session, Employee, Computerdef insert(s):c1 = Computer(model="Dell", number="1111")c2 = Computer(model="Surface", number="2222")c3 = Computer(model="MacBook Pro", number="3333")e1 = Employee(name="Jack", computer=c1)e2 = Employee(name="Mary", computer=c2)e3 = Employee(name="Tome", computer=c3)s.add_all([e1, e2, e3])s.commit()def select(s):e = s.query(Employee).filter(Employee.id == 1).scalar()if e:print(e)print(e.computer)c = s.query(Computer).filter(Computer.id == 2).scalar()if c:print(c)print(c.employee)def update_1(s):s.query(Employee).filter(Employee.id == 3).update({Employee.computer_id: None})s.commit()def update_2(s):c = s.query(Computer).filter(Computer.id == 3).scalar()e = s.query(Employee).filter(Employee.id == 3).scalar()if c and e:e.computer = cs.commit()session = Session()
# insert(session)
# select(session)
# update_1(session)
update_2(session)