Use aiohttp instead of requests in feed plugin

Author: JulienPalard

CONTRIBUTING.rst

@@ -7,6 +7,10 @@ First, if you want to add a cool plugin, consider submit a pull request to the
 
 Feel free to clone the project on `GitHub <https://github.com/gawel/irc3>`_.
 
+To test your change you can run irc3 in debug mode using::
+
+    $ irc3 --debug path-to-your-conf.ini
+
 Once you made a change, try to add a test for your feature/fix. At least assume
 that you have'nt broke anything by running tox::
 
@@ -40,9 +44,7 @@ And check the result::
 
     $ firefox .tox/docs/tmp/html/index.html
 
-The project is `buildout <https://github.com/buildout/buildout>`_ ready. You
-can generate binaries using it instead of virtualenv::
+The project uses ``setuptools``, you can test-install it using `pip`:
 
-    $ python bootstrap.py
-    $ bin/buildout
-    $ bin/irc3 -h
+    $ pip install .
+    $ irc3 -h

irc3/compat.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 import sys
-import types
 
 PY34 = bool(sys.version_info[0:2] >= (3, 4))
 PY35 = bool(sys.version_info[0:2] >= (3, 5))

irc3/plugins/feeds.py

@@ -4,7 +4,6 @@
 import irc3
 import datetime
 from irc3.compat import asyncio
-from concurrent.futures import ThreadPoolExecutor
 
 __doc__ = '''
 ==========================================
@@ -22,9 +21,9 @@
         irc3.plugins.feeds
 
     [irc3.plugins.feeds]
-    channels = #irc3                             # global channel to notify
-    delay = 5                                    # delay to check feeds
-    directory = ~/.irc3/feeds                    # directory to store feeds
+    channels = #irc3                          # global channel to notify
+    delay = 5                                 # delay to check feeds in minutes
+    directory = ~/.irc3/feeds                 # directory to store feeds
     hook = irc3.plugins.feeds.default_hook       # dotted name to a callable
     fmt = [{name}] {entry.title} - {entry.link}  # formatter
 
@@ -34,7 +33,7 @@
     github/irc3.fmt = [{feed.name}] New commit: {entry.title} - {entry.link}
     # custom channels
     github/irc3.channels = #irc3dev #irc3
-    # custom delay
+    # custom delay in minutes
     github/irc3.delay = 10
 
 Hook is a dotted name refering to a callable (function or class) wich take a
@@ -63,6 +62,12 @@
 
 '''
 
+HEADERS = {
+    'User-Agent': 'python-aiohttp/irc3/feeds',
+    'Cache-Control': 'max-age=0',
+    'Pragma': 'no-cache',
+}
+
 
 def default_hook(entries):
     """Default hook called for each entry"""
@@ -76,21 +81,6 @@ def dispatcher(messages):
     return dispatcher
 
 
-def fetch(args):
-    """fetch a feed"""
-    session = args['session']
-    for feed, filename in zip(args['feeds'], args['filenames']):
-        try:
-            resp = session.get(feed, timeout=5)
-            content = resp.content
-        except Exception:  # pragma: no cover
-            pass
-        else:
-            with open(filename, 'wb') as fd:
-                fd.write(content)
-    return args['name']
-
-
 ISO_FORMAT = "%Y-%m-%dT%H:%M:%S"
 
 
@@ -102,7 +92,7 @@ def parse(feedparser, args):
 
     for filename in args['filenames']:
         try:
-            with open(filename + '.updated') as fd:
+            with open(filename + '.updated', encoding="UTF-8") as fd:
                 updated = datetime.datetime.strptime(
                     fd.read()[:len("YYYY-MM-DDTHH:MM:SS")], ISO_FORMAT
                 )
@@ -140,14 +130,6 @@ def parse(feedparser, args):
 class Feeds:
     """Feeds plugin"""
 
-    PoolExecutor = ThreadPoolExecutor
-
-    headers = {
-        'User-Agent': 'python-requests/irc3/feeds',
-        'Cache-Control': 'max-age=0',
-        'Pragma': 'no-cache',
-    }
-
     def __init__(self, bot):
         bot.feeds = self
         self.bot = bot
