Show HN: Reactive Signals for Python – inspired by Angular's reactivity model | Hacker News Reader

Show HN: Reactive Signals for Python – inspired by Angular's reactivity model

github.com

31 pointsbuibuibui19 comments

Hey everyone, I built reaktiv, a small reactive signals library for Python, inspired by Angular’s reactivity model. It lets you define Signals, Computed Values, and Effects that automatically track dependencies and update efficiently. The main focus is async-first reactivity without external dependencies.

Here is an example code:

``` import asyncio from reaktiv import Signal, ComputeSignal, Effect

async def main(): count = Signal(0) doubled = ComputeSignal(lambda: count.get() * 2)

    async def log_count():
        print(f"Count: {count.get()}, Doubled: {doubled.get()}")

    Effect(log_count).schedule()
    count.set(5)  # Triggers: "Count: 5, Doubled: 10"
    await asyncio.sleep(0)  # Allow effects to process

asyncio.run(main()) ```

Open articleView on HN

For those of use that haven't use Angular, can you explain what this is doing? And what's it do that we can't already do with asyncio? Also, it's weird you have to call `asyncio.sleep(0)`. Seems like you should be able to `await count.set(5)`.

I did look at the repo, but I couldn't easily understand the use case and what exactly it's doing from examples alone.

I’ve seen how the frontend world solved state management with reactivity, but I couldn’t find anything similar for backend development in Python. So I built reaktiv to bring the same automatic dependency tracking and state propagation to async Python code.

In frontend frameworks, Signals eliminate the need for manual subscriptions and event handling, making state updates more efficient. I wanted the same benefits for backend systems where changes to shared state should propagate automatically, without polling, callbacks, or race conditions.

For example, if a config value or cached result updates, anything depending on it should react immediately. With Signals, you don’t have to manually notify consumers or keep track of what depends on what. It just works (hopefully).

About `asyncio.sleep(0)`: it’s just there to keep the example concise. Normally, you’d be inside an async function that naturally yields control, like handling a request or waiting for I/O.

backend has cache and database for state management. for more complex cases, better use temporal, dbos etc.

How many caches and how many databases? I’ve got more than I’d like of either and making sure the right things are happening at the right time is not as trivial as you make it sound.

https://github.com/ReactiveX/RxPY

As an Angular/RxJS developer, RxPY was the first library I tried in my projects (about two years ago, I think). However, it had too many footguns that I didn’t encounter with RxJS. Because of that, I opted for simple asyncio Queues instead. Now that Angular is transitioning from RxJS to Signals, I was curious if something similar exists in the Python ecosystem.

What kinds of Python programs would benefit from Signals? Most of the Python I've written is either small CLI tools that run to completion, or boring web servers using frameworks like Django that are request-response oriented, which also has a "run-to-completion" model for a request's handler code.

Also, I looked at your implementation, and your ComputedSignal implementation is not sound - it can give inconsistent results with respect to the input signals depending on the ordering of reads when you have a ComputedSignal that depends on other ComputedSignal.

In my case, I’ve built Python applications that poll multiple sensor data, perform calculations, and push results to the frontend. The frontend is also allowed to configure the measurements on the fly, so I needed some form of reactivity in my backend to handle dynamic changes smoothly.

Could you share an example of the problem you’re referring to?

I didn’t have time to work up a repro, but the issue is a classic one with the basic algorithm you use for your ComputeSignal graph. The issue is that during the fan-out after a change at a root, a ComputeSignal that’s recomputing can read a stale out of date value from a ComputeSignal that has not yet recomputed for the change.

The simplest solution to implement is to do two phase propagation, the first phase marks all ComputeSignal as “dirty” so any read of them must re-compute. Then in the second phase you do what you’re already doing which is to walk down, re-computing, and stop at ComputeSignals that didn’t change. There are other approaches like using a logical clock that have different balances between time complexity at write time and time complexity at read time.

Besides the correctness issue, the eager depth first strategy can re-compute the same ComputeSignal multiple times for a single change to a root. This can also be solved with the dirty phase, because once a dirty ComputeSignal becomes clean you don’t need to re-compute it again during that write’s fan-out.

Shiny from the above commenter is a good example of the dirty / “invalidate” style of algorithm.

https://github.com/posit-dev/py-shiny/blob/main/shiny/reacti...

