Source code for aws_resource_search.res.ssm

# -*- coding: utf-8 -*-

import typing as T
import dataclasses
from datetime import datetime

import sayt.api as sayt
import aws_arns.api as arns
import aws_console_url.api as acu

from .. import res_lib as rl

if T.TYPE_CHECKING:
    from ..ars_def import ARS


[docs]@dataclasses.dataclass class SsmParameter(rl.ResourceDocument): # fmt: off type: str = dataclasses.field(metadata={"field": sayt.NgramWordsField(name="type", minsize=2, maxsize=4, stored=True)}) tier: str = dataclasses.field(metadata={"field": sayt.NgramWordsField(name="tier", minsize=2, maxsize=4, stored=True)}) param_arn: str = dataclasses.field(metadata={"field": sayt.StoredField(name="param_arn")}) # fmt: on @property def last_modified_date(self) -> datetime: return rl.get_datetime(self.raw_data, "LastModifiedDate") @property def description(self) -> str: return rl.get_description(self.raw_data, "Description")
[docs] @classmethod def from_resource(cls, resource, bsm, boto_kwargs): return cls( raw_data=resource, type=resource["Type"], tier=resource["Tier"], id=resource["Name"], name=resource["Name"], param_arn=arns.res.SSMParameter.new( aws_account_id=bsm.aws_account_id, aws_region=bsm.aws_region, name=resource["Name"], ).to_arn(), )
@property def title(self) -> str: return rl.format_key_value("name", self.name) @property def subtitle(self) -> str: return "{}, {}, {}".format( rl.format_key_value("type", self.type), rl.format_key_value("tier", self.tier), self.short_subtitle, ) @property def autocomplete(self) -> str: return self.name @property def arn(self) -> str: return self.param_arn
[docs] def get_console_url(self, console: acu.AWSConsole) -> str: return console.ssm.get_parameter(name_or_arn=self.arn)
[docs] @classmethod def get_list_resources_console_url(cls, console: acu.AWSConsole) -> str: return console.ssm.parameters
# fmt: off
[docs] def get_details(self, ars: "ARS") -> T.List[rl.DetailItem]: from_detail = rl.DetailItem.from_detail url = self.get_console_url(console=ars.aws_console) detail_items = rl.DetailItem.get_initial_detail_items(doc=self, ars=ars) with rl.DetailItem.error_handling(detail_items): res = ars.bsm.ssm_client.get_parameter(Name=self.name) param = res["Parameter"] detail_items.extend([ from_detail("LastModifiedDate", self.raw_data.get("LastModifiedDate", "NA"), url=url), from_detail("Type", param.get("Type", "NA"), url=url), from_detail("Version", param.get("Version", "NA"), url=url), from_detail("DataType", param.get("DataType", "NA"), url=url), from_detail("Value", param.get("Value", "NA"), self.one_line(param.get("Value", "NA")), url=url), ]) with rl.DetailItem.error_handling(detail_items): res = ars.bsm.ssm_client.list_tags_for_resource( ResourceType="Parameter", ResourceId=self.name, ) tags = rl.extract_tags(res) detail_items.extend(rl.DetailItem.from_tags(tags, url)) return detail_items
# fmt: on
[docs]class SsmParameterSearcher(rl.BaseSearcher[SsmParameter]): pass
ssm_parameter_searcher = SsmParameterSearcher( # list resources service="ssm", method="describe_parameters", is_paginator=True, default_boto_kwargs={"PaginationConfig": {"MaxItems": 9999, "PageSize": 50}}, result_path=rl.ResultPath("Parameters"), # extract document doc_class=SsmParameter, # search resource_type=rl.SearcherEnum.ssm_parameter.value, fields=SsmParameter.get_dataset_fields(), cache_expire=rl.config.get_cache_expire(rl.SearcherEnum.ssm_parameter.value), more_cache_key=None, )