Merge pull request #1 from GlueOps/feature/update-logging

Feature/update logging

Author: fernandoataoldotcom

glueops/aws.py

@@ -1,17 +1,26 @@
+from typing import List
+
 import boto3
-import glueops.logging
+from botocore.client import BaseClient
+
 
-logger = glueops.logging.configure()
+def create_aws_client(service: str, region: str = 'us-east-1') -> BaseClient:
+    return boto3.client(service, region_name=region)
 
 
-def create_aws_client(service):
-    return boto3.client(service, region_name='us-east-1')
+def get_resource_arns_using_tags(
+    tags: List[dict],
+    aws_resource_filters: List[str]
+) -> List[str]:
+    """Retrieve ARNs of resources with specific tags.
 
+    Args:
+        tags (List[dict]): _in format {"Key": "<the-key>", "Value": "<the-value>"}
+        aws_resource_filters (List[str]): AWS resource filter list
 
-def get_resource_arns_using_tags(tags, aws_resource_filters):
-    """Retrieve ARNs of resources with specific tags."""
-    logger.info(
-        f"Checking to see if this arn was already requested/created with tags: {tags}")
+    Returns:
+        List[str]: ARNs of resources
+    """
     tagging = create_aws_client('resourcegroupstaggingapi')
     tags = {item['Key']: item['Value'] for item in tags}
     response = tagging.get_resources(
@@ -20,8 +29,7 @@ def get_resource_arns_using_tags(tags, aws_resource_filters):
         ],
         ResourceTypeFilters=aws_resource_filters
     )
-
-    arns = [item['ResourceARN']
-            for item in response.get('ResourceTagMappingList', [])]
-
-    return arns
+    return [
+        item['ResourceARN']
+        for item in response.get('ResourceTagMappingList', [])
+    ]

glueops/certificates.py

@@ -3,22 +3,16 @@
 from cryptography.exceptions import UnsupportedAlgorithm, InvalidSignature
 
 
-def extract_serial_number_from_cert_string(cert_string):
-    try:
-        certificate = x509.load_pem_x509_certificate(cert_string.encode(), default_backend())
-        decimal_serial = certificate.serial_number
-        
-        # Convert to hexadecimal
-        hex_serial = format(decimal_serial, 'X')
-        
-        # Ensure an even number of digits for correct byte representation
-        if len(hex_serial) % 2 != 0:
-            hex_serial = '0' + hex_serial
-        
-        # Insert colons between every 2 characters
-        colon_separated_serial = ":".join(hex_serial[i:i+2] for i in range(0, len(hex_serial), 2))
-        return colon_separated_serial
+def extract_serial_number_from_cert_string(cert_string: str) -> str:
+    certificate = x509.load_pem_x509_certificate(cert_string.encode(), default_backend())
+    decimal_serial = certificate.serial_number
 
-    except (ValueError, UnsupportedAlgorithm, InvalidSignature) as e:
-        # You can specify more exceptions as needed based on the specific errors you're encountering.
-        raise ValueError("Provided certificate string is not parsable: " + str(e))
+    # Convert to hexadecimal
+    hex_serial = format(decimal_serial, 'X')
+
+    # Ensure an even number of digits for correct byte representation
+    if len(hex_serial) % 2 != 0:
+        hex_serial = '0' + hex_serial
+
+    # Return serial with colons between every 2 characters
+    return ":".join(hex_serial[i:i+2] for i in range(0, len(hex_serial), 2)).upper()

glueops/checksum_tools.py

@@ -1,6 +1,6 @@
 import zlib
 
-def string_to_crc32(input_string):
+def string_to_crc32(input_string: str) -> str:
     """Compute CRC32 checksum for a given string and return it in hexadecimal format."""
     # Encode the string to bytes
     data_bytes = input_string.encode('utf-8')
@@ -9,6 +9,4 @@ def string_to_crc32(input_string):
     checksum = zlib.crc32(data_bytes) & 0xffffffff
 
     # Convert checksum to hexadecimal
-    hex_checksum = hex(checksum)
-    
-    return hex_checksum
+    return hex(checksum)

glueops/fastapi.py

@@ -0,0 +1,62 @@
+import inspect
+import json
+import logging
+from typing import Union
+
+
+class JsonFormatter(logging.Formatter):
+    def format(self, record):
+        log_entry = {
+            'timestamp': self.formatTime(record, self.datefmt),
+            'name': record.name,
+            'level': record.levelname,
+            'message': record.getMessage(),
+        }
+
+        if record.exc_info:
+            log_entry['exception'] = self.formatException(record.exc_info)
+
+        return json.dumps(log_entry)
+
+
+def configure(
+    name=None,
+    level: Union[str, int]=logging.ERROR,
+    *handlers: logging.Handler
+) -> logging.Logger:
+    """Configure and return a logger with GlueOps default configuration
+
+    Args:
+        name (_type_, optional): The name of the logger. Defaults to None.
+        level (Union[str, int], optional): The log level, as an int or str. Defaults to logging.ERROR.
+            Must be less restrictive than the level applied to additional handlers for those handlers to receive logs
+        *handlers (logging.Handler, optional): Add any additional handlers that may be desired.
+
+    Returns:
+        logging.Logger: Instance of configured logger
+    """
+    if name is None:
+        frame = inspect.stack()[1]
+        module = inspect.getmodule(frame[0])
+        name = module.__name__ if module else "defaultLogger"
+
+    logger = logging.getLogger(name)
+    logger.setLevel(level)
+
+    # create default formatter and handler
+    default_handler = logging.StreamHandler()
+    json_formatter = JsonFormatter()
+    default_handler.setFormatter(json_formatter)
+
+    # remove handlers not configured by this module
+    for handler in logger.handlers[:]:
+        logger.removeHandler(handler)
+
+    # add default handler
+    logger.addHandler(default_handler)
+
+    # add custom handlers
+    for handler in handlers:
+        logger.addHandler(handler)
+
+    return logger