# Python: Asyncio object-oriented style

I'm seeking a review of some code I've written for a small personal project. The project is not yet mature enough for any domain specific details to be of concern - but I'm happy to provide further details/examples should this help the review.

A description of what I hope to achieve and minimal working example has been provided below to help guide the review.

Comments relating to my use of the Asyncio library would be very much appreciated, as this is an area that is quite new to me, and I was not able to find any examples implemented in an OOP style.

Concise problem statement

In a sentence, the Limiter class code should provide an extensible way to restrict the rate at which certain functions can be invoked.

For this example, I have tried to limit the invocation of increment() to 5 calls every 10 seconds.

It is important that this limit is imposed asynchronously so that other computations can be performed in the background. For this example, simply display the count.

I have tried to maintain an OOP style interface.

Minimal working example

Limiter

class Limiter(object):

"""

The purpose of this class is to limit the invocation of increment()
based on a evaluation of 'some_condition' wrt. the class's internal state.

"""

def __init__(self: object) -> None:
self.counter, self.limit = 0, 5

self.loop = asyncio.get_event_loop()

""" Minimal selection of tasks for this demonstation. """

def _increment(self: object) -> None:
""" Increment counter variable. """

print('Incrementing counter')
self.counter += 1

async def reset_counter(self: object) -> None:
""" Periodically (10s) reset the counter value. """

while True:
print('Resetting Counter')
self.counter = 0
await asyncio.sleep(10)

async def display(self: object) -> None:
""" Periodically (1s) display the counter value. """

while True:
print(self.counter)
await asyncio.sleep(1)

async def increment(self: object) -> bool:
""" Provided that some_condition is met, call increment(). """

some_condition = (self.counter < 5)

if (some_condition):
self._increment()
return True

print('some_condition not met - Not able to call increment().')
return False


Caller

class Caller(object):

"""

An object interfacing with the limiter.

Periodicalyl call limiter.increment()
Calls to increment() should be restricted by the limiter based
on the evaluation of 'some_condition' wrt the limiters internal state.

"""

def __init__(self: object) -> None:
self.limiter = Limiter()

self.loop = asyncio.get_event_loop()

""" Minimal selection of tasks for this demonstation. """

async def make_call(self: object) -> None:
""" Periodically (1s) call increment. """

while True:
await self.limiter.increment()
await asyncio.sleep(1)


main.py

def main():

try:
loop = asyncio.get_event_loop()
caller = Caller()
loop.run_forever()

except KeyboardInterrupt:
print('Stopping event loop')
loop.stop()

if __name__ == "__main__":
main()


Example console output

Hodgson:Limiter hodgson\$ python main.py
Resetting Counter
Counter: 0
Incrementing counter
Counter: 1
Incrementing counter
Counter: 2
Incrementing counter
Counter: 3
Incrementing counter
Counter: 4
Incrementing counter
Counter: 5
some_condition not met - Not able to call increment().
Counter: 5
some_condition not met - Not able to call increment().
Counter: 5
some_condition not met - Not able to call increment().
Counter: 5
some_condition not met - Not able to call increment().
Counter: 5
some_condition not met - Not able to call increment().
Resetting Counter
Counter: 0
Incrementing counter
Counter: 1
Incrementing counter
Counter: 2
Incrementing counter
Counter: 3
Incrementing counter
^CStopping event loop


The code can also be found here: Repl.it

I am using Python v3.7.3

# Old style declaration

class Limiter(object):
...


Using object as a base class is no longer necessary. Simply write:

class Limiter:
...


Same for the other classes

    def __init__(self: object) -> None:
...


self is not any object; it must be a Limiter (or class derived from Limiter). The proper type-hint to use would be Limiter, but that type is still being defined so you can’t use it directly. The type-hint ”Limiter” would be possible to use. However, any type-checker knows that the self argument must be of the type of the class. It should be left implicit:

    def __init__(self) -> None:
...


Same for the other methods.

# Synchronous Async method

    async def increment(self: object) -> bool:
...


This method doesn’t await any results. It does not need to be async.

    def increment(self) -> bool:
...


# PEP-8

    if (some_condition):
...


Conditionals do not need to be enclosed in (...)’s.

Methods which are for internal usage only should be marked as “private” (leading underscore). These include schedule_tasks, reset_counter, display, and make_call.

Similarly, non-public data members should have an leading underscore: limiter, loop, limit and count.