@@ -177,7 +159,6 @@ def __init__(self, bot):
             fmt=config.get('fmt', '[{feed.name}] {entry.title} {entry.link}'),
             delay=delay,
             channels=config.get('channels', ''),
-            headers=self.headers,
             time=0,
         )
 
@@ -208,7 +189,16 @@ def __init__(self, bot):
 
     def connection_made(self):
         """Initialize checkings"""
-        self.bot.loop.call_later(10, self.update)
+        self.bot.create_task(self.periodically_update())
+
+    async def periodically_update(self):
+        """After a connection has been made, call update feeds periodically."""
+        if not self.aiohttp or not self.feedparser:
+            return
+        await asyncio.sleep(10)
+        while True:
+            await self.update()
+            await asyncio.sleep(self.delay)
 
     def imports(self):
         """show some warnings if needed"""
@@ -219,15 +209,14 @@ def imports(self):
             self.bot.log.critical('feedparser is not installed')
             self.feedparser = None
         try:
-            import requests
+            import aiohttp
         except ImportError:  # pragma: no cover
-            self.bot.log.critical('requests is not installed')
-            self.session = None
+            self.bot.log.critical('aiohttp is not installed')
+            self.aiohttp = None
         else:
-            self.session = requests.Session()
-            self.session.headers.update(self.headers)
+            self.aiohttp = aiohttp
 
-    def parse(self, *args):
+    def parse(self):
         """parse pre-fetched feeds and notify new entries"""
         entries = []
         for feed in self.feeds.values():
@@ -239,34 +228,37 @@ def messages():
                 if entry:
                     feed = entry.feed
                     message = feed['fmt'].format(feed=feed, entry=entry)
-                    for c in feed['channels']:
-                        yield c, message
+                    for channel in feed['channels']:
+                        yield channel, message
 
         self.dispatcher(messages())
 
-    def update_time(self, future):
-        name = future.result()
-        self.bot.log.debug('Feed %s fetched', name)
-        feed = self.feeds[name]
-        feed['time'] = time.time()
-
-    def update(self):
+    async def update(self):
         """update feeds"""
-        loop = self.bot.loop
-        loop.call_later(self.delay, self.update)
-
         now = time.time()
-        session = self.session
-        feeds = [dict(f, session=session) for f in self.feeds.values()
-                 if f['time'] < now - f['delay']]
-        if feeds:
-            self.bot.log.info('Fetching feeds %s',
-                              ', '.join([f['name'] for f in feeds]))
-            tasks = []
-            for feed in feeds:
-                task = loop.run_in_executor(None, fetch, feed)
-                task.add_done_callback(self.update_time)
-                tasks.append(task)
-            task = self.bot.create_task(
-                asyncio.wait(tasks, timeout=len(feeds) * 2, loop=loop))
-            task.add_done_callback(self.parse)
+        feeds = [feed for feed in self.feeds.values()
+                 if feed['time'] < now - feed['delay']]
+        if not feeds:
+            return
+        self.bot.log.info('Fetching feeds %s',
+                          ', '.join([f['name'] for f in feeds]))
+        timeout = self.aiohttp.ClientTimeout(total=5)
+        async with self.aiohttp.ClientSession(timeout=timeout) as session:
+            await asyncio.gather(
+                *[self.fetch(feed, session) for feed in feeds]
+            )
+        self.parse()
+
+    async def fetch(self, feed, session):
+        """fetch a feed"""
+        for url, filename in zip(feed['feeds'], feed['filenames']):
+            try:
+                async with session.get(url, headers=HEADERS) as resp:
+                    with open(filename, 'wb') as file:
+                        file.write(await resp.read())
+            except Exception:  # pragma: no cover
+                self.bot.log.exception(
+                    "Exception while fetching feed %s", feed['name']
+                )
+        self.bot.log.debug('Feed %s fetched', feed['name'])
+        feed['time'] = time.time()