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)