根据user_id、uid、bot_id查询数据
(
cls: Type[T_WavesUser],
session: AsyncSession,
uid: str,
user_id: str,
bot_id: str,
game_id: Optional[int] = None,
)
| 95 | @classmethod |
| 96 | @with_session |
| 97 | async def select_waves_user( |
| 98 | cls: Type[T_WavesUser], |
| 99 | session: AsyncSession, |
| 100 | uid: str, |
| 101 | user_id: str, |
| 102 | bot_id: str, |
| 103 | game_id: Optional[int] = None, |
| 104 | ) -> Optional[T_WavesUser]: |
| 105 | """ |
| 106 | 根据user_id、uid、bot_id查询数据 |
| 107 | """ |
| 108 | filters = [ |
| 109 | cls.user_id == user_id, |
| 110 | cls.uid == uid, |
| 111 | cls.bot_id == bot_id, |
| 112 | ] |
| 113 | if game_id is not None: |
| 114 | filters.append(cls.game_id == game_id) |
| 115 | sql = select(cls).where(*filters) |
| 116 | result = await session.execute(sql) |
| 117 | data = result.scalars().all() |
| 118 | return data[0] if data else None |
| 119 | |
| 120 | @classmethod |
| 121 | @with_session |
no outgoing calls
no test coverage detected