sqlalchemy 一对多/多对一/多对多/一对一关系定义设置全方位操作方法

本是古典何须时尚 2022-05-12 07:54 569阅读 0赞

**sqlalchemy 作为一款ORM在操作数据库方面非常的方便,这里总结了一些对应关系的设置以及查询方法!**

**使用外键关联表：**  
表设计

from sqlalchemy import Column, ForeignKey
    from sqlalchemy.types import String, Integer, CHAR, BIGINT
    
    class Blog(BaseModel):
        __tablename__ = 'blog'
    
        id = Column(BIGINT, primary_key=True, autoincrement=True)
        title = Column(String(64), server_default='', nullable=False)
        text = Column(String, server_default='', nullable=False)
        user = Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
        create = Column(BIGINT, index=True, server_default='0', nullable=False)
    
    
    class User(BaseModel):
        __tablename__ = 'user'
    
        id = Column(BIGINT, primary_key=True, autoincrement=True)
        name = Column(String(32), server_default='', nullable=False)
        username = Column(String(32), index=True, server_default='', nullable=True)
        password = Column(String(64), server_default='', nullable=False)

提交数据

session = Session()
    user = User(name='first', username=u'新的')
    session.add(user)
    session.flush()#这个为了放回 id
    blog = Blog(title=u'第一个', user=user.id)
    session.add(blog)
    session.commit()

**一: 一对多关系**

*  表设计
    
        class Parent(Base):  # 一
            __tablename__ = 'parent'
            id = Column(Integer, primary_key=True)
            name = Column(String(64),nullable=False)
            full_name = Column(String(64))
            children = relationship("Child")
             # 默认返回的是列表，collection_class=set 加他返回的是集合， 
             #collection_class=attribute_mapped_collection('name') 返回的是name 字典 值从属性中取
             #collection_class=mapped_collection(lambda children: children.name.lower() 这个可以自定义值
           # 在父表类中通过 relationship() 方法来引用子表的类集合
        
            class Child(Base):  #多
                __tablename__ = 'child'
                id = Column(Integer, primary_key=True)
                name = Column(String(64),nullable=False)
                full_name = Column(String(64))
                parent_id = Column(Integer, ForeignKey('parent.id'))
                # 在子表类中通过 foreign key (外键)引用父表的参考字段
 *  添加数据
    
        parent = Parent(name='morgan', full_name='morganlions')
        parent.children = Child(name='child 1 name',full_name='child 1 full name')
        session.add(parent)
        session.commit()
        #如果parent 已经存在
        parent = session.query(Parent).filter(Parent.name=='Morgan')
        children = [
                    Child(name='child 2', full_name='child 2 full name', parent_obj = parent),
                    Child(name='child 3', full_name='child 3 full name', parent_obj = parent),
        ]
        session.add_all(children)
        session.commit()
 *  查询数据（单向关系）
    
        parent = session.query(Parent).get(1)
        print(parent.children)
        #一对多查询
        parent = session.query(Parent).filter(Parent.children.any(Child.name==u'child 1')).first()

**二:多对一关系**

这个相比上面的一对多而言是双向的关系.

