Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Akamai PAPI (Property Manager API) Client
"""Akamai PAPI (Property Manager API) Client
This module defines a client class from the Akamai Property Manager API.
"""
import logging
import re
from urllib.parse import urljoin
import requests
from akamai.edgegrid import EdgeGridAuth
VERSION_NUMBER_REGEX = r"\/papi\/v0\/properties\/prp_\d+\/versions\/(?P<version>\d+)"
LOGGER = logging.getLogger(__name__)
def _get_version_number_from_link(version_link):
regex = re.compile(VERSION_NUMBER_REGEX)
match = regex.search(version_link)
version = int(match.group("version"))
LOGGER.info(f"Returned URL: {version_link} and parsed version: {version}")
return version
class AkamaiClient:
""" Defines a client that calls the Akamai API for certain property manager features. """
def __init__(self, base_url, client_token, client_secret, access_token):
self.base_url = base_url
self.session = requests.Session()
self.session.auth = EdgeGridAuth(
client_token=client_token,
client_secret=client_secret,
access_token=access_token,
max_body=128 * 1024,
)
LOGGER.info(f"Initialized Akamai Client for API: {base_url}")
def _get_api_with_path(self, path):
url = urljoin(self.base_url, path)
LOGGER.info(f"Making Akamai Request for URL: {url}")
return self.session.get(url).json()
def _post_api_with_path(self, path, body):
url = urljoin(self.base_url, path)
LOGGER.info(f"Making Akamai Request for URL: {url}")
return self.session.post(url, json=body).json()
def _put_api_with_path(self, path, body):
url = urljoin(self.base_url, path)
LOGGER.info(f"Making Akamai Request for URL: {url}")
return self.session.put(url, json=body).json()
def get_current_config_version(self, property_id) -> int:
""" Gets the current configuration version that is made for a given property.
Args:
property_id: The property id from akamai.
Returns:
The numerical identifier for the current config version on that property.
"""
LOGGER.info(f"Getting Current Config Version for: {property_id}")
api_path = f"/papi/v1/properties/{property_id}/versions/latest"
return self._get_api_with_path(api_path)["versions"]["items"][0][
"propertyVersion"
]
def create_new_config_version(self, property_id, based_on) -> int:
""" Creates a new configuration version based on a specified version.
This config version is not activated in any environment. It is simply for
creating a new akamai configuration.
Args:
property_id: The property id from akamai.
based_on: The numeric id of the version to say the current config is based on.
Returns:
The numerical identifier for the new config version on that property.
"""
LOGGER.info(
f"Creating New Config Version for: {property_id} based on version: {based_on}"
)
api_path = f"/papi/v1/properties/{property_id}/versions"
version_link = self._post_api_with_path(
api_path, body={"createFromVersion": based_on}
)["versionLink"]
return _get_version_number_from_link(version_link)
def update_config_rules(self, property_id, config_version, rules):
""" Changes the ruleset for a specific config version of a property with new rules.
This config version is not activated in any environment. It is simply for
creating a new akamai configuration.
Args:
property_id: The property id from akamai.
config_version: The numeric id of the version to update.
rules: A dict definting the json rules to update to.
Returns:
A dictionary retaining information about the version of the config rules.
"""
LOGGER.info(f"Updating Config Rules for {property_id} version {config_version}")
api_path = f"/papi/v1/properties/{property_id}/versions/{config_version}/rules"
return self._put_api_with_path(api_path, body=rules)
def activate_config_version(
self, property_id, config_version, network_name, revision, activation_emails=[]
) -> str:
""" Activates a specifc config version on an akamai network for a given property.
This config version is not activated in any environment. It is simply for
creating a new akamai configuration.
Args:
property_id: The property id from akamai.
config_version: The numeric id of the version to update.
network_name: "STAGING" or "PRODUCTION"; the name of the network to activate on.
activation_emails(optional): A list of email addresses to activate on.
revision: Some message to add to the note about the activation.
Returns:
A dictionary retaining information about the version of the config rules.
"""
LOGGER.info(f"Activating Config for {property_id} version {config_version}")
api_path = f"/papi/v1/properties/{property_id}/activations"
return self._post_api_with_path(
api_path,
body={
"propertyVersion": config_version,
"network": network_name,
"note": f"Activated through akamai config deployer cloudformation. Revision: {revision}",
"useFastFallback": network_name == "PRODUCTION",
"notifyEmails": activation_emails,
"acknowledgeAllWarnings": True,
},
)["activationLink"]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.