Last active
August 26, 2024 10:56
-
-
Save sina-salahshour/21a2f05c4a336819d645a43e3e0a9ea6 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from typing import Callable | |
| from dataclasses import dataclass | |
| type Subscription = Callable[[], None] | |
| @dataclass(eq=False) | |
| class Dependency: | |
| subscriptions: set[Subscription] | |
| class State: | |
| CONTEXT: list[Subscription] = [] | |
| CONTEXT_DEPS: list[Dependency] | None = None | |
| def create_signal[T](value: T) -> tuple[Callable[[T], None], Callable[[], T]]: | |
| dependency = Dependency(set()) | |
| def setter(next_value: T): | |
| nonlocal value | |
| if value != next_value: | |
| value = next_value | |
| react_subscribers() | |
| def getter(): | |
| if State.CONTEXT and State.CONTEXT_DEPS is not None: | |
| State.CONTEXT_DEPS.append(dependency) | |
| dependency.subscriptions.add(State.CONTEXT[-1]) | |
| return value | |
| def react_subscribers(): | |
| subs = dependency.subscriptions.copy() | |
| dependency.subscriptions.clear() | |
| for subscriber in subs: | |
| subscriber() | |
| return (setter, getter) | |
| def effect(func: Callable[[], None]): | |
| previous_dependencies: list[Dependency] | None = [] | |
| dependencies: list[Dependency] | None = [] | |
| def execute(): | |
| nonlocal dependencies | |
| nonlocal previous_dependencies | |
| try: | |
| dependencies, State.CONTEXT_DEPS = State.CONTEXT_DEPS, dependencies | |
| State.CONTEXT.append(execute) | |
| func() | |
| finally: | |
| State.CONTEXT.pop() | |
| dependencies, State.CONTEXT_DEPS = State.CONTEXT_DEPS, dependencies | |
| if previous_dependencies is None or dependencies is None: | |
| raise AssertionError( | |
| "Dependency list is None. this should never happen." | |
| ) | |
| for dependency in previous_dependencies: | |
| if dependencies and dependency not in dependencies: | |
| dependency.subscriptions.remove(execute) | |
| previous_dependencies.clear() | |
| dependencies, previous_dependencies = previous_dependencies, dependencies | |
| execute() | |
| def untrack[T](func: Callable[[], T]) -> T: | |
| tmp = [] | |
| tmp, State.CONTEXT = State.CONTEXT, tmp | |
| try: | |
| return func() | |
| finally: | |
| tmp, State.CONTEXT = State.CONTEXT, tmp | |
| # usage ---------------------------------------------------------------------- | |
| set_show, show = create_signal(False) | |
| set_name, name = create_signal("seth") | |
| print("-- effect start --") | |
| @effect | |
| def conditional_print(): | |
| print("call shod", show()) | |
| if show(): | |
| print(name()) | |
| print("-- effect end --") | |
| print("-- name change --") | |
| set_name("sina") # hichi | |
| print("-- show flag change --") | |
| set_show(True) # sina print mishe | |
| print("-- name change --") | |
| set_name("jamile") | |
| print("-- show flag change --") | |
| set_show(False) # call mishe vali print nemishe | |
| print("-- name change --") | |
| set_name("hasan") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment