DEV Community

Developer213
Developer213

Posted on

change s3 path for DQM

import boto3
from botocore.exceptions import NoCredentialsError
from datetime import datetime

class TestConfig:
    def __init__(self, monitoring_uri, model_name):
        self.monitoring_uri = monitoring_uri
        self.model_name = model_name

    def create_config(self):
        # Define the date components
        day = datetime.now().strftime("%d")
        month = datetime.now().strftime("%m")
        year = datetime.now().strftime("%Y")

        # Initialize S3 client
        s3_client = boto3.client("s3")

        # Extract bucket name and construct base S3 path
        bucket_name = self.monitoring_uri.split('/')[2]
        base_path = f"{'/'.join(self.monitoring_uri.split('/')[3:])}/{self.model_name}-monitor/{year}/{month}/{day}/"

        # List objects within the specified day path to retrieve available hour folders
        try:
            response = s3_client.list_objects_v2(Bucket=bucket_name, Prefix=base_path, Delimiter="/")
            hour_folders = [content["Prefix"].split("/")[-2] for content in response.get("CommonPrefixes", [])]

            # Use the first available hour folder if it exists
            hour = hour_folders[0] if hour_folders else None

            if hour:
                # Construct the paths using the dynamically fetched hour
                self.results_constraints_uri = f"{self.monitoring_uri}/{self.model_name}-monitor/{year}/{month}/{day}/{hour}/constraints.json"
                self.results_statistics_uri = f"{self.monitoring_uri}/{self.model_name}-monitor/{year}/{month}/{day}/{hour}/statistics.json"
                self.results_constraints_violations_uri = f"{self.monitoring_uri}/{self.model_name}-monitor/{year}/{month}/{day}/{hour}/constraint_violations.json"

                # Print the paths for verification
                print("Constraints URI:", self.results_constraints_uri)
                print("Statistics URI:", self.results_statistics_uri)
                print("Violations URI:", self.results_constraints_violations_uri)
            else:
                print("No hour folders found in the specified S3 path.")

        except NoCredentialsError:
            print("No AWS credentials found. Please configure your AWS credentials.")

# Instantiate the class with the provided monitoring_uri and model_name
monitoring_uri = ''
model_name = ''
test_config = TestConfig(monitoring_uri, model_name)

# Run create_config to test the hour retrieval and path generation
test_config.create_config()
Enter fullscreen mode Exit fullscreen mode

Top comments (0)