Method _link_to

reflex/istate/shared.py:211–248

Link this shared state to a token. After linking, subsequent access to this shared state will affect the linked token's state, and cause changes to be propagated to all other clients linked to that token. Args: token: The token to link to (Cannot contain

(self, token: str)

        raise ReflexRuntimeError(msg)

    async def _link_to(self, token: str) -> Self:
        """Link this shared state to a token.

        After linking, subsequent access to this shared state will affect the
        linked token's state, and cause changes to be propagated to all other
        clients linked to that token.

        Args:
            token: The token to link to (Cannot contain underscore characters).

        Returns:
            The newly linked state.

        Raises:
            ReflexRuntimeError: If linking fails or token is invalid.
        """
        if not token:
            msg = "Cannot link shared state to empty token."
            raise ReflexRuntimeError(msg)
        if not isinstance(self, SharedState):
            msg = "Can only link SharedState instances."
            raise ReflexRuntimeError(msg)
        if self._linked_to == token:
            return self  # already linked to this token
        if self._linked_to and self._linked_to != token:
            # Disassociate from previous linked token since unlink will not be called.
            self._linked_from.discard(self.router.session.client_token)
        # TODO: Change StateManager to accept token + class instead of combining them in a string.
        if "_" in token:
            msg = f"Invalid token {token} for linking state {self.get_full_name()}, cannot use underscore (_) in the token name."
            raise ReflexRuntimeError(msg)

        # Associate substate with the given link token.
        state_name = self.get_full_name()
        if self._reflex_internal_links is None:
            self._reflex_internal_links = {}
        self._reflex_internal_links[state_name] = token
        return await self._internal_patch_linked_state(token, full_delta=True)

    async def _unlink(self):
        """Unlink this shared state from its linked token.