Module `xcon.providers.ssm_param_store`

Expand source code

from __future__ import annotations

from typing import Optional, Mapping

from xboto import boto_clients

from xcon.provider import AwsProvider, ProviderChain

from .common import handle_aws_exception
from ..directory import Directory, DirectoryListing, DirectoryOrPath, DirectoryItem, DirectoryChain


class SsmParamStoreProvider(AwsProvider):
    attributes_to_skip_while_copying = ['_store_get_params_paginator']
    name = "ssm"
    _store_get_params_paginator = None

    @property
    def _get_params_paginator(self):
        paginator = self._store_get_params_paginator
        if not paginator:
            paginator = boto_clients.ssm.get_paginator('get_parameters_by_path')
            self._store_get_params_paginator = paginator
        return paginator

    def get_item(
            self,
            name: str,
            directory: Optional[DirectoryOrPath],
            directory_chain: DirectoryChain,
            provider_chain: ProviderChain,
            environ: Directory
    ) -> Optional[DirectoryItem]:
        if directory is None:
            return None
        return self._item_only_for_directory(name=name, directory=directory)

    def _item_only_for_directory(
            self, name: str, directory: DirectoryOrPath
    ) -> Optional[DirectoryItem]:
        directory = Directory.from_path(directory)
        if not directory:
            return None

        listing = self.local_cache.get(directory)
        if listing:
            return listing.get_item(name)

        items = []
        try:
            # We need to lookup the directory listing from Secrets Manager.
            pages = self._get_params_paginator.paginate(
                Path=directory.path,
                Recursive=False,
                WithDecryption=True,
            )

            for p in pages:
                for item_info in p['Parameters']:
                    item_path: str = item_info['Name']
                    item = DirectoryItem(
                        directory=directory,
                        name=item_path.split("/")[-1],
                        value=item_info['Value'],
                        source=self.name
                    )
                    items.append(item)
        except Exception as e:
            # This will either re-raise error....
            # or handle it and we will continue/ignore the error.
            handle_aws_exception(e, self, directory)

        # If we got an error, `items` will be empty.
        # In this case, in the future, we won't try to retrieve this directory since we are
        # setting it blank here.
        listing = DirectoryListing(directory=directory, items=items)

        self.log_about_items(
            items=listing.item_mapping().values(),
            path=listing.directory.path
        )

        self.local_cache[directory] = listing
        return listing.get_item(name)

    def retrieved_items_map(
            self, directory: DirectoryOrPath
    ) -> Optional[Mapping[str, DirectoryItem]]:
        directory = Directory.from_path(directory)
        listing = self.local_cache.get(directory)
        if listing is None:
            return None
        return listing.item_mapping()

Classes

class SsmParamStoreProvider

AwsProvider is the Base class for Aws-associated config providers.

There is some aws specific error handing that this class helps with among the aws providers.

This is the Default Doc message, you will want to override this doc-comment in any subclasses.

Expand source code

class SsmParamStoreProvider(AwsProvider):
    attributes_to_skip_while_copying = ['_store_get_params_paginator']
    name = "ssm"
    _store_get_params_paginator = None

    @property
    def _get_params_paginator(self):
        paginator = self._store_get_params_paginator
        if not paginator:
            paginator = boto_clients.ssm.get_paginator('get_parameters_by_path')
            self._store_get_params_paginator = paginator
        return paginator

    def get_item(
            self,
            name: str,
            directory: Optional[DirectoryOrPath],
            directory_chain: DirectoryChain,
            provider_chain: ProviderChain,
            environ: Directory
    ) -> Optional[DirectoryItem]:
        if directory is None:
            return None
        return self._item_only_for_directory(name=name, directory=directory)

    def _item_only_for_directory(
            self, name: str, directory: DirectoryOrPath
    ) -> Optional[DirectoryItem]:
        directory = Directory.from_path(directory)
        if not directory:
            return None

        listing = self.local_cache.get(directory)
        if listing:
            return listing.get_item(name)

        items = []
        try:
            # We need to lookup the directory listing from Secrets Manager.
            pages = self._get_params_paginator.paginate(
                Path=directory.path,
                Recursive=False,
                WithDecryption=True,
            )

            for p in pages:
                for item_info in p['Parameters']:
                    item_path: str = item_info['Name']
                    item = DirectoryItem(
                        directory=directory,
                        name=item_path.split("/")[-1],
                        value=item_info['Value'],
                        source=self.name
                    )
                    items.append(item)
        except Exception as e:
            # This will either re-raise error....
            # or handle it and we will continue/ignore the error.
            handle_aws_exception(e, self, directory)

        # If we got an error, `items` will be empty.
        # In this case, in the future, we won't try to retrieve this directory since we are
        # setting it blank here.
        listing = DirectoryListing(directory=directory, items=items)

        self.log_about_items(
            items=listing.item_mapping().values(),
            path=listing.directory.path
        )

        self.local_cache[directory] = listing
        return listing.get_item(name)

    def retrieved_items_map(
            self, directory: DirectoryOrPath
    ) -> Optional[Mapping[str, DirectoryItem]]:
        directory = Directory.from_path(directory)
        listing = self.local_cache.get(directory)
        if listing is None:
            return None
        return listing.item_mapping()

