Conversation
|
|
||
|
|
||
| def run(): | ||
| loop = asyncio.get_event_loop() |
There was a problem hiding this comment.
посмотри новый синтакс юез лупа
|
|
||
| async def stop(self): | ||
| await self.out_queue.join() | ||
| for t in self._tasks: |
| self.out_queue.task_done() | ||
|
|
||
| async def start(self): | ||
| asyncio.create_task(self._worker()) |
| chat_id=upd[1], | ||
| attachment=upd[3], | ||
| text=upd[2], |
There was a problem hiding this comment.
лучше структурку для такого сделать с тайпингами
| def check_kicked(kicked_users: list, active_users: dict): | ||
| i = 0 | ||
| while i != len(active_users["participants"]): | ||
| if list(active_users["participants"][i].keys())[-1] in kicked_users: | ||
| active_users["participants"] = ( | ||
| active_users["participants"][:i] | ||
| + active_users["participants"][i + 1 :] | ||
| ) | ||
| else: | ||
| i += 1 | ||
| return active_users |
There was a problem hiding this comment.
как то очень неудобно выглядит
| state_photo = Column(Boolean) | ||
| state_in_game = Column(Boolean) | ||
| state_wait_votes = Column(Boolean) | ||
| new_pair = Column(JSONB) | ||
| first_votes = Column(BigInteger, default=0) | ||
| second_votes = Column(BigInteger, default=0) | ||
| state_send_photo = Column(Boolean) | ||
| voters = Column(JSONB) | ||
| amount_users = Column(BigInteger) | ||
| last_winner = Column(Text) |
There was a problem hiding this comment.
не нужно кучу флагов плодить
можно просто строковый стейт сделать и он тогда в 1 поле будет
иначе мб такое, что больше 1 флага тру
| chat_id = Column( | ||
| BigInteger, ForeignKey("game_session.chat_id", ondelete="CASCADE") | ||
| ) |
There was a problem hiding this comment.
тут игровая сессия должна быть, а не чат
| await self.disconnect(app) | ||
|
|
||
| async def make_userlist(self, chat_id): | ||
| async def make_userlist(self, chat_id, app): |
There was a problem hiding this comment.
тайпингов везде сильно не хватает
прогнать бы через mypy
| async def stop(self): | ||
| self.is_running = False | ||
| await self.poll_task | ||
| self.poll_task.cancel() |
There was a problem hiding this comment.
не правильно отменяешь таску, посмотри пример в доке, там еще подождать конца отмены нужно
| @@ -0,0 +1,113 @@ | |||
| import logging | |||
There was a problem hiding this comment.
фикстуры лучше по разным файлам раскидать в зависимости от того, какие сущности используются
| await store.bot_manager.handle_updates( | ||
| Update( | ||
| type="message_new", | ||
| object=UpdateObject( | ||
| id=1, | ||
| chat_id=1, | ||
| body="Загрузить фотографии!", | ||
| type="test_type", | ||
| ), | ||
| ) | ||
| ) | ||
| await store.bot_manager.handle_updates( | ||
| Update( | ||
| type="message_new", | ||
| object=UpdateObject( | ||
| id=1, chat_id=1, body="Начать игру!", type="test_type" | ||
| ), | ||
| ) | ||
| ) | ||
| await store.bot_manager.handle_updates( | ||
| Update( | ||
| type="message_new", | ||
| object=UpdateObject( | ||
| id=1, | ||
| chat_id=1, | ||
| body="Моя статистика!", | ||
| type="test_type", | ||
| ), | ||
| ) | ||
| ) | ||
| await store.bot_manager.handle_updates( |
There was a problem hiding this comment.
Для такого есть @pytest.parametrize()
| async with db_session.begin() as session: | ||
| for i in range(2): | ||
| new_user = ParticipantsModel( | ||
| name=f"olred{i}", | ||
| wins=0, | ||
| chat_id=1, | ||
| owner_id=1 + i, | ||
| photo_id=1, | ||
| access_key="dasda", | ||
| ) | ||
| session.add(new_user) | ||
| await session.commit() |
There was a problem hiding this comment.
такие вещи должны быть в фикстурах
2 participants
Изменения после ревью.