SQLAlchemy
目录
https://docs.sqlalchemy.org.cn/en/20/orm/quickstart.html
https://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")
安装
连接数据库
映射类
Base = declarative_base() 是 SQLAlchemy 中的一个关键语句，它用于定义一个基础类（Base），所有 ORM 模型类都需要继承这个基础类。通过这种方式，SQLAlchemy 能够将 Python 类与数据库表结构建立映射关系。
写好模型类后只需要 Base.metadata.create_all 就可以创建所以继承base类的表
通过session进行管理，绑定数据库引擎
添加记录
然后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 datetime
from 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 Annotated
engine = 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 datetime
from 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 List
engine = 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 datetime
from 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, Set
engine = 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, Role
def 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 datetime
from 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, Set
engine = 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, Computer
def 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 = c
s.commit()
session = Session()
# insert(session)
# select(session)
# update_1(session)
update_2(session)
原文地址：https://blog.csdn.net/2201_75600005/article/details/145095005
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！