4
\$\begingroup\$

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()
        self.schedule_tasks()

    def schedule_tasks(self: object) -> None:
        """ Minimal selection of tasks for this demonstation. """

        self.loop.create_task(self.reset_counter())
        self.loop.create_task(self.display())

    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()
        self.schedule_tasks()

    def schedule_tasks(self: object) -> None:
        """ Minimal selection of tasks for this demonstation. """

        self.loop.create_task(self.make_call())

    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

\$\endgroup\$
2
\$\begingroup\$

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

Wrong receiver type-hint

    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.

| improve this answer | |
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.