Event dispatcher and urwid SDK

Author: Konstantinos Bairaktaris

tests/native/core/test_daemon.py

@@ -15,7 +15,7 @@ def test_daemon_starts(self, patched_tx):
                  cds_host='https://some.host')
 
         # the `interval` we will be using
-        interval = 1
+        interval = .1
 
         daemon = DaemonicThread()
 
@@ -49,7 +49,7 @@ def test_daemon_exception(self, patched_logger, patched_tx):
 
         daemon = DaemonicThread()
 
-        interval = 1
+        interval = .1
         daemon.start_daemon(interval=1)
         time.sleep(interval * 2)
         assert daemon.is_daemon_running(log_errors=False)

transifex/native/__init__.py

@@ -1,5 +1,4 @@
 from transifex.native.core import TxNative
 
-
 tx = TxNative()
 t = tx.translate

transifex/native/core.py

@@ -6,6 +6,7 @@
 from transifex.common.utils import generate_key, parse_plurals
 from transifex.native.cache import MemoryCache
 from transifex.native.cds import CDSHandler
+from transifex.native.events import EventDispatcher
 from transifex.native.rendering import (SourceStringErrorPolicy,
                                         SourceStringPolicy, StringRenderer)
 
@@ -20,6 +21,7 @@ def __init__(self, **kwargs):
         self.hardcoded_language_codes = None
         self.remote_languages = None
 
+        self._event_dispatcher = EventDispatcher()
         self._missing_policy = SourceStringPolicy()
         self._cds_handler = CDSHandler()
         self._cache = MemoryCache()
@@ -62,10 +64,19 @@ def setup(self,
 
         if current_language is not None:
             self.set_current_language(current_language)
+        elif source_language is not None:
+            self.set_current_language(source_language)
 
     def fetch_languages(self, force=False):
         if self.remote_languages is None or force:
-            self.remote_languages = self._cds_handler.fetch_languages()
+            self._event_dispatcher.trigger('FETCHING_LOCALES')
+            try:
+                self.remote_languages = self._cds_handler.fetch_languages()
+            except Exception:
+                self._event_dispatcher.trigger('LOCALES_FETCH_FAILED')
+                raise
+            else:
+                self._event_dispatcher.trigger('LOCALES_FETCHED')
 
         if self.hardcoded_language_codes is not None:
             return [language
@@ -81,7 +92,9 @@ def set_current_language(self, language_code, force=False):
                              format(language_code))
         if language_code not in self._cache or force:
             self.fetch_translations(language_code=language_code, force=True)
+        prev = self.current_language_code
         self.current_language_code = language_code
+        self._event_dispatcher.trigger('LOCALE_CHANGED', prev, language_code)
 
     def fetch_translations(self, language_code=None, force=False):
         """Fetch fresh content from the CDS."""
@@ -95,10 +108,20 @@ def fetch_translations(self, language_code=None, force=False):
                     "Language {} is not supported by the application".
                     format(language_code)
                 )
-            if language_code not in self._cache or force:
-                translations = self._cds_handler.\
-                    fetch_translations(language_code)
-                self._cache.update(translations)
+            self._event_dispatcher.trigger('FETCHING_TRANSLATIONS',
+                                           language_code)
+            try:
+                if language_code not in self._cache or force:
+                    translations = self._cds_handler.\
+                        fetch_translations(language_code)
+                    self._cache.update(translations)
+            except Exception:
+                self._event_dispatcher.trigger('TRANSLATIONS_FETCH_FAILED',
+                                               language_code)
+                raise
+            else:
+                self._event_dispatcher.trigger('TRANSLATIONS_FETCHED',
+                                               language_code)
 
     def translate(self, source_string, language_code=None, _context=None,
                   escape=True, params=None):
