Skip to content

Instantly share code, notes, and snippets.

@jasonforte

jasonforte/README.md

Created July 31, 2024 11:34
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save jasonforte/b7697c12c670a5e8e90eeda3468ecd58 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save jasonforte/b7697c12c670a5e8e90eeda3468ecd58 to your computer and use it in GitHub Desktop.
Download ZIP
Dynamic Configuration Example for AWS AppConfig in Python
Raw
README.md

Dynamic Configuartion using AWS AppConfig

This gist provides a Python class that allows retrieval of dynamic configuration from AWS AppConfig.

Pre-Requisites

This assumes a Freeform, JSON configuration is available in AWS AppConfig as follows:

{
  "featureA": false,
  "temperature": 0.7,
  "bedrockModelId": "anthropic.claude-v2:1"
}

Usage

Instantiate the class in one of the following ways depending on the behavior you want

# Access AWS AppConfig directly via the APIs
dynamic_config = DynamicConfig('api', 'prod-01', 'default', fetch_type='direct')

# Access AWS AppConfig via the AWS AppConfig Agent deployed at http://localhost:2772
dynamic_config = DynamicConfig('api', 'prod-01', 'default', fetch_type='agent', base_url='http://localhost:2772')

Here api is the application, prod-01 is the environment & default is the profile for the application configuartion in AWS AppConfig.

Once the class is instantiated, use the configuration like you would a dict.

feature_a = dynamic_config['featureA'] # returns False
temperature = dynamic_config['temperature'] # returns 0.7
bedrock_model = dynamic_config['bedrockModelId'] # returns 'anthropic.claude-v2:1'

Refreshing the config is handled by a basic cache mechanism in the class or via the agent if using the fetch_type=agent

About the Authors

This is provided by MakeOps. Feel free to use it as-is.

If you need help building an MVP or scaling using AWS contact us here.

