Source code for dynamodb_config_store.config_stores.time_based

""" Time based config store

This config store updates the configuration every x seconds
"""
import threading
import time

from boto.dynamodb2.exceptions import ItemNotFound

from dynamodb_config_store.config_stores import ConfigStore


[docs]class TimeBasedConfigStore(ConfigStore): """ Fetches Stores on a periodic interval from DynamoDB All internal variables and methods are private (with an leading _), as the configuration options from DynamoDB will be set as instance attributes. """ _option_key = None # Option key in DynamoDB _populated = False # True when the first population has been done _store_key = None # Store key in DynamoDB _store_name = None # Name of the Store _table = None # boto.dynamodb2.table.Table _update_interval = 300 # How often, in seconds, to fetch updates def __init__( self, table, store_name, store_key, option_key, update_interval=300): """ Constructor for the TimeBasedConfigStore :type table: boto.dynamodb2.table.Table :param table: Table instance :type store_name: str :param store_name: Name of the DynamoDB Config Store :type store_key: str :param store_key: Key name for the store in DynamoDB. Default _store :type option_key: str :param option_key: Key name for the option in DynamoDB. Default _option :type update_interval: int :param update_interval: How often, in seconds, to fetch updates :returns: None """ self._option_key = option_key self._store_key = store_key self._store_name = store_name self._table = table self._update_interval = update_interval thread = threading.Thread(target=self._run, args=()) thread.daemon = True thread.start() while not self._populated: time.sleep(0.5)
[docs] def _run(self): """ Run periodic fetcher :returns: None """ while True: # Get options from DynamoDB options = self._fetch_options() # Delete old attributes self._delete_instance_attributes() # Populate the attribute list with the new attributes self._attributes = [key for key in options.keys()] # Add new attributes self._set_instance_attributes(options) self._populated = True time.sleep(self._update_interval)
[docs] def _fetch_options(self): """ Retrieve a dictionary with all options and values from DynamoDB :returns: dict -- Dict with {'option': {'key': 'value'}} """ try: items = {} query = {'{}__eq'.format(self._store_key): self._store_name} for item in self._table.query_2(**query): option = item[self._option_key] # Remove metadata del item[self._store_key] del item[self._option_key] items[option] = {k: v for k, v in item.items()} return items except ItemNotFound: return {}