diff 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 diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/dependencies/python/aws_controller.py	Fri Sep 17 21:10:02 2021 +0200
@@ -0,0 +1,71 @@
+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