Raw
config.py
import os
import time
import json
import logging
from typing import Any
import boto3
import requests
os.environ['AWS_DEFAULT_PROFILE'] = 'jason'
os.environ['AWS_DEFAULT_REGION'] = 'eu-west-1'
LOG_LEVEL = os.environ.get('LOG_LEVEL', 'info')
logging.basicConfig(
level=logging.INFO,
format='%(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
logger.setLevel(LOG_LEVEL.upper())
class ConfigFetchBase:
"""
Base class for configuration fetching with refresh functionality.
"""
def __init__(self, application, environment, profile, refresh_interval: int = 60) -> None:
"""
Initialize the ConfigFetchBase instance.
:param application: The application name
:param environment: The environment (e.g., dev, prod)
:param profile: The configuration profile
:param refresh_interval: Time in seconds between config refreshes (default: 60)
"""
self._application = application
self._environment = environment
self._profile = profile
self._refresh_refresh_interval = refresh_interval
self._next_poll_interval_seconds = refresh_interval
self._last_fetch_time = -1 # Initialize to -1 to ensure first fetch
self._last_cached_response = {} # Cache for the last fetched configuration
def can_refresh(self):
"""
Check if it's time to refresh the configuration.
:return: True if refresh is needed, False otherwise
"""
if self._last_fetch_time < 0:
return True # Always refresh if it's the first time
current_time = int(time.time())
return (current_time - self._last_fetch_time) > self._next_poll_interval_seconds
def mark_last_fetch_time(self, last_fetch_time: int = -1):
"""
Update the last fetch time.
:param last_fetch_time: The timestamp to set (default: -1, which uses current time)
"""
self._last_fetch_time = self._last_fetch_time if last_fetch_time != -1 else int(time.time())
def fetch_config(self):
"""
Fetch the configuration. This method should be implemented by subclasses.
:raises NotImplementedError: If not implemented in a subclass
"""
raise NotImplementedError(f'Must implement fetch_config for class {self.__class__.__name__}')
class ConfigFetchDirect(ConfigFetchBase):
"""
Class for fetching configuration directly from AWS AppConfig.
"""
def __init__(self, application, environment, profile, refresh_interval: int = 60) -> None:
"""
Initialize the ConfigFetchDirect instance.
:param application: The application name
:param environment: The environment (e.g., dev, prod)
:param profile: The configuration profile
:param refresh_interval: Time in seconds between config refreshes (default: 60)
"""
super().__init__(application, environment, profile, refresh_interval)
self._client = boto3.client('appconfigdata')
# Start a configuration session with AWS AppConfig
resp = self._client.start_configuration_session(
ApplicationIdentifier=application,
EnvironmentIdentifier=environment,
ConfigurationProfileIdentifier=profile,
RequiredMinimumPollIntervalInSeconds=refresh_interval
)
self._configuration_token = resp['InitialConfigurationToken']
def fetch_config(self) -> dict:
"""
Fetch the latest configuration from AWS AppConfig.
:return: The fetched configuration as a dictionary
"""
if not self.can_refresh():
return self._last_cached_response
resp = self._client.get_latest_configuration(
ConfigurationToken=self._configuration_token
)
self.mark_last_fetch_time()
# Update token and poll interval for next fetch
self._configuration_token = resp['NextPollConfigurationToken']
self._next_poll_interval_seconds = resp['NextPollIntervalInSeconds']
# Parse and cache the configuration
configuration = json.loads(resp['Configuration'].read())
self._last_cached_response = configuration
return configuration
class ConfigFetchAgent(ConfigFetchBase):
"""
Class for fetching configuration from a local AppConfig Agent.
"""
def __init__(self, application, environment, profile, refresh_interval: int = 60, base_url: str= '') -> None:
"""
Initialize the ConfigFetchAgent instance.
:param application: The application name
:param environment: The environment (e.g., dev, prod)
:param profile: The configuration profile
:param refresh_interval: Time in seconds between config refreshes (default: 60)
:param base_url: The base URL of the AppConfig Agent (default: 'http://localhost:2772')
"""
self._base_url = base_url if base_url != '' else 'http://localhost:2772'
super().__init__(application, environment, profile, refresh_interval)
def fetch_config(self) -> dict:
"""
Fetch the latest configuration from the AppConfig Agent.
:return: The fetched configuration as a dictionary
"""
if not self.can_refresh():
return self._last_cached_response
# Make a GET request to the AppConfig Agent
resp = requests.get(f'{self._base_url}/applications/{self._application}/environments/{self._environment}/configurations/{self._profile}')
self.mark_last_fetch_time()
# Parse and cache the configuration
configuration = resp.json()
self._last_cached_response = configuration
return configuration
class DynamicConfig(dict):
"""
A dictionary-like class that automatically updates its configuration.
"""
def __init__(self, application, environment, profile, fetch_type='direct', refresh_interval: int = 60, **kwargs):
"""
Initialize the DynamicConfig instance.
:param application: The application name
:param environment: The environment (e.g., dev, prod)
:param profile: The configuration profile
:param fetch_type: The type of fetcher to use ('direct' or 'agent', default: 'direct')
:param refresh_interval: Time in seconds between config refreshes (default: 60)
"""
self._type = fetch_type
self._config = {}
# Choose the appropriate fetcher based on fetch_type
if fetch_type == 'agent':
self._fetcher = ConfigFetchAgent(application, environment, profile, refresh_interval=refresh_interval, **kwargs)
else:
self._fetcher = ConfigFetchDirect(application, environment, profile, refresh_interval=refresh_interval, **kwargs)
self._config = self._fetcher.fetch_config()
def __getitem__(self, key: Any) -> Any:
"""
Get an item from the configuration, refreshing it first.
:param key: The key to look up in the configuration
:return: The value associated with the key
"""
self._config = self._fetcher.fetch_config()
return self._config[key]
def __repr__(self) -> str:
"""
Return a string representation of the configuration.
:return: A string representation of the configuration
"""
return self._config.__repr__()
def __str__(self) -> str:
"""
Return a string version of the configuration.
:return: A string version of the configuration
"""
return self._config.__str__()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment