Tasks¶
The main advantage of an asynchronous client is that it will be able to run multiple tasks… asynchronously. Which is quite interesting here if you want to access several Streaming APIs, or perform some requests periodically while using a Streaming API.
So I tried to make it easier to create such a program.
Init tasks¶
By default the PeonyClient
makes 2 requests after being
started:
- account/verify_credentials.json (kept as
self.user
) - help/twitter_configuration.json (kept as
self.twitter_configuration
)
If you need to have more informations during the initialization of a client you
should use the init_task()
decorator.
The methods decorated with this decorator will be started on setup.
from peony import PeonyClient, init_tasks
class Client(PeonyClient):
@init_tasks
async def get_setting():
self.settings = await self.api.account.settings.get()
@init_tasks
async def get_likes():
self.likes = await self.api.favorites.list.get(count=200)
Note
The attributes user and twitter_configuration are created by the tasks in PeonyClient.init_tasks() which are the respectively the responses from /1.1/account/verify_credentials.json and /1.1/help/configuration.json. So you can access self.user.id in the class and this will give you the id of the authenticated user.
Note
The attribute twitter_configuration
is used by the method
upload_media when it converts your picture
The task
decorator¶
First you will need to create a subclass of PeonyClient and add a
task()
decorator to the methods that you want to run.
import asyncio
import time
from peony import PeonyClient, task
class AwesomePeonyClient(PeonyClient):
@staticmethod
async def wait_awesome_hour():
""" wait until the next awesome hour """
await asyncio.sleep(-time.time() % 3600)
async def send_awesome_tweet(self, status="Peony is awesome!!"):
""" send an awesome tweet """
await self.api.statuses.update.post(status=status)
@task
async def awesome_loop(self):
""" send an awesome tweet every hour """
while True:
await self.wait_awesome_hour()
await self.send_awesome_tweet()
@task
async def awesome_user(self):
""" The user using this program must be just as awesome, right? """
user = await self.api.account.verify_credentials.get()
print("This is an awesome user", user.screen_name)
def main():
""" start all the tasks """
loop = asyncio.get_event_loop()
# set your api keys here
awesome_client = AwesomePeonyClient(
consumer_key=your_consumer_key,
consumer_secret=your_consumer_secret,
access_token=your_access_token,
access_token_secret=your_access_token_secret,
loop=loop
)
awesome_client.run() # you can also use the run_tasks()
# coroutine if you need it
if __name__ == '__main__':
main()
Note
The run_tasks()
method can be used
instead of run()
to start the tasks.
Just keep in mind that run()
is a
wrapper around run_tasks()
with some
basic features such as handling KeyboardInterrupt
and run
close()
when all the tasks are complete.