There are several high-level web application frameworks for Python that are based around this concept. It's particularly useful for data-oriented apps that let the user tweak parameters and update (arbitrarily complex) calculations and visualizations in response. I personally work on one called Shiny (https://shiny.posit.co/py/) but there are others including Reflex.dev and Solara.dev.

(I haven't looked at Reaktiv beyond the readme but it's clearly based on the same concepts, albeit it's "only" the reactive primitives and doesn't provide the rest of the stack like the frameworks I mentioned do)

Cool (probably), but it's not clear what it does or how useful it is when one doesn't know anything about Angular signals. Could you explain?

Also, care to compare to RxPY? (https://github.com/ReactiveX/RxPY)

Haha, I actually used RxPY a lot in the past! Back when I was more into Angular/RxJS and less familiar with Python, I thought I could apply the same concurrency patterns using RxPY. But I burned myself pretty badly. RxPY is powerful, but it’s also complex, and getting it to play nicely with asyncio (especially with typing) was frustrating enough that I eventually moved away from it.

RxPY is a full reactive programming toolkit with observables, operators, and complex stream processing, while Signals are a more lightweight and declarative way to track state dependencies. RxPY is great for handling event streams, but for simple state management and reactive updates, Signals are much easier to reason about.

I don’t know how to edit my submission, but after reading the comments, I realize the example code didn’t do a good job of demonstrating what my code does. Here’s a more verbose example showing a stock ticker:

  import asyncio
  from reaktiv import Signal, ComputeSignal, Effect
  
  async def main():
      # Real-time stock prices
      apple_price = Signal(195.00)
      google_price = Signal(2750.00)
  
      # User's portfolio
      shares = Signal({
          'AAPL': 100,
          'GOOGL': 50
      })
  
      # Computed total portfolio value
      portfolio_value = ComputeSignal(lambda: (
          shares.get()['AAPL'] * apple_price.get() +
          shares.get()['GOOGL'] * google_price.get()
      ))
  
      # Price alert system
      async def check_alerts():
          if apple_price.get() > 200:
              print("AAPL alert: Above $200!")
          if google_price.get() < 2700:
              print("GOOGL alert: Below $2700!")
  
      # Automatic updates
      async def live_updates():
          while True:
              await asyncio.sleep(1)
              apple_price.set(apple_price.get() * 1.01)  # +1%
              google_price.set(google_price.get() * 0.995)  # -0.5%
              print(f"AAPL: ${apple_price.get():,.2f}  GOOGL: ${google_price.get():,.2f}")
  
      # Track portfolio value
      async def monitor_portfolio():
          print(f"Current value: ${portfolio_value.get():,.2f}")
  
      # Set up effects
      alerts_effect = Effect(check_alerts)
      updates_effect = Effect(live_updates)
      portfolio_effect = Effect(monitor_portfolio)
  
      alerts_effect.schedule()
      updates_effect.schedule()
      portfolio_effect.schedule()
  
      # Run for 5 seconds
      await asyncio.sleep(5)
  
  asyncio.run(main())

Example output:

  Current value: $157,000.00
  AAPL: $196.95  GOOGL: $2,736.25
  Current value: $156,507.50
  AAPL: $198.92  GOOGL: $2,722.57
  Current value: $156,020.39
  AAPL: $200.91  GOOGL: $2,708.96
  AAPL alert: Above $200!
  Current value: $155,538.66
  AAPL: $202.92  GOOGL: $2,695.41
  AAPL alert: Above $200!
  GOOGL alert: Below $2700!

HN doesn't support markdown unfortunately. As such your code block's formatting is all messed up there.

Not your fault of course. Just thought you'd wanna know.

P.S. - HN staff please add markdown support. I'm sure this gets mentioned 10 times a day, so this is mention 11.

All you need to do is put 2 spaces before your code. I'm sure OP could have managed.

  Like this

https://news.ycombinator.com/formatdoc

I'm aware. Markdown is more popular and (in my opinion) easier to use. Tapping two spaces before each line of my code is a lot more effort than typing 3 backticks before and after the whole snippet. You're asking me to track indentation, on a potentially already indentation-based language (I mostly paste Python), without the help of an IDE. No thanks.

Sure, they could have figured it out. But it's annoying.

Ah, I didn't knew about that. I assumed HN also used Markdown.

Reply on news.ycombinator.com