Context Manager

class kivy_trio.context.ContextVarContextManager(context_var=None, value=None)

Bases: object

Sets the context variable context_var to value when this object is entered in a with block.

The context variable context_var is restored to its previous value when the block is exited.

E.g.:

>>> from contextvars import ContextVar
>>> my_context = ContextVar('my_context')
>>> my_context.get(12)
12
>>> with ContextVarContextManager(my_context, 42):
...     print(my_context.get())
42
context_var: Optional[ContextVar] = None

A ContextVar instance that will be set to value, when we enter the with block.

value: Any = None

The value to which the context_var will be set to.

await kivy_trio.context.initialize_from_trio(clock=None)

Call this in a trio context to initialize Kivy and trio so kivy can execute async functions in this trio thread context e.g. when kivy_run_in_async() is used. Similar so that trio can call kivy. It must be called after kivy is running and its Clock is active.

Initializing Kivy/trio is only needed if trio is running in a separate thread than kivy or when there are multiple trio event loops running in different threads. However, initializing when they are in the same thread significantly improves performance (see initialize_shared_thread()).

Parameters:

clock (ClockBase) – An optional kivy clock, in case the default kivy clock is not used, e.g. during testing. If it’s None, Clock is used.

Return type:

None

E.g. the following example shows how kivy can run async code in trio when trio is running in a second thread. Notice trio calling this once kivy is running.

When run, pressing the button will show that the async function is running in the second trio thread:

import trio
from threading import Thread, get_ident
from kivy.app import App
from kivy.lang import Builder
from kivy_trio.to_trio import kivy_run_in_async_quiet
from kivy_trio.context import initialize_from_trio

class DemoApp(App):

    def build(self):
        return Builder.load_string(
            "Button:\n    on_release: app.do_something_async()")

    @kivy_run_in_async_quiet
    async def do_something_async(self):
        await trio.sleep(.1)
        print('trio thread ID:', get_ident())

    def _trio_thread_target(self):
        async def runner():
            await initialize_from_trio()
            # now that kivy thread is initialized with trio token,
            # do_something_async can be safely called
            while self.get_running_app() is not None:
                await trio.sleep(.1)

        trio.run(runner)

    def on_start(self):
        print('kivy thread ID:', get_ident())
        # start trio once the kivy Clock is running
        thread = Thread(target=self._trio_thread_target)
        thread.start()

DemoApp().run()
kivy_trio.context.initialize_kivy_thread(token, clock=None)

Initializes only the Kivy thread with the trio token similar to initialize_from_trio(), except that you manually call this in the kivy thread passing the trio token acquired seperately and that you should also call initialize_trio_thread() in the trio thread.

Parameters:

  • token (Optional[TrioToken]) – The TrioToken to use to run async code. Can be None if trio and kivy are running in the same thread and if called from a trio context, then we automatically get the token.

  • clock (ClockBase) – An optional kivy clock, in case the default kivy clock is not used, e.g. during testing. If it’s None, Clock is used.

Return type:

None

E.g.:

>>> # get the trio token for your trio event loop from the trio thread
>>> from trio.lowlevel import current_trio_token
>>> trio_token = current_trio_token()
>>> initialize_trio_thread()
>>> ...
>>> # later, in the kivy thread use the token to init the kivy thread
>>> initialize_kivy_thread(trio_token)
>>> # now you can use kivy_run_in_async(_quiet) to run async code
kivy_trio.context.initialize_shared_thread(token=None, clock=None)

Initializes the shared trio and kivy thread with the trio token and kivy clock similar to initialize_from_trio(), except that you manually call this in the kivy/trio thread passing the trio token (optionally).

When kivy and trio are running from the same thread, we don’t need to initialize either, however, initializing significanly imporves performance.

This must be called only after both kivy and trio are running and from the main shared thread under the trio context (unless you provide the token).

Parameters:

  • token (Optional[TrioToken]) – The TrioToken to use to run async code. Can be None, and then we automatically get the currently active token.

  • clock (ClockBase) – An optional kivy clock, in case the default kivy clock is not used, e.g. during testing. If it’s None, Clock is used.

Return type:

None

E.g.:

import trio
from kivy.app import App
from kivy.lang import Builder
from kivy_trio.to_trio import kivy_run_in_async_quiet
from kivy_trio.context import initialize_shared_thread

class DemoApp(App):

    def build(self):
        return Builder.load_string(
            "Button:

on_release: app.do_something_async()”)

@kivy_run_in_async_quiet async def do_something_async(self):

await trio.sleep(.1) print(‘ran async’)

def on_start(self):

initialize_shared_thread() # now you can run async code

trio.run(DemoApp().async_run, ‘trio’)

kivy_trio.context.initialize_trio_thread(token=None, clock=None)

Initializes only the trio thread with the trio token and clock similar to initialize_from_trio(), except that you manually call this in the trio thread passing the trio token (optionally) and that you must also call initialize_kivy_thread() in the kivy thread with the token.

Parameters:

  • token (Optional[TrioToken]) – The TrioToken to use to run async code. Can be None, and then we automatically get the currently active token.

  • clock (ClockBase) – An optional kivy clock, in case the default kivy clock is not used, e.g. during testing. If it’s None, Clock is used.

