import os import socket import time import iperf3 import markdown2 import psutil import json from flask import render_template import functions.version as version from lib.api_logger import APILogger from lib.athenz_client import AthenzClient from lib.s3_client import S3Client from products.dbs import MySQLHandler, PostgreSQLHandler, MongoDBHandler, RedisHandler, CassandraHandler, OpenSearchHandler class APIFunctions: # Configure logging def __init__(self, env='prod'): self.env = env self.api_logger = APILogger(log_file='cqa-test-app.log', environment=self.env) self.swagger_path = '/swagger' self.api_spec_path = '/swagger.json' # Initialize database handlers self.mysql_handler = MySQLHandler(self.api_logger) self.postgresql_handler = PostgreSQLHandler(self.api_logger) self.mongodb_handler = MongoDBHandler(self.api_logger) self.redis_handler = RedisHandler(self.api_logger) self.cassandra_handler = CassandraHandler(self.api_logger) self.opensearch_handler = OpenSearchHandler(self.api_logger) # Cache for AthenzClient instances (keyed by account_name:provider_domain) self._athenz_clients = {} #set env (prod or dev) def set_env(self, env): self.env = env if self.env == 'dev': self.access_key = os.getenv("DEV_S3_ACCESS_KEY") self.secret_key = os.getenv("DEV_S3_SECRET_KEY") elif self.env in ['prod', 'prod_secret', 'prod_top_secret', 'ssk_prod']: self.access_key = os.getenv("PROD_S3_ACCESS_KEY") self.secret_key = os.getenv("PROD_S3_SECRET_KEY") else: raise ValueError(f"Invalid environment: {self.env}. Must be either 'dev' or 'prod' or 'prod_secret' or 'prod_top_secret' or 'ssk_prod'") def run_cmd(self, reqparse, request): # Parsing arguments with call rest_api parser = reqparse.RequestParser() parser.add_argument('cmd', required=True, type=str, help='need to input cmd..') args = parser.parse_args() # Get 'cmd' argument and execute shell command by os.popen cmd = args['cmd'] if not cmd: return self.make_common_response( res_msg={'msg': 'Bad Request, cmd is empty or None'}, request=request, res_code=400 ) res = os.popen(cmd).read() # Return response [CMD] command, [Reponse] response logs from shell return self.make_common_response( res_msg={'msg': '[CMD]:' + cmd + ', [Response]:' + res}, request=request ) def run_iperf(self, reqparse, request): # Check Iperf Installed in server os (ex: sudo yum install iperf3 -y) if self._is_iperf_installed() is False: return self.make_common_response( res_msg={'msg': 'Internal Server Error, There is no "iperf3" in server, cannot run iperf testing..'}, request=request, res_code=500 ) # Parsing arguments with call rest_api (POST) parser = reqparse.RequestParser() parser.add_argument('host_ip', required=True, type=str, help='need to input "host_ip" for running iperf') parser.add_argument('host_port', required=True, type=str, help='need to input "host_port" for running iperf') parser.add_argument('duration', required=True, type=int, help='need to input "duration"(sec) for setting test time') parser.add_argument('protocol', required=True, type=str, help='need to input "protocol"(tcp/udp) for setting test protocol') params = parser.parse_args() # Check params if not params['host_ip'] or not params['host_port'] or not params['duration'] or not params['protocol']: return self.make_common_response( res_msg={'msg': 'Bad Request, some params are empty or None..'}, request=request, res_code=400 ) if params['protocol'] != 'tcp' and params['protocol'] != 'udp': return self.make_common_response( res_msg={'msg': 'Bad Request, protocol param must input "tcp" or "udp"..'}, request=request, res_code=400 ) # Set iperf3 client with parameters client = iperf3.Client() client.server_hostname = params['host_ip'] client.port = params['host_port'] client.duration = params['duration'] client.protocol = params['protocol'] res = client.run() # Check if 'iperf3' has Error in response, return error msg with 500 code (Internal error) if res.error: return self.make_common_response( res_msg={'msg': f'Internal Server Error, iperf3 Error={res.error}'}, request=request, res_code=500 ) # Return iperf3 testing result (tcp, udp) if client.protocol == 'tcp': sent_transfer = self._cal_to_mega(res.sent_bytes) recv_transfer = self._cal_to_mega(res.received_bytes) sent_bitrate = self._cal_to_mega(res.sent_bps) recv_bitrate = self._cal_to_mega(res.received_bps) res_msg = { 'msg': f'[TCP Result]: Host={client.server_hostname}:{client.port}, ' f'Duration={res.duration} sec, ' f'Transfer=(Sent){sent_transfer} MBytes, ' f'(Recv){recv_transfer} MBytes, ' f'Bitrate=(Sent){sent_bitrate} Mbit/sec, ' f'(Recv){recv_bitrate} Mbit/sec ' } else: transfer = self._cal_to_mega(res.bytes) bitrate = self._cal_to_mega(res.bps) lost_packet_percent = round(res.lost_percent, 2) res_msg = { 'msg': f'[UDP Result]: Host={client.server_hostname}:{client.port}, ' f'Duration={res.duration} sec, ' f'Transfer={transfer} MBytes, ' f'Bitrate={bitrate} Mbit/sec, ' f'LostPacket={lost_packet_percent} % ' } return self.make_common_response(res_msg=res_msg, request=request) def make_common_response( self, request, res_msg=None, res_code=200, log_sensitive=True ): if res_msg is None: res_msg = {"msg": "Test OK"} if request is None: return { "status": res_code, "body": res_msg, "headers": {"Content-Type": "text/json"}, } # Set Additional info to response message res_msg["code"] = res_code res_msg["method"] = request.method res_msg["path"] = request.path res_msg["server_info"] = {} res_msg["server_info"]["app_version"] = version.VERSION res_msg["server_info"]["server_host"] = request.host res_msg["server_info"]["container_id"], res_msg["server_info"]["server_ip"] = self._get_server_info() res_msg["server_info"]["server_ip"] = os.environ.get("HOST_IP", res_msg["server_info"]["server_ip"]) res_msg["client_info"] = {} res_msg["client_info"]["client_ip"] = self._get_client_ip(request) # Logging API information log_msg = res_msg.copy() if not log_sensitive: log_msg["msg"] = "***" request_logs = f'code={log_msg["code"]}, method={log_msg["method"]}, path={log_msg["path"]}, msg={log_msg["msg"]}' server_logs = f'app_version={log_msg["server_info"]["app_version"]}, server_ip={os.environ.get("HOST_IP", log_msg["server_info"]["server_ip"])}, ' \ f'server_host={log_msg["server_info"]["server_host"]}, container_id={log_msg["server_info"]["container_id"]}' client_logs = f'client_ip={log_msg["client_info"]["client_ip"]}' self.api_logger.logger.info(f'{request_logs} || {server_logs} || {client_logs}') self.make_request_header_log(request=request) # Set response result to json format json_res = json.dumps(res_msg, indent=4) return json_res, res_code def set_swagger_ui(self, app, flask_swagger_ui): # Set Swagger UI & path swaggerui_blueprint = flask_swagger_ui.get_swaggerui_blueprint( self.swagger_path, self.api_spec_path, config={ 'validatorUrl': "none", 'app_name': "CQA Test Application" } ) app.register_blueprint(swaggerui_blueprint, url_prefix=self.swagger_path) def _cal_to_mega(self, val): return round(val / 1000000, 2) def _is_iperf_installed(self): res = os.popen("iperf3 -v").read() if res.find("iperf 3") > -1: return True return False def render_markdown(self, request, file_path): # Build the full path to the requested Markdown file full_path = os.path.join('./resources/wiki', file_path) # Check if the requested file exists and is a Markdown file if not os.path.isfile(full_path) or not file_path.endswith('.md'): return self.make_common_response( res_msg={'msg': 'Error: ' + full_path + ': File not exists'}, request=request, res_code=404 ) # Read the Markdown content with open(full_path, 'r', encoding='utf-8') as md_file: markdown_content = md_file.read() # Convert Markdown to HTML html_content = markdown2.markdown( markdown_content, extras=["tables", "fenced-code-blocks", "cuddled-lists"] ) # Check if the requested file exists css_file_path = './static/css/md_file.css' if not os.path.isfile(css_file_path): return self.make_common_response( res_msg={'msg': 'Error: CSS file not found'}, request=request, res_code=404 ) # Render the HTML content using the template rendered_html_content = render_template( 'basic.html', title=file_path.replace('.md', '').capitalize(), content=html_content ) # Logging self.make_common_response( res_msg={'msg': 'Converted to HTML successfully'}, request=request ) # Return the HTML content as the response return rendered_html_content # make request headers logs through request header def make_request_header_log(self, request): log_prefix = 'request headers : ' log_contents = '' # Make request header log contents by request headers for key in request.headers.keys(): log_contents += f'[{key}]={request.headers[key]}, ' # Check if log contents is not empty, save logs if log_contents != '': log_contents = ', '.join(log_contents.split(', ')[:-1]) self.api_logger.logger.debug(log_prefix + log_contents) def _get_client_ip(self, request): # Get the client IP from the 'X-Forwarded-For' header client_ip = request.headers.get('X-Forwarded-For', request.remote_addr) if client_ip and client_ip.lower() != 'unknown': # Header contains a comma-separated list of IPs. e.g. X-Forwarded-For: client, proxy1, proxy2 client_ip = client_ip.split(',')[0].strip() else: # List of headers to check if 'X-Forwarded-For' didn't provide the IP headers_to_check = [ 'Proxy-Client-IP', 'WL-Proxy-Client-IP', 'HTTP_CLIENT_IP', 'HTTP_X_FORWARDED_FOR' ] # Iterate through the list of headers and check if any of them contains the client IP for header in headers_to_check: client_ip = request.headers.get(header) if client_ip and client_ip.lower() != 'unknown': break # If no valid client IP was found in any headers, fallback to the remote address if not client_ip: client_ip = request.remote_addr return client_ip def _get_server_info(self): # Get the hostname of the server. hostname = socket.gethostname() if not hostname: return "unknown", "unknown" # Get the IP address of the server. try: server_ip = socket.gethostbyname(hostname) except Exception as e: self.api_logger.logger.debug(f"Failed to get server ip from hostname={hostname}") return hostname, "unknown" return hostname, server_ip def establish_connection(self, host, port, duration): with socket.create_connection((host, port)) as sock: # Send a message to server that the connection is established try: sock.sendall(b"Establishing connection for duration") except socket.error as e: print(f"An error occurred while sending data: {e}") return False # Or handle the error as needed # Keep connection within a 'duration' period time.sleep(duration) def get_established_connections_count(self, port): connections = psutil.net_connections() # Get the number of established connections based on port number established_count = sum(1 for conn in connections if conn.status == psutil.CONN_ESTABLISHED and ( conn.laddr.port == port)) return established_count def get_athenz_access_token(self, reqparse, request): request_body = request.get_json(silent=True) if request_body: self.api_logger.logger.info(f"Request body: {request_body}") parser = reqparse.RequestParser() parser.add_argument( "account_name", required=True, type=str, help="need to input 'account_name'", ) parser.add_argument( "provider_domain", required=True, type=str, help="need to input 'provider_domain'", ) args = parser.parse_args() try: cache_key = f"{args['account_name']}:{args['provider_domain']}" if cache_key not in self._athenz_clients: self._athenz_clients[cache_key] = AthenzClient( args["account_name"], args["provider_domain"] ) athenz_client = self._athenz_clients[cache_key] access_token = athenz_client.get_access_token() return self.make_common_response( res_msg={"msg": access_token}, request=request, log_sensitive=False ) except Exception as e: self.api_logger.logger.error(f"Failed to get access token: {str(e)}") return self.make_common_response( res_msg={"msg": f"Internal Server Error, Athenz Error={str(e)}"}, request=request, res_code=500, ) def put_test_fos(self, reqparse, request, bucket_name, obj_name): is_success, response = self._check_fos_path_and_permission(bucket_name, obj_name) if not is_success: return response # Check bucket and object name if not bucket_name or not obj_name: return self.make_common_response( res_msg={"msg": "Bad Request, invalid bucket or object"}, request=request, res_code=400, ) # Parse "data" input from request body request_body = request.get_json(silent=True) if request_body: self.api_logger.logger.info(f"Request body: {request_body}") parser = reqparse.RequestParser() parser.add_argument( "data", required=True, type=str, help="need to put the content in 'data'", ) input_data = parser.parse_args()["data"] # Download object file from flava object storage s3_client = S3Client(self.access_key, self.secret_key, self.env) is_success, response = s3_client.download_object(bucket_name=bucket_name, obj_name=obj_name, dst_name=obj_name) if not is_success: return self.make_common_response( res_code=response["ResponseMetadata"]["HTTPStatusCode"], res_msg={"msg": f"{response['Error']['Message']}"}, request=request, ) # Downloaded file update and upload file to flava object storage is_success, response = self._file_write_and_upload_fos( file_name=obj_name, input_data=input_data, dest_bucket=bucket_name, dest_obj=obj_name, client=s3_client, request=request ) if not is_success: return response # Get object file from flava object storage and return response _, response = s3_client.get_object(bucket_name=bucket_name, obj_name=obj_name) return self.make_common_response( res_code=response["ResponseMetadata"]["HTTPStatusCode"], res_msg={ "msg": f"Successfully put the data into FOS Bucket={bucket_name}, Object={obj_name}, Content={response['Body'].read().decode('utf-8')}" }, request=request, ) def get_fos_object(self, request, bucket_name, obj_name): is_success, response = self._check_fos_path_and_permission(bucket_name, obj_name) if not is_success: return response # Get object file from flava object storage and return response s3_client = S3Client(self.access_key, self.secret_key, self.env) is_success, response = s3_client.get_object(bucket_name=bucket_name, obj_name=obj_name) if not is_success: return { "code": response["ResponseMetadata"]["HTTPStatusCode"], "msg": f"{response['Error']['Message']}" } return { "code": response["ResponseMetadata"]["HTTPStatusCode"], "msg": f"Successfully get the data into FOS Bucket={bucket_name}, Object={obj_name}, Content={response['Body'].read().decode('utf-8')}" } def test_fos_crud(self, reqparse, request, bucket_name, obj_name): #get env value from the FE request_body = request.get_json(silent=True) env = request_body.get('env', 'prod') self.set_env(env) #init S3Client s3_client = S3Client(self.access_key, self.secret_key, env) is_success, response = self._check_fos_path_and_permission(bucket_name, obj_name) if not is_success: return response all_result = "" # Parse "data" input from request body request_body = request.get_json(silent=True) if request_body: self.api_logger.logger.info(f"Request body: {request_body}") parser = reqparse.RequestParser() parser.add_argument( "data", required=True, type=str, help="need to put the content in 'data'", ) input_data = parser.parse_args()["data"] # 1. CREATE New Object: Upload new file to bucket object on flava object storage init_data = "This is new object made by cqa-test-app." is_success, response = self._file_write_and_upload_fos( file_name=obj_name, input_data=init_data, dest_bucket=bucket_name, dest_obj=obj_name, client=s3_client, request=request ) if not is_success: return response # Validate New object exist and init data is_success, response = self._validate_get_fos_object(s3_client, bucket_name, obj_name, init_data) if not is_success: return response all_result += "CREATE/GET object-PASS, " # ADD 2s time sleep for testing check time.sleep(2) # 2. UPDATE Object: Upload updated file with input data to bucket object on flava object storage is_success, response = self._file_write_and_upload_fos( file_name=obj_name, input_data=input_data, dest_bucket=bucket_name, dest_obj=obj_name, client=s3_client, request=request ) if not is_success: return response # Validate Updated object exist and updated data is_success, response = self._validate_get_fos_object(s3_client, bucket_name, obj_name, input_data) if not is_success: return response all_result += "UPDATE/GET object-PASS, " # ADD 2s time sleep for testing check time.sleep(2) # 3. DELETE OBJECT is_success, response = s3_client.delete_object(bucket_name=bucket_name, obj_name=obj_name) if not is_success: return { "code": response["ResponseMetadata"]["HTTPStatusCode"], "msg": f"{response['Error']['Message']}" } all_result += "DELETE object-PASS" # ADD 2s time sleep for testing check time.sleep(2) # Completed FOS Object CRUD Test, Return result return { "code": 200, "msg": f"All Passed CRUD Test: [{all_result}], Bucket={bucket_name}, Object={obj_name}" } #upload object to flava object storage (reuse test_fos_crud code) def create_fos_object(self, reqparse, request, bucket_name, obj_name): self._initialize_s3_client(request, bucket_name, obj_name) all_result = "" request_body = request.get_json(silent=True) if request_body: self.api_logger.logger.info(f"Request body: {request_body}") parser = reqparse.RequestParser() parser.add_argument( "data", required=True, type=str, help="need to put the content in 'data'", ) input_data = parser.parse_args()["data"] # CREATE New Object: Upload new file to bucket object on flava object storage is_success, response = self._file_write_and_upload_fos( file_name=obj_name, input_data=input_data, dest_bucket=bucket_name, dest_obj=obj_name, client=self.s3_client, request=request ) if not is_success: return response # Validate New object exist and init data is_success, response = self._validate_get_fos_object(self.s3_client, bucket_name, obj_name, input_data) if not is_success: return response all_result += "Upload object-PASS" return { "code": 200, "msg": f"Passed Upload Test: [{all_result}], Bucket={bucket_name}, Object={obj_name}" } #delete object from flava object storage (reuse test_fos_crud code) def delete_fos_object(self, request, bucket_name, obj_name): env = request.args.get('env', 'prod') self.set_env(env) self.s3_client = S3Client(self.access_key, self.secret_key, env) all_result = "" # Check if the object exists before deleting is_success, response = self.s3_client.get_object(bucket_name=bucket_name, obj_name=obj_name) if not is_success: return { "code": 404, "msg": f"Object not found: {response['Error']['Message']}" } # DELETE OBJECT from s3 is_success, response = self.s3_client.delete_object(bucket_name=bucket_name, obj_name=obj_name) if not is_success: return { "code": response["ResponseMetadata"]["HTTPStatusCode"], "msg": f"{response['Error']['Message']}" } all_result += "DELETE object-PASS" # Delete local file if it exists try: os.remove(obj_name) except FileNotFoundError: pass return { "code": 200, "msg": f"Passed Delete Test: [{all_result}], Bucket={bucket_name}, Object={obj_name}" } #get list of objects from flava object storage def get_list_object(self, request, bucket_name): env = request.args.get('env', 'prod') self.set_env(env) #init S3Client self.s3_client = S3Client(self.access_key, self.secret_key, env) is_success, response = self._check_fos_path_and_permission(bucket_name, None) if not is_success: return response all_result = "" is_success, response = self.s3_client.get_list_object(bucket_name) if not is_success: return { "code": response["ResponseMetadata"]["HTTPStatusCode"], "msg": f"{response['Error']['Message']}" } all_result += "List object-PASS" #save only key(obj_name) of object object_keys = [obj['Key'] for obj in response] return { "code": 200, "msg": f"Passed List Test: [{all_result}], Bucket={bucket_name}, Object={object_keys}" } #update bucket (enable or disable static website) def update_bucket(self, request, bucket_name, action): self._initialize_s3_client(request, bucket_name, None) all_result = "" # check the action input if action not in ['enable', 'disable']: return{ "code": 400, "msg": "Bad Request, action must be 'enable' or 'disable'" } is_success, response = self.s3_client.update_bucket(bucket_name, action) if not is_success: return { "code": response["ResponseMetadata"]["HTTPStatusCode"], "msg": f"{response['Error']['Message']}" } all_result += "Update bucket-PASS" return { "code": 200, "msg": f"Passed Update Bucket Test: [{all_result}], Bucket={bucket_name}, Action={action}" } #download object from flava object storage. (you can see downloaded file '/explorer/') def download_object(self, request, bucket_name, obj_name): env = request.args.get('env', 'prod') self.set_env(env) #init S3Client self.s3_client = S3Client(self.access_key, self.secret_key, env) # Check bucket and object name if not bucket_name or not obj_name: return self.make_common_response( res_msg={"msg": "Bad Request, invalid bucket or object"}, request=request, res_code=400, ) all_result = "" is_success, response = self.s3_client.download_object(bucket_name=bucket_name, obj_name=obj_name, dst_name=obj_name) if not is_success: return self.make_common_response( res_code=response["ResponseMetadata"]["HTTPStatusCode"], res_msg={"msg": f"{response['Error']['Message']}"}, request=request, ) all_result += "Download object-PASS" return { "code": 200, "msg": f"Passed Download Test: [{all_result}], Bucket={bucket_name}, Object={obj_name}" } # Init methods (used for create/delete/update) def _initialize_s3_client(self, request, bucket_name, obj_name): request_body = request.get_json(silent=True) or {} env = request_body.get('env', 'prod') self.set_env(env) self.s3_client = S3Client(self.access_key, self.secret_key, env) self._check_fos_path_and_permission(bucket_name, obj_name) def _check_fos_path_and_permission(self, bucket_name, obj_name=None): # Check bucket name if not bucket_name: return False, { "code": 400, "msg": "Bad Request, invalid bucket" } # Check object name (if required) if obj_name is not None and not obj_name: return False, { "code": 400, "msg": "Bad Request, invalid object" } # Check & Get keys from environment env if not self.access_key or not self.secret_key: return False, { "code": 403, "msg": "Forbidden, need to access key and secret key on env" } return True, None def _validate_get_fos_object(self, s3_client, bucket_name, obj_name, expected_data): # READ(GET) Object: Get object file from flava object storage (dev) and return response is_success, response = s3_client.get_object(bucket_name=bucket_name, obj_name=obj_name) if not is_success: return False, { "code": response["ResponseMetadata"]["HTTPStatusCode"], "msg": f"{response['Error']['Message']}" } # Validate FOS Object data, If failed return result object_data = response['Body'].read().decode('utf-8') if object_data != expected_data: return False, { "code": 200, "msg": f"Failed to validate FOS Object data [Expected Result]={expected_data}, [Actual Result]={object_data}" } return True, None def _file_write_and_upload_fos(self, file_name, input_data, dest_bucket, dest_obj, client, request): # File write with file_name and input_data try: with open(file_name, 'w') as file: file.write(input_data) except Exception as e: return self.make_common_response( res_msg={"msg": f"Internal Server Error, Failed to write file {file_name}, {e}"}, request=request, res_code=500, ) # Upload File to dest FOS object of bucket is_success, response = client.upload_object( bucket_name=dest_bucket, obj_name=dest_obj, src_name=file_name ) if not is_success: return False, { "code": response.status_code, "msg": f"{response.text}" } # After success upload file, set object acl(='public-read') to flava object storage (dev) is_success, response = client.set_object_acl( bucket_name=dest_bucket, obj_name=dest_obj ) if not is_success: return False, { "code": response["ResponseMetadata"]["HTTPStatusCode"], "msg": f"{response['Error']['Message']}" } # Delete the local file after successful upload and ACL setting try: os.remove(file_name) except Exception as e: self.api_logger.logger.error(f"Failed to delete file {file_name}: {e}") return True, None #DBS: connect to host def connect_to_db(self, product, service, user=None, password=None, database=None): if product == 'mysql': return self.mysql_handler.connect(service, user, password, database) elif product == 'postgresql': return self.postgresql_handler.connect(service, user, password, database) elif product == 'mongodb': return self.mongodb_handler.connect(service, user, password, database) elif product == 'redis': return self.redis_handler.connect(service, user, password, database) elif product == 'cassandra': return self.cassandra_handler.connect(service, user, password, database) elif product == 'opensearch': return self.opensearch_handler.connect(service, user, password, database) else: return { "code": 400, "msg": "Bad Request, invalid product" } # DBS: execute query def execute_query(self, product, query, database=None): if product == 'mysql': return self.mysql_handler.execute_query(query, database) elif product == 'postgresql': return self.postgresql_handler.execute_query(query, database) elif product == 'mongodb': return self.mongodb_handler.execute_query(query, database) elif product == 'redis': return self.redis_handler.execute_query(query, database) elif product == 'cassandra': return self.cassandra_handler.execute_query(query, database) elif product == 'opensearch': return self.opensearch_handler.execute_query(query, database) else: return { "code": 400, "msg": "Invalid product or connection failed" } # DBS: execute OpenSearch API query def execute_opensearch_api_query(self, service, user, password, method, endpoint, payload=None): try: # Create a temporary OpenSearch handler for the API call temp_handler = OpenSearchHandler(self.api_logger) # Connect first connect_result = temp_handler.connect(service, user, password) if connect_result.get('code') != 200: return connect_result # Execute the API query return temp_handler.execute_api_query(method, endpoint, payload) except Exception as e: return { "code": 500, "msg": f"Internal Server Error: {e}" } # FOS Dual Environment: Write to both ssk_prod and prod simultaneously def write_to_dual_env(self, reqparse, request, bucket_name, obj_name): # Parse "data" input from request body request_body = request.get_json(silent=True) if request_body: self.api_logger.logger.info(f"Request body: {request_body}") parser = reqparse.RequestParser() parser.add_argument( "data", required=True, type=str, help="need to put the content in 'data'", ) input_data = parser.parse_args()["data"] # Initialize clients for both ssk_prod and prod environments envs = ['ssk_prod', 'prod'] results = {} for env_name in envs: try: # Set environment and get credentials self.set_env(env_name) is_success, response = self._check_fos_path_and_permission(bucket_name, obj_name) if not is_success: results[env_name] = { "success": False, "code": response.get("code", 500), "msg": response.get("msg", "Unknown error") } continue # Initialize S3Client for this environment s3_client = S3Client(self.access_key, self.secret_key, env_name) # Upload file to FOS is_success, response = self._file_write_and_upload_fos( file_name=f"{obj_name}_{env_name}", input_data=input_data, dest_bucket=bucket_name, dest_obj=obj_name, client=s3_client, request=request ) if not is_success: results[env_name] = { "success": False, "code": response.get("code", 500), "msg": response.get("msg", "Upload failed") } continue # Validate uploaded object is_success, response = self._validate_get_fos_object(s3_client, bucket_name, obj_name, input_data) if not is_success: results[env_name] = { "success": False, "code": response.get("code", 500), "msg": response.get("msg", "Validation failed") } continue results[env_name] = { "success": True, "code": 200, "msg": f"Successfully uploaded and validated", "bucket": bucket_name, "object": obj_name, "data": input_data } except Exception as e: # Catch any unexpected errors self.api_logger.logger.error(f"Unexpected error for {env_name}: {str(e)}") results[env_name] = { "success": False, "code": 500, "msg": f"Unexpected error: {str(e)}" } # Determine overall success - check if all environments that have results are successful overall_success = len(results) > 0 and all(results[env]["success"] for env in results.keys()) return { "code": 200 if overall_success else 500, "overall_success": overall_success, "environments": results } # FOS Dual Environment: List objects from both ssk_prod and prod simultaneously def list_from_dual_env(self, request, bucket_name): # Initialize clients for both ssk_prod and prod environments envs = ['ssk_prod', 'prod'] results = {} for env_name in envs: try: # Set environment and get credentials self.set_env(env_name) is_success, response = self._check_fos_path_and_permission(bucket_name, None) if not is_success: results[env_name] = { "success": False, "code": response.get("code", 500), "msg": response.get("msg", "Unknown error"), "objects": [] } continue # Initialize S3Client for this environment s3_client = S3Client(self.access_key, self.secret_key, env_name) # Get list of objects list_result = s3_client.get_list_object(bucket_name) # Handle inconsistent return types from get_list_object if isinstance(list_result, tuple) and len(list_result) == 2: is_success, response = list_result elif isinstance(list_result, dict): # Error case - returned dict instead of tuple is_success = False response = list_result else: # Unexpected return type is_success = False response = {"code": 500, "msg": "Unexpected response format"} if not is_success: # Safely extract error information error_code = 500 error_msg = "Failed to list objects" try: if isinstance(response, dict): if "ResponseMetadata" in response and "HTTPStatusCode" in response["ResponseMetadata"]: error_code = response["ResponseMetadata"]["HTTPStatusCode"] if "Error" in response and "Message" in response["Error"]: error_msg = response["Error"]["Message"] elif "msg" in response: error_msg = response["msg"] elif "code" in response: error_code = response["code"] except Exception as e: self.api_logger.logger.error(f"Error parsing response for {env_name}: {e}") results[env_name] = { "success": False, "code": error_code, "msg": error_msg, "objects": [] } continue # Extract object keys safely object_keys = [] if isinstance(response, list): try: object_keys = [obj['Key'] for obj in response if isinstance(obj, dict) and 'Key' in obj] except Exception as e: self.api_logger.logger.error(f"Error extracting object keys for {env_name}: {e}") results[env_name] = { "success": False, "code": 500, "msg": f"Error processing object list: {str(e)}", "objects": [] } continue results[env_name] = { "success": True, "code": 200, "msg": f"Successfully listed {len(object_keys)} object(s)", "bucket": bucket_name, "objects": object_keys, "count": len(object_keys) } except Exception as e: # Catch any unexpected errors self.api_logger.logger.error(f"Unexpected error for {env_name}: {str(e)}") results[env_name] = { "success": False, "code": 500, "msg": f"Unexpected error: {str(e)}", "objects": [] } # Determine overall success - check if all environments that have results are successful overall_success = len(results) > 0 and all(results[env]["success"] for env in results.keys()) return { "code": 200 if overall_success else 500, "overall_success": overall_success, "environments": results } # MySQL Dual Environment: Connect to both MySQL databases simultaneously def connect_to_dual_mysql(self, request): request_body = request.get_json(silent=True) if not request_body: return { "code": 400, "overall_success": False, "msg": "Bad Request, missing request body" } # Get connection info for both environments ssk_info = request_body.get('ssk', {}) kks_info = request_body.get('kks', {}) envs = { 'ssk': ssk_info, 'kks': kks_info } results = {} for env_name, db_info in envs.items(): try: service = db_info.get('service') user = db_info.get('user') password = db_info.get('password') database = db_info.get('database') # Validate required fields if not service or not user or not password: results[env_name] = { "success": False, "code": 400, "msg": "Missing required fields (service, user, password)" } continue # Try to connect result = self.mysql_handler.connect(service, user, password, database) if result.get('code') == 200: results[env_name] = { "success": True, "code": 200, "msg": result.get('msg', 'Connected successfully'), "service": service, "database": database or 'N/A' } else: results[env_name] = { "success": False, "code": result.get('code', 500), "msg": result.get('msg', 'Connection failed') } except Exception as e: self.api_logger.logger.error(f"Connection error for {env_name}: {str(e)}") results[env_name] = { "success": False, "code": 500, "msg": f"Connection error: {str(e)}" } # Determine overall success - check if all environments that have results are successful overall_success = len(results) > 0 and all(results[env]["success"] for env in results.keys()) return { "code": 200 if overall_success else 500, "overall_success": overall_success, "environments": results } # MySQL Dual Environment: Execute query on both MySQL databases simultaneously def execute_dual_mysql_query(self, request): request_body = request.get_json(silent=True) if not request_body: return { "code": 400, "overall_success": False, "msg": "Bad Request, missing request body" } query = request_body.get('query') database = request_body.get('database') # Get connection info for reconnection ssk_info = request_body.get('ssk', {}) kks_info = request_body.get('kks', {}) if not query: return { "code": 400, "overall_success": False, "msg": "Bad Request, missing query" } envs = { 'ssk': ssk_info, 'kks': kks_info } results = {} for env_name, db_info in envs.items(): try: # Reconnect before executing query to ensure correct connection if db_info: service = db_info.get('service') user = db_info.get('user') password = db_info.get('password') db = db_info.get('database') or database if service and user and password: # Create a temporary MySQL handler for this environment temp_handler = MySQLHandler(self.api_logger) connect_result = temp_handler.connect(service, user, password, db) if connect_result.get('code') != 200: results[env_name] = { "success": False, "code": connect_result.get('code', 500), "msg": f"Reconnection failed: {connect_result.get('msg', 'Unknown error')}", "query": query } continue # Execute query result = temp_handler.execute_query(query, database) else: # Use existing connection result = self.mysql_handler.execute_query(query, database) else: # Use existing connection result = self.mysql_handler.execute_query(query, database) if result.get('code') == 200: results[env_name] = { "success": True, "code": 200, "msg": result.get('msg', 'Query executed successfully'), "query": query, "database": database or 'N/A', "columns_names": result.get('columns_names', []), "results": result.get('results', []) } else: results[env_name] = { "success": False, "code": result.get('code', 500), "msg": result.get('msg', 'Query execution failed'), "query": query } except Exception as e: self.api_logger.logger.error(f"Query execution error for {env_name}: {str(e)}") results[env_name] = { "success": False, "code": 500, "msg": f"Query execution error: {str(e)}", "query": query } # Determine overall success - check if all environments that have results are successful overall_success = len(results) > 0 and all(results[env]["success"] for env in results.keys()) return { "code": 200 if overall_success else 500, "overall_success": overall_success, "environments": results }