aiotest.py

aiotest.py

import aiohttp, asyncio, async_timeout
import ui

class AsyncUIView(ui.View):
  def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.running = True
    self._loop = asyncio.get_event_loop()
    self._session = aiohttp.ClientSession(loop=self._loop)
    
  def start_loop(self):
    try:
      self._loop.run_until_complete(self._runner())
    finally:
      self._loop.close()
      asyncio.set_event_loop(asyncio.new_event_loop()) # Clean Pythonista asyncio loop environment
    
  def run(self, coro):
    self._loop.create_task(coro)
  
  def will_close(self):
    self.running = False
    if not self._session.closed:
      self._session.close()
    
  async def _runner(self):
    while self.running:
      await self._loop.create_task(asyncio.sleep(0.5))
      
  async def get(self, url):
    with async_timeout.timeout(10):
      async with self._session.get(url) as response:
        return await response.text()
  
if __name__ == '__main__':
  
  class DemoView(ui.View):
    def __init__(self, async_view, *args, **kwargs):
      super().__init__(*args, **kwargs)
      self.aio = async_view
      self.web1 = ui.WebView(flex='WH', frame=(0,0,self.width,self.height/2))
      self.add_subview(self.web1)
      self.web2 = ui.WebView(flex='WH', frame=(0,self.height/2,self.width,self.height/2), background_color='lightgrey')
      self.add_subview(self.web2)
      self.counter = 0
      self.btn = btn = ui.Button(flex='WH', frame=self.bounds, tint_color=(.77, .22, .03), action=self.do_fetch)
      self.add_subview(btn)
      self.update_interval = 0.5
    
    def update(self):
      self.counter += 1
      self.btn.title = f'#{self.counter} - Click me'
      
    def do_fetch(self, sender):
      loading = '<span style="font-size: xx-large;">Loading...</span>'
      self.web1.load_html(loading)
      self.aio.run(self.load_page('http://python.org', self.web1))
      self.web2.load_html(loading)
      self.aio.run(self.load_page('http://omz-software.com/pythonista/', self.web2))
      
    async def load_page(self, url, webview):
      html = await self.aio.get(url)
      webview.load_html(html)

    
  v = AsyncUIView()
  v.present('full_screen')
  
  v.add_subview(DemoView(v, frame=v.bounds))
  
  v.start_loop()