*  表设计
    
        class Parent(Base):#一
            __tablename__ = 'parent'
            id = Column(Integer, primary_key=True)
            name = Column(String(64), nullable=False)
            full_name = Column(String(64))
            children = relationship("Child", back_populates="parent", lazy="dynamic")
        
        class Child(Base):#多
            __tablename__ = 'child'
            id = Column(Integer, primary_key=True)
            name = Column(String(64), nullable=False)
            full_name = Column(String(64))
            parent_id = Column(Integer, ForeignKey('parent.id'))
            parent = relationship("Parent", order_by = 'Parent.id' back_populates="children")
            # 子表类中附加一个 relationship() 方法
         # 并且在(父)子表类的 relationship() 方法中使用 relationship.back_populates 参数
 *  添加数据
    
        这个和单向的一样，不过可以反过来操作
 *  查询数据(双向关系）
    
        parent = session.query(Parent).get(1)
        print(parent.children) 
        children = session.query(Child).get(1)
        print(children.parent)
        # lazy="dynamic" 我们添加 这个后可以子查询关联表的时候自由的选择
        print(parent.children.all())
        print(patent.children.filter(Child.name='child1').first()
        #多对一查询父亲
        children = session.query(Child).filter(Child.parent_obj.has(Parent.name=='morgan')).first()

**三:多对多关系**

1.  单向多对多  
    **通过中间表 association 关联表 链接 left 和 right 两张表 left -> right ,建立的是单向关系**
    
        # 多对多关系中的两个表之间的一个关联表
        association_table = Table('association', Base.metadata,
            Column('left_id', Integer, ForeignKey('left.id')),
            Column('right_id', Integer, ForeignKey('right.id'))
        )
        
        class Parent(Base):
            __tablename__ = 'left'
            id = Column(Integer, primary_key=True)
            children = relationship("Child",secondary=association_table)
            # 在父表中的 relationship() 方法传入 secondary 参数，其值为关联表的表名
        
        class Child(Base):
            __tablename__ = 'right'
            id = Column(Integer, primary_key=True)
2.  双向多对多  
    **通过中间表 association 关联表 链接 left 和 right 两张表 left <-> right ,建立的是双向关系**
    
        association_table = Table('association', Base.metadata,
            Column('left_id', Integer, ForeignKey('left.id')),
            Column('right_id', Integer, ForeignKey('right.id'))
        )
        
        class Parent(Base):
            __tablename__ = 'left'
            id = Column(Integer, primary_key=True)
            children = relationship("Child", secondary=association_table,back_populates="parents")
        
        class Child(Base):
            __tablename__ = 'right'
            id = Column(Integer, primary_key=True)
            parents = relationship( "Parent",secondary=association_table,back_populates="children")

下面是一种比较简洁的建立双向多对多关系的方法,上面的方法中我们用的是 back\_populates ,下面我们将其换成,backref,这样我们只需要在第一个双向关系建立的时候,建立反向双向关系。

association_table = Table('association', Base.metadata,
        Column('left_id', Integer, ForeignKey('left.id')),
        Column('right_id', Integer, ForeignKey('right.id'))
    )
    
    class Parent(Base):
        __tablename__ = 'left'
        id = Column(Integer, primary_key=True)
        name = Column(String(64),nullable=False, index=True)
        full_name = Column(String(64))
        children = relationship("Child",secondary=association_table, backref="parents")
    
    class Child(Base):
        __tablename__ = 'right'
        name = Column(String(64),nullable=False, index=True)
        full_name = Column(String(64))
        id = Column(Integer, primary_key=True)

secondary 可以作为回调函数返回一个值

class Parent(Base):
        __tablename__ = 'left'
        id = Column(Integer, primary_key=True)
        children = relationship("Child",secondary=lambda: association_table,backref="parents")

**多对多关系添加数据：**

parent = Parent(name='Morgan',full_name='MorganLions')
    parent.children = [Child(name='child name'，full_name='child full name')]
    #提交数据
    session.add(parent)
    session.commit()
    #如果是双向关系也可以这样添加
    child = Child(name='child name',full_name='child full name'}
    child.parents = [Parent(name='Morgan', full_name ='Morgan Lions')]
    session.add(child)
    session.commit()
    session = Session()
    #找出数据在添加
    parent = session.query(Parent).filter(Parent.name=='Morgan').one()
    parent.children =  [Child(name='child name'，full_name='child full name')]
    session.commit()
    #删除多对多中间关系，不删除实体，这里我们对应的是children
        #1清除parent 中的 children 数据
    parent = session.query(Parent).filter(Parent.name='Morgan').one()
    parent.children=[]
    session.commit()
        #2删除两个表对应的关系
    children = session.query(Child).filter(Child.name='child 1').one()
    parent.children.remove(children)
    session.commit()
    #删除实体，不删除关系
    #  1:创建要实验的数据
    parent = session.query(Parent).filter(Parent.name=='Morgan').one()
    children = Child(name='child 5)
    parent.children = [children]
    session.commit()
    #2: 删除实体
    children  = session.query(Child).filter(Child.name=='child 5').one()
    session.delete(children)
    session.commit()

**多对多关系查询数据**  
1：这是一种方法通过对象的方法去转换，当然你也可以直接返回去用，我需要列表的形式，还有另外的一种方法，自己写个方法在模型文件中，就是你建立数据库文件的表中添加相应的方法，这里我贴出自己的方法，欢迎指正，谢谢。

#单条数据查询
        parent = session.query(Parent).get(1)
        #找到儿子
        print(parent.children)
        children = session.query(Child).get(1)
        #找到父亲
        print(children.parent)
        #多条数据查询
        parent = session.query(Parent).order_by(Parent.id)
        count = parent.count() #计数
        page = ... 
        limit = ...
        parent = parent.offset(page).limit(limit)
        result = []
        for p in parent.all():
            arg ={}
            arg['id'] = p.id
            arg['name'] = p.name
            arg['full_name'] = p.full_name
            arg['children'] = {c.name:c.full_name for c in p.children}
            result.append(arg)
        print(result)

2：在模型中添加如下的方法，这个方法，如果去关联多对多查询，比较慢，因为本质上而言，还是和数据库的多表查询没有什么区别，还是要转换成对应的sql语句去执行，速度自然就慢了，建议在列表中尽量不要用这种多表联合查询。性能很不好。

def column_dict(self, params=None, operation=None, relation=None):
        '''
        sql查询返回字典
        :param params: 需要显示的字段
        :param operation:操作方法
                        {'one':func1 'two': func2 , 'param':'提示术语'}
                        {'one':func1 'two': 'param':'提示术语'}
                        {'one':func1 'two': func2 }
                        param 作为第二个函数的参数,也可以作为单独的提示
        :param relation
        :return:
        '''
        model_dict = dict(self.__dict__)
        del model_dict['_sa_instance_state']
        if params:
            keys = [k for k in model_dict.keys()]
            for key in keys:
                if key not in params:
                    del model_dict[key]
        if relation:
            for r in relation:
                rel = eval('self.{0}'.format(r))
                result = [self.change_str(operation, x.column_dict()) for x in rel]
                model_dict['{0}'.format(r)] = result
        if operation:
            model_dict = self.change_str(operation, model_dict)
            if isinstance(model_dict, str):
                return False, 'model_dict', 0
        return model_dict
    
    Base.column_dict = column_dict
    
    @staticmethod
    def change_str(operation, model_dict):
        '''
        改变输出类型
        :param operation:
        :param model_dict:
        :return:
        '''
    
        for key, funcs in operation.items():
            method = model_dict[key]
            func = funcs.get('one')
            second_func = funcs.get('two')
            param = funcs.get('param')
            if param and func and second_func:
                model_dict[key] = func(method) if method else second_func(param)
            elif second_func is None and func and param:
                model_dict[key] = func(method) if method else param
            elif param is None and func and second_func:
                model_dict[key] = func(method) if method else second_func()
            else:
                return '操作函数设置错误'
        return model_dict
    #调用
    parent = session.query(Parent)
    
       operation = {
             'name: {'one': func, 'param': '参数'},
         }
         relation = ['children']
         result = [x.column_dict(operation=operation, relation=relation) for x in parent]

这个方法也可以不传参数，直接调用，可以直接返回字典，返回的是所有字段，如果要自己选定字段需要传参数params=‘字段1,字段二’，这个方法可以关联很多表，但速度可想而知。在不关联的情况下查询速度180ms 左右，关联一个表600ms 左右，关联 二个 1000ms左右，以此类推。还是那句话，不建议列表中关联查询，更不要用这样的多对多的关系。

**sqlalchemy自动关系**  
看到这里你可能会想，太麻烦了，有没有能自动关联的关系呢，答案是有的，在relationship 中有个参数cascade  
**cascade 所有的可选字符串项是:**

all , 所有操作都会自动处理到关联对象上.
    save-update , 关联对象自动添加到会话.
    delete , 关联对象自动从会话中删除.
    delete-orphan , 属性中去掉关联对象, 则会话中会自动删除关联对象.**只适用于一对多的关系**
    merge , session.merge() 时会处理关联对象.
    refresh-expire , session.expire() 时会处理关联对象.
    expunge , session.expunge() 时会处理关联对象.

**delete**

class Blog(BaseModel):
        __tablename__ = 'blog'
    
        id = Column(BIGINT, primary_key=True, autoincrement=True)
        title = Column(String(64), server_default='', nullable=False)
        user = Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
    
    
    class User(BaseModel):
        __tablename__ = 'user'
    
        id = Column(BIGINT, primary_key=True, autoincrement=True)
        name = Column(String(32), server_default='', nullable=False)
    
        blog_list = relationship('Blog', cascade='save-update, delete')
    
    
    if __name__ == '__main__':
        session = Session()
    
        #user = User(name=u'用户')
        #user.blog_list = [Blog(title=u'哈哈')]
        #session.add(user)
        user = session.query(User).first()
        session.delete(user)
        session.commit()

**merge 有则修改，无则创建**

class Blog(BaseModel):
        __tablename__ = 'blog'
        id = Column(BIGINT, primary_key=True, autoincrement=True)
        title = Column(String(64), server_default='', nullable=False)
        user = Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
    class User(BaseModel):
        __tablename__ = 'user'
        id = Column(BIGINT, primary_key=True, autoincrement=True)
        name = Column(String(32), server_default='', nullable=False)
        blog_list = relationship('Blog', cascade='save-update, delete, delete-orphan, merge')
    if __name__ == '__main__':
        session = Session()
        user = User(id=1, name='1')
        session.add(user)
        session.commit(user)
        user = User(id=1, blog_list=[Blog(title='哈哈')])
        session.merge(user)
        session.commit()

**属性代理**

from sqlalchemy.ext.associationproxy import association_proxy
    
    class Blog(BaseModel):
        __tablename__ = 'blog'
    
        id = Column(Integer, autoincrement=True, primary_key=True)
        title = Column(Unicode(32), nullable=False, server_default='')
        user = Column(Integer, ForeignKey('user.id'), index=True)
        #对应的属性代理设置
        def __init__(self, title):
            self.title = title
    
    
    class User(BaseModel):
        __tablename__ = 'user'
    
        id = Column(Integer, autoincrement=True, primary_key=True)
        name = Column(Unicode(32), nullable=False, server_default='')
    
        blog_list = relationship('Blog')
        blog_title_list = association_proxy('blog_list', 'title')

有的时候我们只需要处理对应的一个，比如上面这个，我们指出了blog\_list 中的title

#设置与调用
    session = Session()
    
    user = User(name='xxx')
    user.blog_list = [Blog(title='ABC')]
    session.add(user)
    session.commit()
    
    user = session.query(User).first()
    print user.blog_title_list
    #如果我们希望像下面这样更改数据
    user = session.query(User).first()
    user.blog_title_list = ['NEW']
    session.add(user)
    session.commit()
    #需要在代理设置那个类中添加对应的 __init__ 方法
    def __init__(self, title):
        self.title = title
    #也可以这样 
     blog_title_list = association_proxy('blog_list', 'title',creator=lambda t: User(title=t))

查询方法

class Blog(BaseModel):
        __tablename__ = 'blog'
    
        id = Column(Integer, autoincrement=True, primary_key=True)
        title = Column(Unicode(32), server_default='')
        user = Column(Integer, ForeignKey('user.id'), index=True)
    
        user_obj = relationship('User')
        user_name = association_proxy('user_obj', 'name')

查询：

blog = session.query(Blog).filter(Blog.user_name == u'XX').first()

反过来查询

user = session.query(User).filter(User.blogs_title.contains('A')).first()