Return type:

None

E.g.:

>>> # get the trio token for your trio event loop from the trio thread
>>> from trio.lowlevel import current_trio_token
>>> trio_token = current_trio_token()
>>> initialize_trio_thread()
>>> ...
>>> # later, in the kivy thread use the token to init the kivy thread
>>> initialize_kivy_thread(trio_token)
>>> # now you can use kivy_run_in_async(_quiet) to run async code
kivy_trio.context.kivy_clock: ContextVar = <ContextVar name='kivy_clock'>

A ContextVar that contains the Clock kivy_trio uses to call into kivy to execute callbacks.

It will typically default to Clock in functions that need a kivy clock when it has not been initialized.

kivy_trio.context.kivy_thread: ContextVar = <ContextVar name='kivy_thread' default=None>

A ContextVar that contains the Clock, or None (the default) kivy_trio uses to determine whether the current thread is the kivy thread.

We only set its value to the current kivy Clock in the main kivy thread. Then, from any thread we compare its value to kivy_clock. If they are the same, we know this thread is the main Kivy thread. Otherwise, it’s a different thread and we use callbacks to schedule kivy code to execute in the main kivy thread.

with kivy_trio.context.kivy_thread_context(token, clock=None)

The same as initialize_kivy_thread(), but in context manager form so that the tokens are reset when exiting the context.

This ensures that any initialization is cleared when exiting so that it can be run again if needed with different kivy clock.

Parameters:

  • token (Optional[TrioToken]) – The TrioToken to use to run async code. Can be None if trio and kivy are running in the same thread and if called from a trio context, then we automatically get the token.

  • clock (ClockBase) – An optional kivy clock, in case the default kivy clock is not used, e.g. during testing. If it’s None, Clock is used.

E.g.:

>>> # get the trio token for your trio event loop from the trio thread
>>> import trio
>>> from trio.lowlevel import current_trio_token
>>> trio_token = None
>>> async def my_func():
...     global trio_token
...     trio_token = current_trio_token()
...     with trio_thread_context():
...         await trio.sleep(10000)
>>> trio.run(my_func)
>>> ...
>>> # now back in the kivy thread
>>> with kivy_thread_context(trio_token):
...     MyApp().run()
>>> ...
with kivy_trio.context.shared_thread_context(token=None, clock=None)

The same as initialize_shared_thread(), but in context manager form so that the tokens are reset when exiting the context.

This ensures that any initialization is cleared when exiting so that it can be run again if needed with different trio or kivy tokens.

Parameters:

  • token (Optional[TrioToken]) – The TrioToken to use to run async code. Can be None, and then we automatically get the currently active token.

  • clock (ClockBase) – An optional kivy clock, in case the default kivy clock is not used, e.g. during testing. If it’s None, Clock is used.

E.g.:

import trio
from kivy.app import App
from kivy.lang import Builder
from kivy_trio.to_trio import kivy_run_in_async_quiet
from kivy_trio.context import shared_thread_context

class DemoApp(App):

    def build(self):
        return Builder.load_string(
            "Button:

on_release: app.do_something_async()”)

@kivy_run_in_async_quiet async def do_something_async(self):

await trio.sleep(.1) print(‘ran async’)

async def async_run(self, async_lib=None):
with shared_thread_context():

await super(DemoApp, self).async_run(‘trio’)

trio.run(DemoApp().async_run, ‘trio’)

kivy_trio.context.trio_entry: ContextVar = <ContextVar name='trio_entry'>

A ContextVar that contains the TrioToken kivy_trio uses to call into trio to execute async code.

kivy_trio.context.trio_thread: ContextVar = <ContextVar name='trio_thread' default=None>

A ContextVar that contains the TrioToken, or None (the default) kivy_trio uses to determine whether the current thread is the trio thread that runs trio_entry.

We only set its value to the current TrioToken in a trio thread. Then, from any thread we compare its value to trio_entry. If they are the same, we know this thread is the trio thread that generated trio_entry. Otherwise, it’s a different thread and we use safe callbacks to schedule trio code to execute in that thread.

with kivy_trio.context.trio_thread_context(token=None, clock=None)

The same as initialize_trio_thread(), but in context manager form so that the tokens are reset when exiting the context.

This ensures that any initialization is cleared when exiting so that it can be run again if needed with different trio tokens.

Parameters:

  • token (Optional[TrioToken]) – The TrioToken to use to run async code. Can be None, and then we automatically get the currently active token.

  • clock (ClockBase) – An optional kivy clock, in case the default kivy clock is not used, e.g. during testing. If it’s None, Clock is used.

E.g.:

>>> # get the trio token for your trio event loop from the trio thread
>>> import trio
>>> from trio.lowlevel import current_trio_token
>>> trio_token = None
>>> async def my_func():
...     global trio_token
...     trio_token = current_trio_token()
...     with trio_thread_context():
...         await trio.sleep(10000)
>>> trio.run(my_func)
>>> ...
>>> # now back in the kivy thread
>>> with kivy_thread_context(trio_token):
...     MyApp().run()
>>> ...