Skip to content

Deprecated

This API is considered legacy due to bad design decisions. A better API will be introduced in the next minor version (v0.8.0). You can continue using this API for now, but it is recommended to switch to the new API once it is available.

from reactivity.hmr.api import *

LifecycleMixin

Bases: BaseReloader

Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
 
class LifecycleMixin(BaseReloader):
    def run_with_hooks(self):
        self._original_main_module = sys.modules["__main__"]
        sys.modules["__main__"] = self.entry_module
        call_pre_reload_hooks()
        self.effect = HMR_CONTEXT.effect(self.run_entry_file)
        call_post_reload_hooks()

    def clean_up(self):
        self.effect.dispose()
        self.entry_module.load.dispose()
        self.entry_module.load.invalidate()
        sys.modules["__main__"] = self._original_main_module

run_with_hooks 

run_with_hooks()
Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
 8
 9
10
11
12
13
 
def run_with_hooks(self):
    self._original_main_module = sys.modules["__main__"]
    sys.modules["__main__"] = self.entry_module
    call_pre_reload_hooks()
    self.effect = HMR_CONTEXT.effect(self.run_entry_file)
    call_post_reload_hooks()

clean_up 

clean_up()
Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
15
16
17
18
19
 
def clean_up(self):
    self.effect.dispose()
    self.entry_module.load.dispose()
    self.entry_module.load.invalidate()
    sys.modules["__main__"] = self._original_main_module

SyncReloaderAPI

Bases: SyncReloader, LifecycleMixin

Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
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
 
class SyncReloaderAPI(SyncReloader, LifecycleMixin):
    def __enter__(self):
        from threading import Thread

        self.run_with_hooks()
        self.thread = Thread(target=self.start_watching)
        self.thread.start()
        return super()

    def __exit__(self, *_):
        self.stop_watching()
        self.thread.join()
        self.clean_up()

    async def __aenter__(self):
        from asyncio import ensure_future, sleep, to_thread

        await to_thread(self.run_with_hooks)
        self.future = ensure_future(to_thread(self.start_watching))
        await sleep(0)
        return super()

    async def __aexit__(self, *_):
        self.stop_watching()
        await self.future
        self.clean_up()

__enter__ 

__enter__()
Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
23
24
25
26
27
28
29
 
def __enter__(self):
    from threading import Thread

    self.run_with_hooks()
    self.thread = Thread(target=self.start_watching)
    self.thread.start()
    return super()

__exit__ 

__exit__(*_)
Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
31
32
33
34
 
def __exit__(self, *_):
    self.stop_watching()
    self.thread.join()
    self.clean_up()

__aenter__ async 

__aenter__()
Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
36
37
38
39
40
41
42
 
async def __aenter__(self):
    from asyncio import ensure_future, sleep, to_thread

    await to_thread(self.run_with_hooks)
    self.future = ensure_future(to_thread(self.start_watching))
    await sleep(0)
    return super()

__aexit__ async 

__aexit__(*_)
Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
44
45
46
47
 
async def __aexit__(self, *_):
    self.stop_watching()
    await self.future
    self.clean_up()

AsyncReloaderAPI

Bases: AsyncReloader, LifecycleMixin

Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
class AsyncReloaderAPI(AsyncReloader, LifecycleMixin):
    def __enter__(self):
        from asyncio import run
        from threading import Event, Thread

        self.run_with_hooks()

        e = Event()

        async def task():
            e.set()
            await self.start_watching()

        self.thread = Thread(target=lambda: run(task()))
        self.thread.start()
        e.wait()
        return super()

    def __exit__(self, *_):
        self.stop_watching()
        self.thread.join()
        self.clean_up()

    async def __aenter__(self):
        from asyncio import ensure_future, sleep, to_thread

        await to_thread(self.run_with_hooks)
        self.future = ensure_future(self.start_watching())
        await sleep(0)
        return super()

    async def __aexit__(self, *_):
        self.stop_watching()
        await self.future
        self.clean_up()

__enter__ 

__enter__()
Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
def __enter__(self):
    from asyncio import run
    from threading import Event, Thread

    self.run_with_hooks()

    e = Event()

    async def task():
        e.set()
        await self.start_watching()

    self.thread = Thread(target=lambda: run(task()))
    self.thread.start()
    e.wait()
    return super()

__exit__ 

__exit__(*_)
Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
68
69
70
71
 
def __exit__(self, *_):
    self.stop_watching()
    self.thread.join()
    self.clean_up()

__aenter__ async 

__aenter__()
Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
73
74
75
76
77
78
79
 
async def __aenter__(self):
    from asyncio import ensure_future, sleep, to_thread

    await to_thread(self.run_with_hooks)
    self.future = ensure_future(self.start_watching())
    await sleep(0)
    return super()

__aexit__ async 

__aexit__(*_)
Source code in .venv/lib/python3.12/site-packages/reactivity/hmr/api.py 
81
82
83
84
 
async def __aexit__(self, *_):
    self.stop_watching()
    await self.future
    self.clean_up()