一对多和多对多
数据库表结构设计是程序项目开发前的重要环节，后期数据库操作都是围绕着这个已经设计好的表结构进行，如果表结构设计有问题，整个程序项目就有存在需要整个推翻重构的风险...
数据库表结构除了简单的单表操作以外，还有一对多、多对多等。
一对多
基于SQLAlchemy我们可以先创建如下结构的2张表，然后来看看具体怎样通过外键ForeignKey或者relationship联表操作
创建表
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:[email protected]:3306/s13?charset=utf8",
max_overflow=5)
Base = declarative_base()
# 创建用户组表
class Group(Base):
__tablename__ = ‘group‘
nid = Column(Integer, primary_key=True, autoincrement=True)
caption = Column(String(32))
# 创建用户表
class User(Base):
__tablename__ = ‘user‘
nid = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(32))
# 添加了外键 表示再增加用户时 group_id 只能是表group中存在的id 否则报错
group_id = Column(Integer, ForeignKey(‘group.nid‘))
def init_db():
Base.metadata.create_all(engine)
def drop_db():
Base.metadata.drop_all(engine)
# init_db执行创建表操作
# init_db()
# 实例化类 开始后续的查询操作
Session = sessionmaker(bind=engine)
session = Session()
-------------- 插入数据 ---------------------
# 插入组名
# session.add(Group(caption=‘运维‘))
# session.add(Group(caption=‘开发‘))
# session.commit()
# 插入用户
# session.add_all([
# User(name=‘user_01‘, group_id=1),
# User(name=‘user_02‘, group_id=1),
# User(name=‘user_03‘, group_id=2),
# ])
# session.commit()
__repr__
单表查询结果获取的是一个内存对象，我们可以在表中增加一个特殊对象方法__repr__，自定义查询显示结果
# 获得的只是个对象 ret = session.query(User).filter(User.name == ‘user_01‘).all() print(ret) # 输出 [<day13.s1.User object at 0x00000288737C1898>] # 在User表中增加__repr__方法 查询后会自动执行 class User(Base): __tablename__ = ‘user‘ nid = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(32)) group_id = Column(Integer, ForeignKey(‘group.nid‘)) # __repr__方法会自动执行 def __repr__(self): temp = ‘%s %s %s‘ %(self.nid, self.name, self.group_id) return temp # 再次执行后结果 [1 User_01 1] obj = ret[0] print(obj.nid, obj.name, obj.group_id) # 输出 1 User_01 1 # 仅查用户名 ret = session.query(User.name).all() print(ret) # 输出 [(‘User_01‘,), (‘User_02‘,), (‘User_03‘,)]
联表查询
# 联表查询原理是自动给你生成sql语句然后执行 sql = session.query(User).join(Group) print(sql) ret = session.query(User).join(Group).all() print(ret) # left join # ret = session.query(User).join(Group, isouter=True).all() # print(ret) # 输出 SELECT user.nid AS user_nid, user.name AS user_name, user.group_id AS user_group_id FROM user INNER JOIN `group` ON `group`.nid = user.group_id [1 user_01 1, 2 user_02 1, 3 user_03 2]
指定映射关系
# 指定映射关系联表查询 ret = session.query(User.name, Group.caption).join(Group).all() print(ret) # 输出 [(‘user_01‘, ‘运维‘), (‘user_02‘, ‘运维‘), (‘user_03‘, ‘开发‘)]
relationship 正向查找
为了简化联合查询，我们还可以创建一个2个表之间的虚拟关系relationship，该关系与表结构无关，仅方便我们后续查询
class User(Base):
__tablename__ = ‘user‘
nid = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(32))
# 外键
group_id = Column(Integer, ForeignKey(‘group.nid‘))
# 创建虚拟关系 relationship 一般与外键配合使用
group = relationship("Group", backref=‘uuu‘)
def __repr__(self):
temp = ‘%s %s %s‘ %(self.nid, self.name, self.group_id)
return temp
ret = session.query(User).all()
for obj in ret:
# obj 代表 User
# group 代表新 Group
print(obj.nid, obj.name, obj.group_id, obj.group.nid, obj.group.caption)
# 输出
1 user_01 1 1 运维
2 user_02 1 1 运维
3 user_03 2 2 开发
relationship 反向查找
需求：查找表2中的所有事运维职位的人
# 查所有是 运维 的人 ret = session.query(User.name, Group.caption).join(Group, isouter=True). filter(Group.caption == ‘运维‘)).all() print(ret) # 利用relationship 新方式反向查询 obj = session.query(Group).filter(Group.caption == ‘运维‘).first() print(obj.nid, obj.caption) # uuu 代表在这个组下面的所有人 是一个列表 print(obj.uuu) # 输出 [(‘user_01‘, ‘运维‘), (‘user_02‘, ‘运维‘)] 1 运维 [1 user_01 1, 2 user_02 1]
多对多
例如公司里的很多服务器同时有包括root用户在内的很多账户可以登陆，就是一个简单的多对多结构
老方法查找 C2 主机有哪些账户可以登陆
# 1、找到 hostname 为c1的nid host_obj = session.query(Host).filter(Host.hostname == ‘c2‘).first() print(host_obj.nid) # 2、指定映射关系查找 对应主机用户ID host_to_host_user = session.query(HostToHostUser.host_user_id).filter # (HostToHostUser.host_id == host_obj.nid).all() print(host_to_host_user) # [(1,), (2,), (3,)] # [1, 2, 3] r = zip(*host_to_host_user) # 3、查找到用户 users = session.query(HostUser.username).filter(HostUser.nid.in_(list(r)[0])).all() print(users) # 输出 1 [(1,), (3,)] [(‘root‘,), (‘db‘,)]
利用 relationship 关系简化多对多查询
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:[email protected]:3306/s13",
max_overflow=5)
Base = declarative_base()
class Host(Base):
__tablename__ = ‘host‘
nid = Column(Integer, primary_key=True, autoincrement=True)
hostname = Column(String(32))
port = Column(String(32))
ip = Column(String(32))
class HostUser(Base):
__tablename__ = ‘host_user‘
nid = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(32))
class HostToHostUser(Base):
__tablename__ = ‘host_to_host_user‘
nid = Column(Integer, primary_key=True, autoincrement=True)
host_id = Column(Integer, ForeignKey(‘host.nid‘))
host_user_id = Column(Integer, ForeignKey(‘host_user.nid‘))
# 建立关系
host = relationship(‘Host‘, backref=‘h‘)
host_user = relationship(‘HostUser‘, backref=‘u‘)
def init_db():
Base.metadata.create_all(engine)
def drop_db():
Base.metadata.drop_all(engine)
# 创建表
# init_db()
# 插入数据
Session = sessionmaker(bind=engine)
session = Session()
# session.add_all([
# Host(hostname=‘c1‘, port=‘22‘, ip=‘1.1.1.1‘),
# Host(hostname=‘c2‘, port=‘22‘, ip=‘1.1.1.2‘),
# ])
# session.commit()
# session.add_all([
# HostUser(username=‘root‘),
# HostUser(username=‘sa‘),
# HostUser(username=‘db‘),
# ])
# session.commit()
# session.add_all([
# HostToHostUser(host_id=1, host_user_id=1),
# HostToHostUser(host_id=1, host_user_id=2),
# HostToHostUser(host_id=2, host_user_id=1),
# HostToHostUser(host_id=2, host_user_id=3),
# ])
# session.commit()
# relationship 多对多查询
host_obj = session.query(Host).filter(Host.hostname == ‘c2‘).first()
for item in host_obj.h:
print(item.host_user.username)
# 输出
root
db
另一种简单方式
from sqlalchemy import create_engine,and_,or_,func,Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,ForeignKey
from sqlalchemy.orm import sessionmaker,relationship
engine = create_engine("mysql+pymysql://root:[email protected]:3306/s13",
max_overflow=5)
Base = declarative_base()
class HostToHostUser(Base):
__tablename__ = ‘host_to_host_user‘
nid = Column(Integer, primary_key=True, autoincrement=True)
host_id = Column(Integer, ForeignKey(‘host.nid‘))
host_user_id = Column(Integer, ForeignKey(‘host_user.nid‘))
class Host(Base):
__tablename__ = ‘host‘
nid = Column(Integer, primary_key=True, autoincrement=True)
hostname = Column(String(32))
port = Column(String(32))
ip = Column(String(32))
# 在其中一张表中加如下secondary
host_user = relationship(‘HostUser‘, secondary=HostToHostUser.__table__,
backref=‘h‘)
class HostUser(Base):
__tablename__ = ‘host_user‘
nid = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(32))
Session = sessionmaker(bind=engine)
session = Session()
host_obj = session.query(Host).filter(Host.hostname == ‘c2‘).first()
print(host_obj.host_user)
for item in host_obj.host_user:
print(item.username)
# 输出
[<__main__.HostUser object at 0x000001E44AF49390>, <__main__.HostUser object at
0x000001E44AF493C8>]
root
db
时间： 2024-10-02 06:02:42