@@ -191,3 +214,10 @@ def push_source_strings(self, strings, purge=False):
         """
         response = self._cds_handler.push_source_strings(strings, purge)
         return response.status_code, json.loads(response.content)
+
+    # Events
+    def on(self, label, callback):
+        self._event_dispatcher.on(label, callback)
+
+    def off(self, label, callback):
+        self._event_dispatcher.off(label, callback)

transifex/native/events.py

@@ -0,0 +1,25 @@
+class EventDispatcher(object):
+    LABELS = ['FETCHING_TRANSLATIONS', 'TRANSLATIONS_FETCHED',
+              'TRANSLATIONS_FETCH_FAILED', 'LOCALE_CHANGED',
+              'FETCHING_LOCALES', 'LOCALES_FETCHED', 'LOCALES_FETCH_FAILED']
+
+    def __init__(self):
+        self.callbacks = {}
+
+    def on(self, label, callback):
+        self._require_label(label)
+        self.callbacks.setdefault(label, set()).add(callback)
+
+    def off(self, label, callback):
+        self._require_label(label)
+        # Can raise KeyError if callback is not there
+        self.callbacks.get(label, set()).remove(callback)
+
+    def trigger(self, label, *args, **kwargs):
+        self._require_label(label)
+        for callback in self.callbacks.get(label, []):
+            callback(*args, **kwargs)
+
+    def _require_label(self, label):
+        if label not in self.LABELS:
+            raise ValueError("Label '{}' is not supported".format(label))

transifex/native/urwid/__init__.py

@@ -0,0 +1,119 @@
+""" Utilities for integrating urwid applications with Transifex Native. """
+
+import urwid
+from transifex.native import t, tx
+
+
+class Variable(object):
+    """ Holds a value and will trigger events on change.
+
+            >>> v = Variable(1)
+            >>> v.on_change(lambda: print("New value: " + v.get()))
+            >>> v.set(v.get() + 1)
+            <<< # New value: 2
+    """
+
+    def __init__(self, value):
+        self._value = value
+        self._callbacks = set()
+
+    def get(self):
+        return self._value
+
+    def set(self, value):
+        if value != self._value:
+            self._value = value
+            for callback in self._callbacks:
+                callback(value)
+
+    def on_change(self, callback):
+        self._callbacks.add(callback)
+
+    def off_change(self, callback):
+        self._callbacks.remove(callback)
+
+
+class T(urwid.Text):
+    """ Usage:
+
+        Render the string in the current language. Will rerender on language
+        change:
+
+            >>> T("Hello world")
+
+        Render using the variable as template parameter, will rerender on
+        language change and if the parameter changes value:
+
+            >>> variable = Variable("Bob")
+            >>> T("Hello {username}", {'username': variable})
+            >>> variable.set("Jill")
+
+        Render inside an untranslatable wrapper template:
+
+            >>> T("Hello world", wrapper="Translation: {}")
+    """
+
+    def __init__(self, source_string, params=None, wrapper=None, _context=None,
+                 _charlimit=None, _comment=None, _occurrences=None, _tags=None,
+                 *args, **kwargs):
+        if params is None:
+            params = {}
+
+        self._source_string = source_string
+        self._params = params
+        self._wrapper = wrapper
+
+        tx.on("LOCALE_CHANGED", self.rerender)
+        for key, value in self._params.items():
+            try:
+                value.on_change(self.rerender)
+            except AttributeError:
+                pass
+
+        super().__init__("", *args, **kwargs)
+        self.rerender()
+
+    def rerender(self, *args, **kwargs):
+        params = {}
+        for key, value in self._params.items():
+            try:
+                params[key] = value.get()
+            except AttributeError:
+                params[key] = value
+
+        translation = t(self._source_string, params=params)
+
+        if self._wrapper is not None:
+            self.set_text(self._wrapper.format(translation))
+        else:
+            self.set_text(translation)
+
+
+def language_picker(source_language=None):
+    """ Returns an array of radio buttons for language selection.
+
+        The 'source_language' must be a dictionary describing the source
+        language, with at least the 'name' and 'code' fields. If unset,
+        `{'name': "English", 'code': "en"}` will be used
+    """
+
+    if source_language is None:
+        source_language = {'name': "English", 'code': "en"}
+
+    languages = tx.fetch_languages()
+    if not any((language['code'] == source_language['code']
+                for language in languages)):
+        languages = [source_language] + languages
+
+    language_group, language_radio = [], []
+    for language in languages:
+        button = urwid.RadioButton(language_group, language['name'])
+        urwid.connect_signal(button, 'change', _on_language_select,
+                             language['code'])
+        language_radio.append(button)
+    return language_radio
+
+
+def _on_language_select(radio_button, new_state, language_code):
+    if new_state:
+        tx.set_current_language(language_code)