sveltish 1.1.2

A Svelte Stores implementation in Python.

Sveltish

Svelte Stores are one of the secret weapons of the Svelte framework (the recently voted most loved web framework).

Stores allow easy reactive programming by presenting an Observer pattern that is as simple as necessary, but not simpler.

Install

pip install sveltish

How to use

Sometimes, you’ll have values that need to be accessed by multiple unrelated objects.

For that, you can use stores. It is a very simple implementation (around 100 lines of code) of the Observer/Observable pattern.

A store is simply an object with a subscribe method that allows interested parties to be notified when its value changes.

Writable Stores

from sveltish.stores import Writable

count = Writable(0)
history = []  # logging for testing
# subscribe returns an unsubscriber
def record(x): 
    history.append(x)
    print(history)
stop = count.subscribe(record)
count

[0]

Writable(0)

We just created a store, count. Its value can be accessed via a callback we pass in the count.subscribe method:

A Writable can be set from the outside. When it happens, all its subscribers will react.

def increment(): count.update(lambda x: x + 1)
def decrement(): count.update(lambda x: x - 1)
def reset(): count.set(0)

count.set(3)
increment()
decrement()
decrement()
reset()
count.set(42)


test_eq(history, [0, 3, 4, 3, 2, 0, 42])

[0, 3]
[0, 3, 4]
[0, 3, 4, 3]
[0, 3, 4, 3, 2]
[0, 3, 4, 3, 2, 0]
[0, 3, 4, 3, 2, 0, 42]

The unsubscriber, in this example the stop function, stops the notifications to the subscriber.

stop()
reset()
count.set(22)
test_eq(history, [0, 3, 4, 3, 2, 0, 42])
count

Writable(22)

Notice that you can still change the store but there was no print message this time. There was no observer listening.

Note

Observer, Subscriber and Callback are used as synomyms here.

When we subscribe new callbacks, they will be promptly informed of the current state of the store.

stop  = count.subscribe(lambda x: print(f"Count is now {x}"))
stop2 = count.subscribe(lambda x: print(f"double of count is {2*x}"))

Count is now 22
double of count is 44

reset()

double of count is 0
Count is now 0

stop()
stop2()

You can create an empty Writable Store.

store = Writable()
history = []
unsubscribe = store.subscribe(lambda x: history.append(x))
unsubscribe()
test_eq(history, [None])

If you try to unsubscribe twice, it won’t break. It just does nothing the second time… and in the third time… and…

unsubscribe(), unsubscribe(), unsubscribe()

(None, None, None)

Stores assume mutable objects.

Note

In Python everythong is an object. Here we are calling an object something that is not a primitive (eg. int, bool, etc)

class Bunch:
    __init__ = lambda self, **kw: setattr(self, '__dict__', kw)

obj = Bunch()
called = 0
store = Writable(obj)
def callback(x):
    global called
    called += 1
stop = store.subscribe(callback)

test_eq(called, 1)
obj.a = 1 #type: ignore
store.set(obj)
test_eq(called, 2)

Readable Stores

However… It is clear that not all stores should be writable by whoever has a reference to them. Many times you want a single publisher of change in store that is only consumed (subscribed) by many other objects. For those cases, we have readable stores.

Note

The Publisher Subscriber (PubSub) pattern is a variant of the Observable/Observer pattern.

from sveltish.stores import Readable

# from sveltish.stores import Readable
# from threading import Event, Thread
# import time


# def start(set): # the start function is the publisher
#     stopped = Event()
#     def loop(): # needs to be in a separate thread
#         while not stopped.wait(1): # in seconds
#             set(time.localtime())
#     Thread(target=loop).start()    
#     return stopped.set
   
# now = Readable(time.localtime(), start)

Note

The loop needs to be in its own thread, otherwise the function would never return and we would wait forever.

# now

While there is no subscriber, the Readable will not be updated.

# now

# OhPleaseStop = now.subscribe(lambda x: print(time.strftime(f"%H:%M:%S", x), end="\r"))

# time.sleep(3)
# OhPleaseStop()

A Readable store without a start function is a constant value and has no meaning for us. Therefore, start is a required argument.

try:
    c = Readable(0) # shoud fail
except Exception as error:
    print(error)

__init__() missing 1 required positional argument: 'start'

Note

The Svelte Store api allow you to create a Readable Store without a Notifier. See discussion here.

Derived Stores

A Derived Store stores a value based on the value of another store.

from sveltish.stores import Derived

count = Writable(1)
stopCount = count.subscribe(lambda x: print(f"count is {x}"))
double = Derived(count, lambda x: x * 2)
stopDouble = double.subscribe(lambda x: print(f"double is {x}"))
test_eq(double.get(), 2*count.get())

count is 1
double is 2

count.set(2)
test_eq(double.get(), 4)

double is 4
count is 2

Building on our previous example, we can create a store that derives the elapsed time since the original store was started.

# def calc_elapsed(then):
#     now = time.localtime()
#     return time.mktime(now) - time.mktime(then)

# elapsed = Derived(now, calc_elapsed)

# stopElapsed = elapsed.subscribe(lambda x: print(f"Elapsed time: {x} seconds.", end="\r"))

# time.sleep(10)
# stopElapsed()

Derived stores allow us to transform the value of a store. In RxPy they are called operators. You can build several operators like: filter, fold, map, zip…

Let’s build a custom filter operator:

user = Writable({"name": "John", "age": 32})
stopLog = user.subscribe(lambda x: print(f"User: {x}"))

User: {'name': 'John', 'age': 32}

name = Derived(user, lambda x: x["name"])
stopName = name.subscribe(lambda x: print(f"Name: {x}"))

Name: John

user.update(lambda x: x | {"age": 45})

User: {'name': 'John', 'age': 45}

Updating the age does not trigger the name subscriber.

user.update(lambda x: x | {"name": "Fred"})

User: {'name': 'Fred', 'age': 45}
Name: Fred

Only changes to the name of the user triggers the name subscriber.

Another cool thing about Derived Stores is that you can derive from a list of stores. Let’s build a zip operator.

a = Writable([1,2,3,4])
b = Writable([5,6,7,8])
a,b

(Writable([1, 2, 3, 4]), Writable([5, 6, 7, 8]))

zipper = Derived([a,b], lambda a,b: list(zip(a,b)))
test_eq(zipper.get(), [(1, 5), (2, 6), (3, 7), (4, 8)])

a.set([4,3,2,1])
test_eq(zipper.get(), [(4, 5), (3, 6), (2, 7), (1, 8)])