import asyncio import logging from typing import Dict, List, Optional, Any, Union from concurrent.futures import ThreadPoolExecutor, as_completed from app.models.mysql_model import MySQLModel logger = logging.getLogger(__name__) class DatabaseService: """Service for handling multi-database operations in flask-api""" def __init__(self): self.mysql = MySQLModel() self.executor = ThreadPoolExecutor(max_workers=10) self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}") def execute_mysql_query(self, query: str, params: tuple = None) -> List[Dict]: """Execute MySQL query""" try: self.logger.debug(f"Executing MySQL query") return self.mysql.execute_query(query, params) except Exception as e: self.logger.error(f"MySQL query error: {e}") return [{"error": f"MySQL query failed: {str(e)}"}] async def execute_queries_async(self, queries: List[Dict]) -> Dict[str, List[Dict]]: """Execute multiple queries asynchronously""" loop = asyncio.get_event_loop() tasks = [] self.logger.info(f"Executing {len(queries)} queries asynchronously") for query_config in queries: db_type = query_config.get("database") query_id = query_config.get("id", f"{db_type}_query") if db_type == "mysql": task = loop.run_in_executor( self.executor, self.execute_mysql_query, query_config.get("query"), query_config.get("params"), ) else: self.logger.warning(f"Unknown database type: {db_type}") continue tasks.append((query_id, task)) results = {} for query_id, task in tasks: try: results[query_id] = await task self.logger.debug(f"Async query {query_id} completed successfully") except Exception as e: self.logger.error(f"Async query {query_id} failed: {e}") results[query_id] = [{"error": f"Query {query_id} failed: {str(e)}"}] self.logger.info(f"Completed {len(results)} async queries") return results def execute_parallel_queries(self, queries: List[Dict]) -> Dict[str, List[Dict]]: """Execute multiple queries in parallel using threads""" futures = {} results = {} self.logger.info(f"Executing {len(queries)} queries in parallel") with ThreadPoolExecutor(max_workers=min(len(queries), 10)) as executor: # Submit all queries for query_config in queries: db_type = query_config.get("database") query_id = query_config.get("id", f"{db_type}_query") if db_type == "mysql": future = executor.submit( self.execute_mysql_query, query_config.get("query"), query_config.get("params"), ) else: self.logger.warning(f"Unknown database type: {db_type}") continue futures[query_id] = future # Collect results for query_id, future in futures.items(): try: results[query_id] = future.result(timeout=60) self.logger.debug( f"Parallel query {query_id} completed successfully" ) except Exception as e: self.logger.error(f"Parallel query {query_id} failed: {e}") results[query_id] = [ {"error": f"Query {query_id} failed: {str(e)}"} ] self.logger.info(f"Completed {len(results)} parallel queries") return results def combine_data( self, datasets: Dict[str, List[Dict]], join_config: Optional[Dict] = None ) -> List[Dict]: """Combine data from multiple databases""" try: if not datasets: self.logger.warning("No datasets provided for combination") return [] # Filter out error results valid_datasets = { k: v for k, v in datasets.items() if v and not (len(v) == 1 and "error" in v[0]) } if not valid_datasets: self.logger.warning("No valid datasets found after filtering errors") return [] if len(valid_datasets) == 1: self.logger.info("Single dataset, returning as-is") return list(valid_datasets.values())[0] if join_config: self.logger.info("Joining datasets with configuration") return self._join_datasets(valid_datasets, join_config) else: # Simple concatenation self.logger.info("Concatenating datasets") combined = [] for dataset_name, data in valid_datasets.items(): for item in data: item["_source"] = dataset_name combined.append(item) return combined except Exception as e: self.logger.error(f"Error combining data: {e}") return [{"error": f"Data combination failed: {str(e)}"}] def _join_datasets( self, datasets: Dict[str, List[Dict]], join_config: Dict ) -> List[Dict]: """Join datasets based on configuration""" try: primary_dataset = join_config.get("primary") join_field = join_config.get("join_field") join_type = join_config.get("type", "left") # left, inner, outer if primary_dataset not in datasets: raise ValueError(f"Primary dataset '{primary_dataset}' not found") primary_data = datasets[primary_dataset] result = [] self.logger.info( f"Joining {len(datasets)} datasets on field '{join_field}'" ) for primary_row in primary_data: combined_row = primary_row.copy() # Join with other datasets for dataset_name, data in datasets.items(): if dataset_name == primary_dataset: continue # Find matching rows matching_rows = [ row for row in data if row.get(join_field) == primary_row.get(join_field) ] if matching_rows: # Merge data from matching rows for match in matching_rows: for key, value in match.items(): if key != join_field: combined_row[f"{dataset_name}_{key}"] = value elif join_type == "left": # Add null values for missing data in left join sample_row = data[0] if data else {} for key in sample_row.keys(): if key != join_field: combined_row[f"{dataset_name}_{key}"] = None result.append(combined_row) self.logger.info(f"Join completed with {len(result)} records") return result except Exception as e: self.logger.error(f"Error joining datasets: {e}") return [{"error": f"Dataset join failed: {str(e)}"}] def get_database_stats(self) -> Dict: """Get database connection statistics""" try: self.logger.info("Getting database connection statistics") mysql_stats = self.mysql.test_connection() return { "mysql": mysql_stats, "service_status": "operational", } except Exception as e: self.logger.error(f"Error getting database stats: {e}") return {"error": str(e), "service_status": "error"} def cleanup(self): """Cleanup resources""" try: self.logger.info("Cleaning up database service resources") self.mysql.cleanup() self.executor.shutdown(wait=True) self.logger.info("Database service cleanup completed") except Exception as e: self.logger.error(f"Error during cleanup: {e}")