Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
import asyncio
import functools
import json
import secrets
import aiohttp
from concurrent.futures import ALL_COMPLETED
class FeedUpdater:
def __init__(self, feeds, loop):
self.feeds = feeds
self.loop = loop
self.session = aiohttp.ClientSession(loop=loop)
async def _fetch(self, url):
async with self.session.get(url) as response:
status = response.status
assert status == 200
data = await response.text()
return url, data
async def __call__(self):
tasks = [self._fetch(url) for url in self.feeds.keys()]
done, pending = await asyncio.wait(
for task in done:
url, data = task.result()
self.feeds[url] = f"{len(data)}.{secrets.token_hex(4)}"
# TODO placeholder
print(json.dumps(self.feeds, sort_keys=True, indent=4))
def __del__(self):
def schedule_func(func, args=None, kwargs=None, interval=60, *, loop):
if args is None:
args = []
if kwargs is None:
kwargs = {}
async def periodic_func():
while True:
await func(*args, **kwargs)
await asyncio.sleep(interval, loop=loop)
return loop.create_task(periodic_func())
create_scheduler = lambda loop: functools.partial(schedule_func, loop=loop)
# USAGE =====================================================
loop = asyncio.new_event_loop()
schedule = create_scheduler(loop=loop)
feeds = {
"": None,
"": None
update = FeedUpdater(feeds=feeds, loop=loop)
refresh_task = schedule(update, interval=3)
Copy link

hbasria commented Jun 10, 2018

Line 53 should be like this

create_scheduler = lambda loop: functools.partial(schedule, loop=loop)

Copy link

numberoverzero commented Jan 9, 2019

@hbasria you are correct! Thanks for the note, I also renamed the function slightly to remove the duplication in the usage section.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment