自动化运维Python系列之ForeignKey、relationship联表查询

一对多和多对多数据库表结构设计是程序项目开发前的重要环节，后期数据库操作都是围绕着这个已经设计好的表结构进行，如果表结构设计有问题，整个程序项目就有存在需要整个推翻重构的风险...数据库表结构除了简单的单表操作以外，还有一对多、多对多等。 一对多基于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:123@10.0.0.111: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:123@10.0.0.111: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:123@10.0.0.111: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