from django.core.management.base import BaseCommand from django.conf import settings import pymongo import pymysql import decimal from bson import ObjectId import sys class Command(BaseCommand): help = 'Transfer data from MySQL to MongoDB' def handle(self, *args, **options): # Get MongoDB configuration from settings.py mongo_settings = settings.DATABASES['mongodb'] # Define a dictionary to map MySQL tables to MongoDB collections # table_to_collection = { # 'vw_offices': 'v1_offices', # 'vw_agents': 'v1_agents', # 'vw_properties_location': 'v1_properties_location', # 'vw_features': 'v1_features', # 'vw_medias': 'v1_medias', # 'vw_properties_feature': 'v1_properties_feature', # 'vw_properties_media': 'v1_properties_media', # 'vw_properties': 'v1_properties', # 'vw_types': 'v1_types', # 'vw_usages': 'v1_usages', # } table_to_collection = {} table_to_nested_collection = { 'vw_properties': { 'collection': 'v1_properties_complete', 'single_data': { 'type_id': {'table': 'vw_types', 'field': 'type_id', 'alias': 'type'}, 'usage_id': {'table': 'vw_usages', 'field': 'usage_id', 'alias': 'usage'}, 'office_id': {'table': 'vw_offices', 'field': 'office_id', 'alias': 'office'}, 'agent_id': {'table': 'vw_agents', 'field': 'agent_id', 'alias': 'agent'}, 'location_id': {'table': 'vw_properties_location', 'field': 'property_location_id', 'alias': 'location'}, }, 'multiple_data': [ {'property_id': {'table': 'vw_properties_media', 'field': 'property_id', 'alias': 'medias'}}, ], 'query': { 'property_id': { 'table': 'vw_features', 'field': '*', 'filter': 'feature_id IN (SELECT feature_id FROM vw_properties_feature WHERE property_id={})', 'alias': 'features', }, } }, # Add other tables as needed } # Configure MongoDB connection using settings mongo_client = pymongo.MongoClient( host=mongo_settings['CLIENT']['host'], port=mongo_settings['CLIENT']['port'], username=mongo_settings['CLIENT']['username'], password=mongo_settings['CLIENT']['password'], authSource=mongo_settings['CLIENT']['authSource'], authMechanism=mongo_settings['CLIENT']['authMechanism'], ) mongo_db = mongo_client[mongo_settings['NAME']] # Use the database name from settings # Connect to MySQL using pymysql mysql_settings = settings.DATABASES['default'] mysql_conn = pymysql.connect( host=mysql_settings['HOST'], port=mysql_settings['PORT'], user=mysql_settings['USER'], password=mysql_settings['PASSWORD'], db=mysql_settings['NAME'], ) mysql_cursor = mysql_conn.cursor() try: # Loop through the table-to-collection mapping and transfer data for mysql_table, mongo_collection_name in table_to_collection.items(): mongo_db.drop_collection(mongo_collection_name) try: mysql_cursor.execute(f"SELECT * FROM {mysql_table}") table_data = mysql_cursor.fetchall() # Fetch column names separately column_names = [desc[0] for desc in mysql_cursor.description] # Create a list to store the converted data converted_data = [] # Iterate through each row of data and convert it to a dictionary for row in table_data: row_dict = dict(zip(column_names, row)) # Convert Decimal objects to float before inserting into MongoDB for key, value in row_dict.items(): if isinstance(value, decimal.Decimal): row_dict[key] = float(value) if key == "property_id": row_dict[key] = int(value) converted_data.append(row_dict) # Insert the converted data into the MongoDB collection mongo_collection = mongo_db[mongo_collection_name] mongo_collection.insert_many(converted_data) self.stdout.write(self.style.SUCCESS(f'Data transfer from {mysql_table} to {mongo_collection_name} completed successfully!')) except Exception as e: self.stdout.write(self.style.ERROR(f'An error occurred: {str(e)} - {sys.exc_info()}')) # Transfer nested data dynamically for nested_table, nested_info in table_to_nested_collection.items(): if isinstance(nested_info, str): # If nested_info is a string, use it as the collection name directly nested_collection_name = nested_info nested_relations = {} mongo_db.drop_collection(nested_collection_name) elif isinstance(nested_info, dict): # If nested_info is a dictionary, use its values nested_collection_name = nested_info.get('collection', '') nested_relations = nested_info.get('single_data', {}) multiple_data = nested_info.get('multiple_data', []) query_data = nested_info.get('query', {}) mongo_db.drop_collection(nested_collection_name) else: # Handle unexpected format self.stdout.write(self.style.ERROR(f'Unexpected format for nested_info. Skipping... - {nested_info} - {sys.exc_info()}')) continue try: # Check if the collection already exists if nested_collection_name in mongo_db.list_collection_names(): #print(nested_collection_name) mongo_db.drop_collection(nested_collection_name) # Fetch data from the nested table mysql_cursor.execute(f"SELECT * FROM {nested_table}") nested_table_data = mysql_cursor.fetchall() # Fetch column names separately nested_column_names = [desc[0] for desc in mysql_cursor.description] # Create a list to store the converted nested data converted_nested_data = [] # Iterate through each row of data and convert it to a dictionary for row in nested_table_data: row_dict = dict(zip(nested_column_names, row)) # Convert Decimal objects to float before inserting into MongoDB for key, value in row_dict.items(): if isinstance(value, decimal.Decimal): row_dict[key] = float(value) if key == "property_id": row_dict[key] = int(value) # Fetch single related data for relation_name, related_table_info in nested_relations.items(): #print(f"Fetching related data for relation: {relation_name}") related_field = related_table_info['field'] # Retrieve the mapped field from the related_table_info related_table = related_table_info['table'] related_cursor = mysql_conn.cursor() try: # Check if the relation_name is not None before executing the query if row_dict[relation_name] is not None: relation_value = row_dict[relation_name] #print(f"Fetching related data for relation '{relation_name}' with value: {relation_value}") # Construct the query based on the relation_name and use the mapped field query = f"SELECT * FROM {related_table} WHERE {related_field} = %s" related_cursor.execute(query, (relation_value,)) related_table_data = related_cursor.fetchall() related_column_names = [desc[0] for desc in related_cursor.description] converted_related_data = [] for related_row in related_table_data: related_row_dict = dict(zip(related_column_names, related_row)) # Convert Decimal objects to float before inserting into MongoDB for key, value in related_row_dict.items(): if isinstance(value, decimal.Decimal): related_row_dict[key] = float(value) if key == "property_id": # Adjust the key to match the field in the related table related_row_dict[key] = int(value) converted_related_data.append(related_row_dict) # Add related data to the main row_dict relation_key = related_table_info.get('alias', relation_name) row_dict[relation_key] = converted_related_data[0] if converted_related_data else None except Exception as e: #print(f"Error fetching related data for relation '{relation_name}': {str(e)}") self.stdout.write(self.style.ERROR(f'An error occurred while fetching related data: {str(e)}')) finally: related_cursor.close() # Fetch multiple related data for relation_info in multiple_data: relation_key, related_table_info = next(iter(relation_info.items()), (None, None)) if not relation_key or not related_table_info: self.stdout.write(self.style.ERROR( 'Invalid format for multiple_data. Skipping... - {relation_key} - {related_table_info} - {sys.exc_info()}')) continue # print(f"Fetching related data for relation: {relation_key}") related_cursor = mysql_conn.cursor() try: if row_dict[relation_key] is not None: relation_value = row_dict[relation_key] # print(f"Fetching related data for relation '{relation_key}' with value: {relation_value}") query = f"SELECT * FROM {related_table_info['table']} WHERE {related_table_info['field']} = %s" related_cursor.execute(query, (relation_value,)) related_table_data = related_cursor.fetchall() related_column_names = [desc[0] for desc in related_cursor.description] converted_related_data = [] for related_row in related_table_data: related_row_dict = dict(zip(related_column_names, related_row)) for key, value in related_row_dict.items(): if isinstance(value, decimal.Decimal): related_row_dict[key] = float(value) if key == "property_id": related_row_dict[key] = int(value) relation_alias = related_table_info.get('alias', relation_key) if relation_alias not in row_dict: row_dict[relation_alias] = [] row_dict[relation_alias].append(related_row_dict) except Exception as e: #print(f"Error fetching related data for relation '{relation_key}': {str(e)} - {sys.exc_info()}") self.stdout.write(self.style.ERROR( f'An error occurred while fetching related data: {str(e)} - {sys.exc_info()}')) finally: related_cursor.close() # Fetch data using the query for query_name, query_info in query_data.items(): query_alias = query_info.get('alias', query_name) query_table = query_info.get('table', '') query_field = query_info.get('field', '') query_filter = query_info.get('filter', '') # print(f"Fetching data for query: {query_name}") related_cursor = mysql_conn.cursor() # Initialize converted_query_data outside the loop converted_query_data = [] try: #print(f"query_name : {query_name}") if row_dict[query_name] is not None: query_value = row_dict[query_name] #print(f"query_value : {query_value}") # print(f"Fetching data for query '{query_name}' with value: {query_value}") query_filter = query_filter.format(query_value) # Apply the format to the query filter query = f"SELECT {query_field} FROM {query_table} WHERE {query_filter}" #print(f"query : {query}") related_cursor.execute(query) query_table_data = related_cursor.fetchall() #print(f"query_table_data : {query_table_data}") query_column_names = [desc[0] for desc in related_cursor.description] # Populate converted_query_data with query results converted_query_data = [dict(zip(query_column_names, query_row)) for query_row in query_table_data] except Exception as e: #print(f"Error fetching data for query '{query_name}': {str(e)}") self.stdout.write(self.style.ERROR( f'An error occurred while fetching data: {str(e)} - {sys.exc_info()}')) finally: related_cursor.close() # Assign the populated converted_query_data to row_dict[query_alias] outside the loop row_dict[query_alias] = converted_query_data if converted_query_data else None #print(f"query_alias : {query_alias}") #print(f"row_dict : {row_dict[query_alias]}") converted_nested_data.append(row_dict) # Generate a unique ObjectId for the _id field if it doesn't exist for item in converted_nested_data: if '_id' not in item: unique_id = ObjectId() item['_id'] = unique_id # Insert the converted nested data into the MongoDB collection nested_mongo_collection = mongo_db[nested_collection_name] nested_mongo_collection.insert_many(converted_nested_data) self.stdout.write( self.style.SUCCESS(f'Data transfer from {nested_table} to {nested_collection_name} completed successfully!')) except Exception as e: #print(f"Error fetching data from table '{nested_table}': {str(e)}") self.stdout.write(self.style.ERROR( f'An error occurred while fetching data: {str(e)} - {sys.exc_info()}')) finally: # Close cursor mysql_cursor.close() except Exception as e: self.stdout.write(self.style.ERROR(f'An error occurred: {str(e)} - {sys.exc_info()}')) finally: # Close database connections mysql_conn.close() mongo_client.close()