""" AUTHOR: Khushal P Soonderji DATE: Friday, 6th Jun., 2025 OBJECTIVE: To have a load balancer that gives us objects in a round-robin or random fashion from a pool of objects. This way the load can be spread over all the objects evenly. Imagine a Telegram Bot sending out all the messages alone. It'll eventually get rate-limited. But if you have a pool of objects, you could use all bots in a round-robin fashion and not run into rate limits for a long time. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # For working with doubly-ended queues: from collections import deque # For randomization: import random # To work with datatypes: from typing import List, Any # For debugging: from icecream import IceCreamDebugger # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class ObjectLoadBalancer: def __init__( self, objs: List[Any] = None, debug = True, debug_prefix = "ObjLB (C) | ", debug_only_errors = True ): """ To initialize the class that holds all the objects in the pool and gives them out one by one as needed. THIS IS NOT A POOL WHERE THE OBJECT HAS TO BE RETURNED. THIS IS SIMPLY A POINTER THAT POINTS TO THE NEXT OBJECT SO THAT LOAD GETS DISTRIBUTED EVENLY. :param objs: The objects to hold in the pool. More can be added later with the 'add' method. :param debug: Whether, or not, you would like to show debugging messages on the terminal. :param debug_prefix: The prefix string to identify the debugging messages. :param debug_only_errors: Whether you would like to show all debugging messages or just error messages. """ # Prepare the debugging utility: self._debug_prefix = debug_prefix self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) if not debug: self._printer.disable() self._debug_only_errors = debug_only_errors # Add the initial set of objects to the pool: self.__pool = deque() if objs is None: pass elif isinstance(objs, list): self.__pool.extend(objs) else: self.__pool.append(objs) def add(self, obj: Any) -> None: """ To add an object to the pool. :param obj: The object to add. :return: None """ self.__pool.append(obj) self._printer("Object added to the pool.") def remove(self) -> Any | None: """ To remove one object from the pool. :return: The object that was removed. """ removed_obj = None try: removed_obj = self.__pool.popleft() self._printer("Object removed from the pool.") except ValueError: pass return removed_obj def clear(self) -> None: """ Clears ALL the objects from the pool. :return: None """ self.__pool.clear() self._printer("Pool cleared.") def get_next(self, default: Any = None) -> Any | None: """ Gets the next object from the pool. Works as a round-robin load-balancer. :param default: The default object to return if the pool is empty. :return: The next object from the pool. None if there are no objects in the pool. """ # Check if there is anything in the pool: if not self.__pool: self._printer("The pool is empty!") return default # Pick the next item from the pool, # shift the pointer to the item after that, # and return the picked item: obj = self.__pool[0] self.__pool.rotate(-1) return obj def get_random(self, default: Any = None) -> Any | None: """ Gets a random object from the pool. :param default: The default object to return if the pool is empty. :return: A random object from the pool. None if there are no objects in the pool. """ # Check if there is anything in the pool: if not self.__pool: self._printer("The pool is empty!") return default # Pick a random object from the pool, # shift the pointer to the object after that one, # return the picked item: index = random.randint(0, len(self.__pool) - 1) obj = self.__pool[index] self.__pool.rotate(-1 * (index + 1)) return obj # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": my_objs = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] my_lb = ObjectLoadBalancer(my_objs) print("RNDM:", my_lb.get_random()) print("NEXT:", my_lb.get_next()) print("NEXT:", my_lb.get_next()) print("NEXT:", my_lb.get_next()) print("\n\n---\n\n") my_lb.add(10) my_lb.add(11) my_lb.add(12) my_lb.add(13) my_lb.add(14) my_lb.add(15) for _ in range(20): print("NEXT:", my_lb.get_next()) print("\n\n---\n\n") my_lb.remove() my_lb.remove() my_lb.remove() for _ in range(20): print("NEXT:", my_lb.get_next()) print("\n\n---\n\n") my_lb.clear() print("NEXT:", my_lb.get_next()) print("NEXT:", my_lb.get_next(default = "test")) print("\n\n---\n\n")