payment/payment_backend/repositories/user.py

import itertools

from custom_decorators import singleton
from database import Database
from models import User


@singleton
class UserRepository:
    def __init__(self, config):
        self.db = Database(config['MYSQL'])

    def get_and_update(self, user):
        users = []
        cursor = self.db.execute_query(
            *user.select_sql(query_columns=["id", "name", "phone", "email", "wallet_address", "payment_method"],
                             condition="OR")
        )
        same_users = cursor.fetchall()
        new_user = not len(same_users)
        # 对用户已存在的属性判断是否有新属性
        update_user = set()
        update_sqls = []
        update_params_list = []
        delete_params = []
        exist_conflicting_attr = False
        for same_user in same_users:
            same_user = User(*same_user)
            exist_conflicting_attr = False
            different_attrs = user.get_difference(same_user)
            # 用于判断是否有新属性
            update_sql_params = []
            update_params = []
            for k, v in different_attrs.items():
                new_attr, exist_attr = v
                if exist_attr is None:
                    setattr(same_user, k, new_attr)
                    update_sql_params.append(f"{k}=%s")
                    update_params.append(new_attr)
                elif new_attr is not None:
                    # 出现冲突的属性，考虑新增一行记录
                    exist_conflicting_attr = True
                    break

            if same_user in update_user:
                delete_params.append((same_user.id,))
            else:
                users.append(same_user)
            exist_new_attr = bool(update_params)
            if exist_new_attr:
                update_user.add(same_user)
                update_sqls.append(f'UPDATE users SET {",".join(update_sql_params)} WHERE id=%s;')
                update_params.append(same_user.id)
                update_params_list.append(update_params)
        sql_flag = False
        try:
            if delete_params:
                sql_flag = True
                self.db.get_connection().cursor().executemany("DELETE FROM users WHERE id=%s", delete_params)
            if update_user:
                sql_flag = True
                self.db.get_connection().cursor().execute("".join(update_sqls),
                                                          list(itertools.chain.from_iterable(update_params_list)),
                                                          multi=True)
            if sql_flag:
                self.db.commit()
        except Exception:
            self.db.rollback()
            raise
        if new_user or exist_conflicting_attr:
            try:
                self.db.execute_query(*user.insert_sql())
                self.db.commit()
            except Exception:
                self.db.rollback()
                raise
            users.append(user)
        return users

    def exist(self, user):
        cursor = self.db.execute_query(*user.exists_sql())
        exist = cursor.fetchone()
        cursor.close()
        return bool(exist)