A state with a background task.
| 1677 | |
| 1678 | |
| 1679 | class BackgroundTaskState(BaseState): |
| 1680 | """A state with a background task.""" |
| 1681 | |
| 1682 | order: List[str] = [] |
| 1683 | dict_list: Dict[str, List[int]] = {"foo": [1, 2, 3]} |
| 1684 | |
| 1685 | @xt.var |
| 1686 | def computed_order(self) -> List[str]: |
| 1687 | """Get the order as a computed var. |
| 1688 | |
| 1689 | Returns: |
| 1690 | The value of 'order' var. |
| 1691 | """ |
| 1692 | return self.order |
| 1693 | |
| 1694 | @xt.background |
| 1695 | async def background_task(self): |
| 1696 | """A background task that updates the state.""" |
| 1697 | async with self: |
| 1698 | assert not self.order |
| 1699 | self.order.append("background_task:start") |
| 1700 | |
| 1701 | assert isinstance(self, StateProxy) |
| 1702 | with pytest.raises(ImmutableStateError): |
| 1703 | self.order.append("bad idea") |
| 1704 | |
| 1705 | with pytest.raises(ImmutableStateError): |
| 1706 | # Even nested access to mutables raises an exception. |
| 1707 | self.dict_list["foo"].append(42) |
| 1708 | |
| 1709 | with pytest.raises(ImmutableStateError): |
| 1710 | # Direct calling another handler that modifies state raises an exception. |
| 1711 | self.other() |
| 1712 | |
| 1713 | with pytest.raises(ImmutableStateError): |
| 1714 | # Calling other methods that modify state raises an exception. |
| 1715 | self._private_method() |
| 1716 | |
| 1717 | # wait for some other event to happen |
| 1718 | while len(self.order) == 1: |
| 1719 | await asyncio.sleep(0.01) |
| 1720 | async with self: |
| 1721 | pass # update proxy instance |
| 1722 | |
| 1723 | async with self: |
| 1724 | # Methods on ImmutableMutableProxy should return their wrapped return value. |
| 1725 | assert self.dict_list.pop("foo") == [1, 2, 3] |
| 1726 | |
| 1727 | self.order.append("background_task:stop") |
| 1728 | self.other() # direct calling event handlers works in context |
| 1729 | self._private_method() |
| 1730 | |
| 1731 | @xt.background |
| 1732 | async def background_task_reset(self): |
| 1733 | """A background task that resets the state.""" |
| 1734 | with pytest.raises(ImmutableStateError): |
| 1735 | # Resetting the state should be explicitly blocked. |
| 1736 | self.reset() |
no outgoing calls