Mercurial > public > tweet-analysis
view dependencies/python/aws_controller.py @ 8:6541622b6127
add tweet analysis method
author | Dennis Concepcion Martin <dennisconcepcionmartin@gmail.com> |
---|---|
date | Fri, 17 Sep 2021 21:10:02 +0200 |
parents | |
children |
line wrap: on
line source
import boto3 import base64 import json from botocore.exceptions import ClientError class AwsSecretsManager: @staticmethod def get_secret(secret_name, region_name='eu-west-2'): """ Get Secret Keys from AWS Secrets Manager :return: """ # Create a Secrets Manager client session = boto3.session.Session() client = session.client( service_name='secretsmanager', region_name=region_name ) try: get_secret_value_response = client.get_secret_value(SecretId=secret_name) except ClientError as e: if e.response['Error']['Code'] == 'DecryptionFailureException': # Secrets Manager can't decrypt the protected secret text using the provided KMS key. raise e elif e.response['Error']['Code'] == 'InternalServiceErrorException': # An error occurred on the server side. raise e elif e.response['Error']['Code'] == 'InvalidParameterException': # You provided an invalid value for a parameter. raise e elif e.response['Error']['Code'] == 'InvalidRequestException': # You provided a parameter value that is not valid for the current state of the resource. raise e elif e.response['Error']['Code'] == 'ResourceNotFoundException': # AWS Can't find the resource that you asked for. raise e else: # Decrypts secret using the associated KMS CMK. # Depending on whether the secret is a string or binary, one of these fields will be populated. if 'SecretString' in get_secret_value_response: secret = get_secret_value_response['SecretString'] else: secret = base64.b64decode(get_secret_value_response['SecretBinary']) return json.loads(secret) class AwsComprehend: @staticmethod def get_sentiment(tweets): """ :param tweets: list (string), required :return: dict """ # Create a Comprehend client session = boto3.session.Session() comprehend = session.client( service_name='comprehend', region_name='eu-west-2' ) response = comprehend.batch_detect_sentiment( TextList=tweets, LanguageCode='en' ) return response