Skip to content

Instantly share code, notes, and snippets.

@Gobot1234
Created January 22, 2022 12:47
Show Gist options
  • Save Gobot1234/f03100a04ec81b8db2d8c6c3f25cf6ca to your computer and use it in GitHub Desktop.
Save Gobot1234/f03100a04ec81b8db2d8c6c3f25cf6ca to your computer and use it in GitHub Desktop.
import asyncio
from typing import TypeVar, Generic
T = TypeVar("T")
AnyMsg = TypeVar("AnyMsg", "Msg", "Msg2")
class Msg(Generic[T]):
body: T
class Msg2(Generic[T]):
body: T
class Request:
id: int
item_ids: list[int]
def sync_gc_wait_for(enum: int, check: "(AnyMsg) -> bool", timeout: float) -> AnyMsg:
...
async def gc_wait_for(enum: int, check: "(AnyMsg) -> bool", timeout: float) -> AnyMsg:
...
ItemCustomizationNotification = 123
item_id = 1234
async def main():
notification: Msg[Request] = sync_gc_wait_for(
ItemCustomizationNotification,
check=lambda msg: msg.body.id == 12345 and msg.body.item_ids[0] == item_id,
timeout=30,
)
reveal_type(notification)
notification: Msg[Request] = await gc_wait_for(
ItemCustomizationNotification,
check=lambda msg: (msg.body.id == 12345 and msg.body.item_ids[0] == item_id),
timeout=30,
)
reveal_type(notification)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment