thread safe PolygonClient

polygon_client.py

class PolygonClient(RESTClient):
    ''' Rate Limited RESTClient with factory method '''

    WAIT_SECONDS = 12
    _class_lock = threading.Lock()

    @classmethod
    def create(cls, *args, **kwargs) -> RESTClient:
        if 'api_key' not in kwargs:
            kwargs['api_key'] = os.environ.get("POLYGON_API_KEY")
        
        if 'paid' not in kwargs:
            paid = os.getenv("POLYGON_IS_PAID_SUBSCRIPTION", "false").lower() in {'true', '1', 't', 'y', 'yes'}
        else:
            paid = kwargs.pop('paid')
        
        if paid:
            return RESTClient(*args, **kwargs)
        else:
            return cls(*args, **kwargs)

    def _get(self, *args, **kwargs):
        with PolygonClient._class_lock:  # We use the class lock, to avoid multiple clients in the same thread to acquire it's own lock.
            logging.info(
                f"\nSleeping {PolygonClient.WAIT_SECONDS} seconds while getting data from Polygon "
                "to avoid hitting the rate limit; "
                "consider a paid Polygon subscription for faster results.\n"
            )
            time.sleep(PolygonClient.WAIT_SECONDS)
            return super()._get(*args, **kwargs)