Welcome to the AWS Code Examples Repository. This repo contains code examples used in the AWS documentation, AWS SDK Developer Guides, and more. For more information, see the Readme.md file below.
|
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
|
||
|
|
# SPDX-License-Identifier: Apache-2.0
|
||
|
|
|
||
|
|
import logging
|
||
|
|
|
||
|
|
import boto3
|
||
|
|
from botocore.exceptions import ClientError
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
# snippet-start:[python.example_code.ssm.OpsItemWrapper.class]
|
||
|
|
# snippet-start:[python.example_code.ssm.OpsItemWrapper.decl]
|
||
|
|
class OpsItemWrapper:
|
||
|
|
"""Encapsulates AWS Systems Manager OpsItem actions."""
|
||
|
|
|
||
|
|
def __init__(self, ssm_client):
|
||
|
|
"""
|
||
|
|
:param ssm_client: A Boto3 Systems Manager client.
|
||
|
|
"""
|
||
|
|
self.ssm_client = ssm_client
|
||
|
|
self.id = None
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_client(cls):
|
||
|
|
"""
|
||
|
|
:return: A OpsItemWrapper instance.
|
||
|
|
"""
|
||
|
|
ssm_client = boto3.client("ssm")
|
||
|
|
return cls(ssm_client)
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ssm.OpsItemWrapper.decl]
|
||
|
|
|
||
|
|
# snippet-start:[python.example_code.ssm.CreateOpsItem]
|
||
|
|
def create(self, title, source, category, severity, description):
|
||
|
|
"""
|
||
|
|
Create an OpsItem
|
||
|
|
|
||
|
|
:param title: The OpsItem title.
|
||
|
|
:param source: The OpsItem source.
|
||
|
|
:param category: The OpsItem category.
|
||
|
|
:param severity: The OpsItem severity.
|
||
|
|
:param description: The OpsItem description.
|
||
|
|
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
response = self.ssm_client.create_ops_item(
|
||
|
|
Title=title,
|
||
|
|
Source=source,
|
||
|
|
Category=category,
|
||
|
|
Severity=severity,
|
||
|
|
Description=description,
|
||
|
|
)
|
||
|
|
self.id = response["OpsItemId"]
|
||
|
|
except self.ssm_client.exceptions.OpsItemLimitExceededException as err:
|
||
|
|
logger.error(
|
||
|
|
"Couldn't create ops item because you have exceeded your open OpsItem limit. "
|
||
|
|
"Here's why: %s: %s",
|
||
|
|
err.response["Error"]["Code"],
|
||
|
|
err.response["Error"]["Message"],
|
||
|
|
)
|
||
|
|
raise
|
||
|
|
except ClientError as err:
|
||
|
|
logger.error(
|
||
|
|
"Couldn't create ops item %s. Here's why: %s: %s",
|
||
|
|
title,
|
||
|
|
err.response["Error"]["Code"],
|
||
|
|
err.response["Error"]["Message"],
|
||
|
|
)
|
||
|
|
raise
|
||
|
|
# snippet-end:[python.example_code.ssm.CreateOpsItem]
|
||
|
|
|
||
|
|
# snippet-start:[python.example_code.ssm.DeleteOpsItem]
|
||
|
|
def delete(self):
|
||
|
|
"""
|
||
|
|
Delete the OpsItem.
|
||
|
|
"""
|
||
|
|
if self.id is None:
|
||
|
|
return
|
||
|
|
try:
|
||
|
|
self.ssm_client.delete_ops_item(OpsItemId=self.id)
|
||
|
|
print(f"Deleted ops item with id {self.id}")
|
||
|
|
self.id = None
|
||
|
|
except ClientError as err:
|
||
|
|
logger.error(
|
||
|
|
"Couldn't delete ops item %s. Here's why: %s: %s",
|
||
|
|
self.id,
|
||
|
|
err.response["Error"]["Code"],
|
||
|
|
err.response["Error"]["Message"],
|
||
|
|
)
|
||
|
|
raise
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ssm.DeleteOpsItem]
|
||
|
|
|
||
|
|
# snippet-start:[python.example_code.ssm.DescribeOpsItem]
|
||
|
|
def describe(self):
|
||
|
|
"""
|
||
|
|
Describe an OpsItem.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
paginator = self.ssm_client.get_paginator("describe_ops_items")
|
||
|
|
ops_items = []
|
||
|
|
for page in paginator.paginate(
|
||
|
|
OpsItemFilters=[
|
||
|
|
{"Key": "OpsItemId", "Values": [self.id], "Operator": "Equal"}
|
||
|
|
]
|
||
|
|
):
|
||
|
|
ops_items.extend(page["OpsItemSummaries"])
|
||
|
|
|
||
|
|
for item in ops_items:
|
||
|
|
print(
|
||
|
|
f"The item title is {item['Title']} and the status is {item['Status']}"
|
||
|
|
)
|
||
|
|
return len(ops_items) > 0
|
||
|
|
except ClientError as err:
|
||
|
|
logger.error(
|
||
|
|
"Couldn't describe ops item %s. Here's why: %s: %s",
|
||
|
|
self.id,
|
||
|
|
err.response["Error"]["Code"],
|
||
|
|
err.response["Error"]["Message"],
|
||
|
|
)
|
||
|
|
raise
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ssm.DescribeOpsItem]
|
||
|
|
|
||
|
|
# snippet-start:[python.example_code.ssm.UpdateOpsItem]
|
||
|
|
def update(self, title=None, description=None, status=None):
|
||
|
|
"""
|
||
|
|
Update an OpsItem.
|
||
|
|
|
||
|
|
:param title: The new OpsItem title.
|
||
|
|
:param description: The new OpsItem description.
|
||
|
|
:param status: The new OpsItem status.
|
||
|
|
:return:
|
||
|
|
"""
|
||
|
|
args = dict(OpsItemId=self.id)
|
||
|
|
if title is not None:
|
||
|
|
args["Title"] = title
|
||
|
|
if description is not None:
|
||
|
|
args["Description"] = description
|
||
|
|
if status is not None:
|
||
|
|
args["Status"] = status
|
||
|
|
try:
|
||
|
|
self.ssm_client.update_ops_item(**args)
|
||
|
|
except ClientError as err:
|
||
|
|
logger.error(
|
||
|
|
"Couldn't update ops item %s. Here's why: %s: %s",
|
||
|
|
self.id,
|
||
|
|
err.response["Error"]["Code"],
|
||
|
|
err.response["Error"]["Message"],
|
||
|
|
)
|
||
|
|
raise
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ssm.UpdateOpsItem]
|
||
|
|
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ssm.OpsItemWrapper.class]
|