Ancestors

Class variables

var attributes_to_skip_while_copying
var botocore_error_ignored_exception : botocore.exceptions.BotoCoreError: Inherited from: AwsProvider.botocore_error_ignored_exception

This means that any attempt to communicat with aws service will probably fail; probable due to a corrupted or missing aws credentials.
var is_cacher: Inherited from: AwsProvider.is_cacher

Easy way to figure out if a provider is a ProviderCacher or just a normal provider. Should be set to True for provider subclasses that are …
var name: Inherited from: AwsProvider.name

This is the value that will normally be set to the items DirectoryItem.source, also displayed when logging out the names of providers …
var needs_directory: Inherited from: AwsProvider.needs_directory

By default, providers can't really use a None for a directory when calling get_item(). If you CAN work with a None directory then set this to …
var query_before_cache_if_possible: Inherited from: AwsProvider.query_before_cache_if_possible

If True, and this is before any other providers that have this set to False, the cacher will be consulted AFTER that provider(s). In this way I'll …

Static methods

def __init_subclass__(thread_sharable=Default, attributes_to_skip_while_copying: Optional[Iterable[str]] = Default, **kwargs)

Inherited from: AwsProvider.__init_subclass__

Args

thread_sharable: If False: While a dependency is lazily auto-created, we will ensure we do it per-thread, and not make it visible …

def grab() ‑> ~T

Inherited from: AwsProvider.grab

Gets a potentially shared dependency from the current udpend.context.XContext …

def proxy() ‑> ~R

Inherited from: AwsProvider.proxy

Returns a proxy-object, that when and attribute is asked for, it will proxy it to the current object of cls …

def proxy_attribute(attribute_name: str) ‑> Any

Inherited from: AwsProvider.proxy_attribute

Returns a proxy-object, that when and attribute is asked for, it will proxy it to the current attribute value on the current object of cls …

Instance variables

var obj : Self: Inherited from: AwsProvider.obj

class property/attribute that will return the current dependency for the subclass it's asked on by calling Dependency.grab, passing no extra …

Methods

def __call__(self, func)

Inherited from: AwsProvider.__call__

This makes Resource subclasses have an ability to be used as function decorators by default unless this method is overriden to provide some other …

def __copy__(self)

Inherited from: AwsProvider.__copy__

Basic shallow copy protection (I am wondering if I should just remove this default copy code) …

def directory_has_error(self, directory: Directory)

Inherited from: AwsProvider.directory_has_error

If a directory had an error in the past, this returns true. For informational purposes only.

def get_item(self, name: str, directory: Optional[DirectoryOrPath], directory_chain: DirectoryChain, provider_chain: ProviderChain, environ: Directory) ‑> Optional[DirectoryItem]

Inherited from: AwsProvider.get_item

Grabs a config value for name in directory …

Expand source code

def get_item(
        self,
        name: str,
        directory: Optional[DirectoryOrPath],
        directory_chain: DirectoryChain,
        provider_chain: ProviderChain,
        environ: Directory
) -> Optional[DirectoryItem]:
    if directory is None:
        return None
    return self._item_only_for_directory(name=name, directory=directory)

def get_value(self, name: str, directory: Optional[DirectoryOrPath], directory_chain: DirectoryChain, provider_chain: ProviderChain, environ: Directory)

Inherited from: AwsProvider.get_value

Gets an item's value for directory from provider. Return None if not found.

def mark_errored_directory(self, directory: Directory)

Inherited from: AwsProvider.mark_errored_directory

If a directory has an error, this is called. For informational purposes only.

def retrieved_items_map(self, directory: DirectoryOrPath) ‑> Optional[Mapping[str, DirectoryItem]]

Inherited from: AwsProvider.retrieved_items_map

Should return a read-only lower-case item name TO item mapping. You can easily get one of these from a DirectoryList object's item_mapping() …

Expand source code

def retrieved_items_map(
        self, directory: DirectoryOrPath
) -> Optional[Mapping[str, DirectoryItem]]:
    directory = Directory.from_path(directory)
    listing = self.local_cache.get(directory)
    if listing is None:
        return None
    return listing.item_mapping()