def start_stream(self) -> None:
asyncio.create_task(StreamEvent(self)())await context.test.wait_message_type(TestModel)
#Есть какая-то тестовая модель
@dataclass
class TestModel(AbstractClass):
test_id: str = ''
id: str = ''
message_type: str = ''
content: MessageContentModel = field(default_factory=MessageContentModel)
text: str = ''
author: UserInfoModel = field(default_factory=UserInfoModel)
timestamp: str = ''
is_read: bool = False
# Мне пришел какой-то json объект, который соответствует этой модели
result = {
'id': '1',
'testId': '112',
'messageType': 'image',
'content': {
'fileId': 'AO6rgpHdvMeBoAGTzuX7x7Wrg_gBAAA',
'fileName': "utf-8''jpeg.jpeg",
'fileSize': 129167
},
'author': {
'id': '0000495',
'name': 'Гость'
},
'timestamp': '2019-05-27T13:39:44.204Z',
'isRead': False
}
# я его просто преобразую в свою модель:
result_model = TestModel.init_from_data(result)
# Получается питон объект
TestModel(test_id='112', id='1', message_type='image', content=MessageContentModel(text='', file_id='AO6rgpHdvMeBoAGTzuX7x7Wrg_gBAAA', file_name="utf-8''jpeg.jpeg", file_size=129167, widget_id='', preview=''), text='', author=UserInfoModel(id='0000495', name='Гость', avatar=''), timestamp='2019-05-27T13:39:44.204Z', is_read=False)
@classmethod
def init_from_data(cls, data: dict):
# какая-нибудь магия.
return cls(**result)b = [1 if True else None or 2 if True else None ... ]для этого есть операторНет он для того что бы шарить переменные между областями видимости. Блоки условного оператора, не влияют на области видимости.
global
, который делает переменную видимой для выполнения в блоках условного оператора и циклах.
@dataclass
class MultiFilter:
functions: list = field(default_factory=list)
mode: str = 'judge_any'
iterable: list = field(default_factory=list, init=False)
def __post_init__(self):
self.pos = len(self.functions) // 2
def judge_half(self, v):
return sum(map(lambda f: f(v), self.functions)) > self.pos
def judge_any(self, v):
return any(map(lambda f: f(v), self.functions))
def judge_all(self, v):
return all(map(lambda f: f(v), self.functions))
def __call__(self, v, mode='judge_any'):
return getattr(self, mode or self.mode)(v)
def __iter__(self):
return (v for v in self.iterable if getattr(self, self.mode)(v))
custom_filter = MultiFilter([
lambda x: x % 2 != 0,
lambda x: x > 10
])
print(list(filter(custom_filter, range(20))))
> [1, 3, 5, 7, 9, 11, 12, 13, 14, 15, 16, 17, 18, 19]
custom_filter.iterable = range(20)
print(list(custom_filter))
> [1, 3, 5, 7, 9, 11, 12, 13, 14, 15, 16, 17, 18, 19]