types-aiobotocore-sns

import asyncio from typing import TYPE_CHECKING from aiobotocore.session import get_session import os if TYPE_CHECKING: from types_aiobotocore_sns.client import SNSClient async def publish_message(): session = get_session() # Ensure AWS credentials/config are available via environment variables or config files # e.g., AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION async with session.create_client("sns") as client: client: "SNSClient" # Explicit type annotation for IDEs/MyPy topic_arn = os.environ.get('AWS_SNS_TOPIC_ARN', 'arn:aws:sns:REGION:ACCOUNT_ID:MyTopic') message = "Hello from types-aiobotocore-sns!" try: response = await client.publish(TopicArn=topic_arn, Message=message) print(f"Message published: {response['MessageId']}") except client.exceptions.TopicNotFoundException: print(f"Error: Topic {topic_arn} not found.") except Exception as e: print(f"An error occurred: {e}") if __name__ == "__main__": # You would typically set AWS_SNS_TOPIC_ARN, AWS_REGION, etc., as environment variables. # For this example, ensure they are set or default values are valid. asyncio.run(publish_message())