29 апреля 2022 Python

Задача реализовать прозрачное связывание "союзами" пользователей сайта. Такое необходимо когда пользователь А запросил союз у пользователя Б и тот одобрил этот союз, тогда у обоих пользователей должен обновиться список союзников и табличная связь будет выглядеть следующим образом

Связав таблицы где 1 колонка это номер строки хозяин, а вторая колонка это номер строки союзника получется, что у осла в союзниках козёл и мишка, а у пятачка союзник только винипух винипух

С точки зрения базы данных это связь многие ко многим и на Sqlalchemy решается через создание ассоциативной таблиц, а учитывая что помимо союзной связи необходимо сохранить состояние связи и даты создания и обновления, то ассоциативная таблица превращается в полноценную модель

class Union(db.Model):
    __tablename__ = "user_unions"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)

    user1_id = db.Column(db.Integer, db.ForeignKey('users.id'), primary_key=True)
    user2_id = db.Column(db.Integer, db.ForeignKey('users.id'), primary_key=True)

    created = db.Column(db.DateTime(timezone=True), server_default=func.now())
    updated = db.Column(db.DateTime(timezone=True), onupdate=func.now())

    enable = db.Column(db.Boolean, default=False, nullable=False)

    def __repr__(self):
        if self.enable:
            return "союз с {} заключён".format(self.user2)
        return "союз с {} не подписан".format(self.user2)


class User(db.Model):
    __tablename__ = "users"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(40), unique=False, nullable=False)
    email = db.Column(EmailType, unique=False)

    def add_user(self, user):
        """ добавление союза с user"""
        union = Union(user1=self, user2=user)
        db.session.add(union)
        db.session.flush()

Всё будет прекрасно до тех пор пока не потребуется у объектов User обратиться к списку союзных пользователей. Дело в том, что метол обращения к списку союзников выглядит по разному для разныз пользователей. Если для того чтобы узнать список союзников осла неоходимо выбрать все вторые колонки в строка в которых в первой колонке номер строки осла, а чтобы выбрать союзников козла, необходимо выбрать все первые колонки в строках где вторая колонка равна номеру строки козла. Таки образом получается метод выбора союзников осла и козла будет разный, а это не красиво.

Но всё таки можно реализовать универсальный метод поиск союзников для каждого пользователя и в SQL виде он будет выглядеть вот так

select * 
  from users 
  where id in (
    select user2_id id  
      from user_unions 
      where user1_id = 1 
    union 
      select user1_id id 
        from user_unions 
          where user2_id =  1
    ) 

тут производится выборка id союзников в ассоциативной таблице для обоих типов запросов, затем результаты объединяются и используются во вложенном запросе при обращении к таблице пользователей.

Sqlalchemy позволяет реализовать и такой достаточно сложный запрос с помощью объектного ORM и тогда метод выбора союзников пользователя будет выглядет следующим образом

class User(db.Model):
    __tablename__ = "users"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(40), unique=False, nullable=False)
    email = db.Column(EmailType, unique=False)

    created = db.Column(db.DateTime(timezone=True), server_default=func.now())


    def get_all_union_users(self):
        """выборка запросов на союз"""
        q1 = db.session.query(Union.user2_id.label("id"))
        q1 = q1.filter(Union.user1==self)
        q2 = db.session.query(Union.user1_id.label("id"))
        q2 = q1.filter(Union.user2==self)
        return User.query.filter(
            User.id.in_(q1.union(q2))
        )

    def get_union_users(self):
        """выборка подтверждённых союзников"""
        q1 = db.session.query(Union.user2_id.label("id"))
        q1 = q1.filter(Union.user1==self, Union.enable==True)
        q2 = db.session.query(Union.user1_id.label("id"))
        q1 = q2.filter(Union.user2==self, Union.enable==True)
        return User.query.filter(
            User.id.in_(q1.union(q2))
        )

Проверяем результат

# проверяем список союзников у каждого пользователя стандартным способом
for u in User.query.all():u, "союзники", u.unions.all()
... 
(Пользователь: осёл , 'союзники', [союз с Пользователь: козёл  заключён, союз с Пользователь: мишка  заключён])
(Пользователь: козёл , 'союзники', [])
(Пользователь: мишка , 'союзники', [])
(Пользователь: пятачёк , 'союзники', [союз с Пользователь: винипух  заключён])
(Пользователь: винипух , 'союзники', [])


# проверяем список союзников у каждого пользователя проверяем через универсальный метод
>>> for u in User.query.all():u, "союзники", u.get_union_users().all()
... 
(Пользователь: осёл , 'союзники', [Пользователь: козёл , Пользователь: мишка ])
(Пользователь: козёл , 'союзники', [Пользователь: осёл ])
(Пользователь: мишка , 'союзники', [Пользователь: осёл ])
(Пользователь: пятачёк , 'союзники', [Пользователь: винипух ])
(Пользователь: винипух , 'союзники', [Пользователь: пятачёк ])