PyCafe - Solara - Real-Time BTC Price Tracker with Solara

app.py
requirements.txt
app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
import solara
import traitlets
import asyncio
from anyio import to_thread
import httpx


# reactive vars are per kernel, so we use
# traitlets, which solara can also observe

class Prices(traitlets.HasTraits):
    btc = traitlets.Float()


prices = Prices(clicks=0)

async def add_clicks_from_task():
    while 1:
        try:
            def update():
                url = "https://api.binance.com/api/v1/ticker/price?symbol=BTCUSDT"
                response = httpx.get(url)
                prices.btc = float(response.json()["price"])
            # on CPython, if we call it from the main event look thread
            # this will fail. Note this will also work on pycafe
            await to_thread.run_sync(update)
        except Exception as e:
            print(e)
        await asyncio.sleep(1)

# run a task in the background
asyncio.create_task(add_clicks_from_task())




def use_trait_observe(has_trait_object, name):
    """Should be fixed in solara > 1.43.0, so you can use solara.use_trait_observe
    """
    _, set_counter = solara.use_state(0)

    def connect():
        def update(change):
            set_counter(lambda x: x+1)

        has_trait_object.observe(update, name)

        def cleanup():
            has_trait_object.unobserve(update, name)

        return cleanup

    solara.use_effect(connect, [has_trait_object, name])
    return getattr(has_trait_object, name)



@solara.component
def Page():
    # make this component update when prices.btc changes
    use_trait_observe(prices, "btc")

    with solara.Card("BTC price", style={"max-width": "280px"}):
        solara.Preformatted(f"$ {prices.btc:,}")
Py.Cafe

solara-shared-state

Real-Time BTC Price Tracker with Solara
