Data Migration Tool Hotfix (#4396)

* Add get_source_value to utils, and use it to get value from object or dict

* GitHub connector

* fix codacy

* remove print statement

* Remove Github Connector

* Data Migration Connector

- Create new connection with boilerplate
- Add Run button in Data Migration Plan

* minor

* fix codacy

* remove pygithub

* Minor

- Remove button from form and add to custom button
- Remove is_custom field
- Add Connector Type 'Custom'
- Show New Connection button only in developer mode

* [fix] logging

- only store the name, not the whole json

* [fix] tests

frappe/data_migration/doctype/data_migration_connector/connectors/base.py

@@ -5,19 +5,19 @@ from frappe.utils.password import get_decrypted_password
 class BaseConnection(with_metaclass(ABCMeta)):
 
 	@abstractmethod
-	def get(self):
+	def get(self, remote_objectname, fields=None, filters=None, start=0, page_length=10):
 		pass
 
 	@abstractmethod
-	def insert(self):
+	def insert(self, doctype, doc):
 		pass
 
 	@abstractmethod
-	def update(self):
+	def update(self, doctype, doc, migration_id):
 		pass
 
 	@abstractmethod
-	def delete(self):
+	def delete(self, doctype, migration_id):
 		pass
 
 	def get_password(self):

frappe/data_migration/doctype/data_migration_connector/data_migration_connector.js

@@ -2,7 +2,46 @@
 // For license information, please see license.txt
 
 frappe.ui.form.on('Data Migration Connector', {
-	refresh: function() {
+	onload(frm) {
+		if(frappe.boot.developer_mode) {
+			frm.add_custom_button(__('New Connection'), () => frm.events.new_connection(frm));
+		}
+	},
+	new_connection(frm) {
+		const d = new frappe.ui.Dialog({
+			title: __('New Connection'),
+			fields: [
+				{ label: __('Module'), fieldtype: 'Link', options: 'Module Def', reqd: 1 },
+				{ label: __('Connection Name'), fieldtype: 'Data', description: 'For e.g: Shopify Connection', reqd: 1 },
+			],
+			primary_action_label: __('Create'),
+			primary_action: (values) => {
+				let { module, connection_name } = values;
 
+				frm.events.create_new_connection(module, connection_name)
+					.then(r => {
+						if (r.message) {
+							const connector_name = connection_name
+								.replace('connection', 'Connector')
+								.replace('Connection', 'Connector')
+								.trim();
+
+							frm.set_value('connector_name', connector_name);
+							frm.set_value('connector_type', 'Custom');
+							frm.set_value('python_module', r.message);
+							frm.save();
+							frappe.show_alert(__(`New module created ${r.message}`));
+							d.hide();
+						}
+					});
+			}
+		});
+
+		d.show();
+	},
+	create_new_connection(module, connection_name) {
+		return frappe.call('frappe.data_migration.doctype.data_migration_connector.data_migration_connector.create_new_connection', {
+			module, connection_name
+		});
 	}
 });

frappe/data_migration/doctype/data_migration_connector/data_migration_connector.json

@@ -49,6 +49,7 @@
    "bold": 0, 
    "collapsible": 0, 
    "columns": 0, 
+   "depends_on": "eval:!doc.is_custom", 
    "fieldname": "connector_type", 
    "fieldtype": "Select", 
    "hidden": 0, 
@@ -61,7 +62,7 @@
    "label": "Connector Type", 
    "length": 0, 
    "no_copy": 0, 
-   "options": "Frappe\nPostgres", 
+   "options": "\nFrappe\nPostgres\nCustom", 
    "permlevel": 0, 
    "precision": "", 
    "print_hide": 0, 
@@ -80,6 +81,7 @@
    "bold": 0, 
    "collapsible": 0, 
    "columns": 0, 
+   "depends_on": "eval:doc.connector_type == 'Custom'", 
    "fieldname": "python_module", 
    "fieldtype": "Data", 
    "hidden": 0, 
@@ -110,7 +112,37 @@
    "bold": 0, 
    "collapsible": 0, 
    "columns": 0, 
-   "default": "localhost", 
+   "fieldname": "authentication_credentials", 
+   "fieldtype": "Section Break", 
+   "hidden": 0, 
+   "ignore_user_permissions": 0, 
+   "ignore_xss_filter": 0, 
+   "in_filter": 0, 
+   "in_global_search": 0, 
+   "in_list_view": 0, 
+   "in_standard_filter": 0, 
+   "label": "Authentication Credentials", 
+   "length": 0, 
+   "no_copy": 0, 
+   "permlevel": 0, 
+   "precision": "", 
+   "print_hide": 0, 
+   "print_hide_if_no_value": 0, 
+   "read_only": 0, 
+   "remember_last_selected_value": 0, 
+   "report_hide": 0, 
+   "reqd": 0, 
+   "search_index": 0, 
+   "set_only_once": 0, 
+   "unique": 0
+  }, 
+  {
+   "allow_bulk_edit": 0, 
+   "allow_on_submit": 0, 
+   "bold": 0, 
+   "collapsible": 0, 
+   "columns": 0, 
+   "default": "", 
    "fieldname": "hostname", 
    "fieldtype": "Data", 
    "hidden": 0, 
@@ -130,7 +162,7 @@
    "read_only": 0, 
    "remember_last_selected_value": 0, 
    "report_hide": 0, 
-   "reqd": 1, 
+   "reqd": 0, 
    "search_index": 0, 
    "set_only_once": 0, 
    "unique": 0
@@ -236,7 +268,7 @@
  "issingle": 0, 
  "istable": 0, 
  "max_attachments": 0, 
- "modified": "2017-10-08 14:34:30.603690", 
+ "modified": "2017-10-26 12:03:40.646348", 
  "modified_by": "Administrator", 
  "module": "Data Migration", 
  "name": "Data Migration Connector", 

frappe/data_migration/doctype/data_migration_connector/data_migration_connector.py

@@ -3,9 +3,11 @@
 # For license information, please see license.txt
 
 from __future__ import unicode_literals
-import frappe
+import frappe, os
 from frappe.model.document import Document
 from frappe import _
+from frappe.modules.export_file import create_init_py
+from .connectors.base import BaseConnection
 from .connectors.postgres import PostGresConnection
 from .connectors.frappe_connection import FrappeConnection
 
@@ -16,14 +18,14 @@ class DataMigrationConnector(Document):
 
 		if self.python_module:
 			try:
-				frappe.get_module(self.python_module)
+				get_connection_class(self.python_module)
 			except:
 				frappe.throw(frappe._('Invalid module path'))
 
 	def get_connection(self):
 		if self.python_module:
-			module = frappe.get_module(self.python_module)
-			return module.get_connection(self)
+			_class = get_connection_class(self.python_module)
+			return _class(self)
 		else:
 			if self.connector_type == 'Frappe':
 				self.connection = FrappeConnection(self)
@@ -32,8 +34,72 @@ class DataMigrationConnector(Document):
 
 		return self.connection
 
-	def get_objects(self, object_type, condition=None, selection="*"):
-		return self.connector.get_objects(object_type, condition, selection)
+@frappe.whitelist()
+def create_new_connection(module, connection_name):
+	if not frappe.conf.get('developer_mode'):
+		frappe.msgprint(_('Please enable developer mode to create new connection'))
+		return
+	# create folder
+	module_path = frappe.get_module_path(module)
+	connectors_folder = os.path.join(module_path, 'connectors')
+	frappe.create_folder(connectors_folder)
 
-	def get_join_objects(self, object_type, join_type, primary_key):
-		return self.connector.get_join_objects(object_type, join_type, primary_key)
+	# create init py
+	create_init_py(module_path, 'connectors', '')
+
+	connection_class = connection_name.replace(' ', '')
+	file_name = frappe.scrub(connection_name) + '.py'
+	file_path = os.path.join(module_path, 'connectors', file_name)
+
+	# create boilerplate file
+	with open(file_path, 'w') as f:
+		f.write(connection_boilerplate.format(connection_class=connection_class))
+
+	# get python module string from file_path
+	app_name = frappe.db.get_value('Module Def', module, 'app_name')
+	python_module = os.path.relpath(
+		file_path, '../apps/{0}'.format(app_name)).replace(os.path.sep, '.')[:-3]
+
+	return python_module
+
+def get_connection_class(python_module):
+	filename = python_module.rsplit('.', 1)[-1]
+	classname = frappe.unscrub(filename).replace(' ', '')
+	module = frappe.get_module(python_module)
+
+	raise_error = False
+	if hasattr(module, classname):
+		_class = getattr(module, classname)
+		if not issubclass(_class, BaseConnection):
+			raise_error = True
+	else:
+		raise_error = True
+
+	if raise_error:
+		raise ImportError(filename)
+
+	return _class
+
+connection_boilerplate = """from __future__ import unicode_literals
+from frappe.data_migration.doctype.data_migration_connector.connectors.base import BaseConnection
+
+class {connection_class}(BaseConnection):
+	def __init__(self, connector):
+		# self.connector = connector
+		# self.connection = YourModule(self.connector.username, self.get_password())
+		# self.name_field = 'id'
+		pass
+
+	def get(self, remote_objectname, fields=None, filters=None, start=0, page_length=10):
+		pass
+
+	def insert(self, doctype, doc):
+		pass
+
+	def update(self, doctype, doc, migration_id):
+		pass
+
+	def delete(self, doctype, migration_id):
+		pass
+
+"""

frappe/data_migration/doctype/data_migration_mapping/data_migration_mapping.py

@@ -5,6 +5,7 @@
 from __future__ import unicode_literals
 import frappe
 from frappe.model.document import Document
+from frappe.utils import get_source_value
 
 class DataMigrationMapping(Document):
 	def get_filters(self):
@@ -26,6 +27,7 @@ class DataMigrationMapping(Document):
 		return fields
 
 	def get_mapped_record(self, doc):
+		'''Build a mapped record using information from the fields table'''
 		mapped = frappe._dict()
 
 		key_fieldname = 'remote_fieldname'
@@ -35,13 +37,19 @@ class DataMigrationMapping(Document):
 			key_fieldname, value_fieldname = value_fieldname, key_fieldname
 
 		for field_map in self.fields:
+			key = get_source_value(field_map, key_fieldname)
+
 			if not field_map.is_child_table:
+				# field to field mapping
 				value = get_value_from_fieldname(field_map, value_fieldname, doc)
-				mapped[field_map.get(key_fieldname)] = value
 			else:
+				# child table mapping
 				mapping_name = field_map.child_table_mapping
-				value = get_mapped_child_records(mapping_name, doc.get(field_map.get(value_fieldname)))
-				mapped[field_map.get(key_fieldname)] = value
+				value = get_mapped_child_records(mapping_name,
+					doc.get(get_source_value(field_map, value_fieldname)))
+
+			mapped[key] = value
+
 		return mapped
 
 def get_mapped_child_records(mapping_name, child_docs):
@@ -53,12 +61,12 @@ def get_mapped_child_records(mapping_name, child_docs):
 	return mapped_child_docs
 
 def get_value_from_fieldname(field_map, fieldname_field, doc):
-	field_name = field_map.get(fieldname_field)
+	field_name = get_source_value(field_map, fieldname_field)
 
 	if field_name.startswith('eval:'):
 		value = frappe.safe_eval(field_name[5:], dict(frappe=frappe))
 	elif field_name[0] in ('"', "'"):
 		value = field_name[1:-1]
 	else:
-		value = doc.get(field_name)
+		value = get_source_value(doc, field_name)
 	return value

frappe/data_migration/doctype/data_migration_plan/data_migration_plan.js

@@ -2,7 +2,9 @@
 // For license information, please see license.txt
 
 frappe.ui.form.on('Data Migration Plan', {
-	refresh: function() {
-
+	onload(frm) {
+		frm.add_custom_button(__('Run'), () => frappe.new_doc('Data Migration Run', {
+			data_migration_plan: frm.doc.name
+		}));
 	}
 });

frappe/data_migration/doctype/data_migration_run/data_migration_run.py

@@ -6,6 +6,7 @@ from __future__ import unicode_literals
 import frappe, json, math
 from frappe.model.document import Document
 from frappe import _
+from frappe.utils import get_source_value
 
 class DataMigrationRun(Document):
 
@@ -233,7 +234,8 @@ class DataMigrationRun(Document):
 	def get_or_filters(self, mapping):
 		or_filters = self.get_last_modified_condition()
 
-		# include docs whose migration_id_field is not set
+		# docs whose migration_id_field is not set
+		# failed in the previous run, include those too
 		or_filters.update({
 			mapping.migration_id_field: ('=', '')
 		})
@@ -268,9 +270,6 @@ class DataMigrationRun(Document):
 		connection = self.get_connection()
 		data = self.get_new_local_data()
 
-		push_insert = self.get_log('push_insert', 0)
-		push_failed = self.get_log('push_failed', [])
-
 		for d in data:
 			# pre process before insert
 			doc = self.pre_process_doc(d)
@@ -282,12 +281,11 @@ class DataMigrationRun(Document):
 						mapping.migration_id_field, response_doc[connection.name_field],
 						update_modified=False)
 				frappe.db.commit()
-				self.set_log('push_insert', push_insert + 1)
+				self.update_log('push_insert', 1)
 				# post process after insert
 				self.post_process_doc(local_doc=d, remote_doc=response_doc)
 			except Exception:
-				push_failed.append(d.as_json())
-				self.set_log('push_failed', push_failed)
+				self.update_log('push_failed', d.name)
 
 		# update page_start
 		self.db_set('current_mapping_start',
@@ -308,9 +306,6 @@ class DataMigrationRun(Document):
 		connection = self.get_connection()
 		data = self.get_updated_local_data()
 
-		push_update = self.get_log('push_update', 0)
-		push_failed = self.get_log('push_failed', [])
-
 		for d in data:
 			migration_id_value = d.get(mapping.migration_id_field)
 			# pre process before update
@@ -318,12 +313,11 @@ class DataMigrationRun(Document):
 			doc = mapping.get_mapped_record(doc)
 			try:
 				response_doc = connection.update(mapping.remote_objectname, doc, migration_id_value)
-				self.set_log('push_update', push_update + 1)
+				self.update_log('push_update', 1)
 				# post process after update
 				self.post_process_doc(local_doc=d, remote_doc=response_doc)
 			except Exception:
-				push_failed.append(d.as_json())
-				self.set_log('push_failed', push_failed)
+				self.update_log('push_failed', d.name)
 
 		# update page_start
 		self.db_set('current_mapping_start',
@@ -344,9 +338,6 @@ class DataMigrationRun(Document):
 		connection = self.get_connection()
 		data = self.get_deleted_local_data()
 
-		push_delete = self.get_log('push_delete', 0)
-		push_failed = self.get_log('push_failed', [])
-
 		for d in data:
 			# Deleted Document also has a custom field for migration_id
 			migration_id_value = d.get(mapping.migration_id_field)
@@ -354,12 +345,11 @@ class DataMigrationRun(Document):
 			self.pre_process_doc(d)
 			try:
 				response_doc = connection.delete(mapping.remote_objectname, migration_id_value)
-				self.set_log('push_delete', push_delete + 1)
+				self.update_log('push_delete', 1)
 				# post process only when action is success
 				self.post_process_doc(local_doc=d, remote_doc=response_doc)
 			except Exception:
-				push_failed.append(d.as_json())
-				self.set_log('push_failed', push_failed)
+				self.update_log('push_failed', d.name)
 
 		# update page_start
 		self.db_set('current_mapping_start',
@@ -377,46 +367,32 @@ class DataMigrationRun(Document):
 		mapping = self.get_mapping(self.current_mapping)
 		data = self.get_remote_data()
 
-		pull_insert = self.get_log('pull_insert', 0)
-		pull_update = self.get_log('pull_update', 0)
-		pull_failed = self.get_log('pull_failed', [])
-
-		def get_migration_id_value(source, key):
-			value = None
-			try:
-				value = source[key]
-			except:
-				value = getattr(source, key)
-			return value
-
 		for d in data:
-			migration_id_value = get_migration_id_value(d, connection.name_field)
+			migration_id_value = get_source_value(d, connection.name_field)
 			doc = self.pre_process_doc(d)
 			doc = mapping.get_mapped_record(doc)
 
 			if migration_id_value:
-				if not local_doc_exists(mapping, migration_id_value):
-					# insert new local doc
-					local_doc = insert_local_doc(mapping, doc)
-
-					self.set_log('pull_insert', pull_insert + 1)
-					# set migration id
-					frappe.db.set_value(mapping.local_doctype, local_doc.name,
-						mapping.migration_id_field, migration_id_value,
-						update_modified=False)
-					frappe.db.commit()
-				else:
-					# update doc
-					local_doc = update_local_doc(mapping, doc, migration_id_value)
-					self.set_log('pull_update', pull_update + 1)
-
-			if local_doc:
-				# post process doc after success
-				self.post_process_doc(remote_doc=d, local_doc=local_doc)
-			else:
-				# failed, append to log
-				pull_failed.append(d)
-				self.set_log('pull_failed', pull_failed)
+				try:
+					if not local_doc_exists(mapping, migration_id_value):
+						# insert new local doc
+						local_doc = insert_local_doc(mapping, doc)
+
+						self.update_log('pull_insert', 1)
+						# set migration id
+						frappe.db.set_value(mapping.local_doctype, local_doc.name,
+							mapping.migration_id_field, migration_id_value,
+							update_modified=False)
+						frappe.db.commit()
+					else:
+						# update doc
+						local_doc = update_local_doc(mapping, doc, migration_id_value)
+						self.update_log('pull_update', 1)
+					# post process doc after success
+					self.post_process_doc(remote_doc=d, local_doc=local_doc)
+				except Exception:
+					# failed, append to log
+					self.update_log('pull_failed', migration_id_value)
 
 		if len(data) < mapping.page_length:
 			# last page, done with pull
@@ -436,6 +412,19 @@ class DataMigrationRun(Document):
 		value = json.dumps(value) if '_failed' in key else value
 		self.db_set(key, value)
 
+	def update_log(self, key, value=None):
+		'''
+		Helper for updating logs,
+		push_failed and pull_failed are stored as json,
+		other keys are stored as int
+		'''
+		if '_failed' in key:
+			# json
+			self.set_log(key, self.get_log(key, []) + [value])
+		else:
+			# int
+			self.set_log(key, self.get_log(key, 0) + (value or 1))
+
 	def get_log(self, key, default=None):
 		value = self.db_get(key)
 		if '_failed' in key:

frappe/data_migration/doctype/data_migration_run/test_data_migration_run.py

@@ -8,13 +8,13 @@ class TestDataMigrationRun(unittest.TestCase):
 	def test_run(self):
 		create_plan()
 
-		description = 'Data migration todo'
+		description = 'data migration todo'
 		new_todo = frappe.get_doc({
 			'doctype': 'ToDo',
 			'description': description
 		}).insert()
 
-		event_subject = 'Data migration event'
+		event_subject = 'data migration event'
 		frappe.get_doc(dict(
 			doctype='Event',
 			subject=event_subject,
@@ -62,7 +62,6 @@ class TestDataMigrationRun(unittest.TestCase):
 
 		# Update
 		self.assertEqual(run.db_get('status'), 'Success')
-		self.assertEqual(run.db_get('push_update'), 1)
 		self.assertEqual(run.db_get('pull_update'), 1)
 
 def create_plan():
@@ -76,7 +75,8 @@ def create_plan():
 		'fields': [
 			{ 'remote_fieldname': 'subject', 'local_fieldname': 'description' },
 			{ 'remote_fieldname': 'starts_on', 'local_fieldname': 'eval:frappe.utils.get_datetime_str(frappe.utils.get_datetime())' }
-		]
+		],
+		'condition': '{"description": "data migration todo" }'
 	}).insert()
 
 	frappe.get_doc({
@@ -87,7 +87,7 @@ def create_plan():
 		'local_doctype': 'ToDo',
 		'local_primary_key': 'name',
 		'mapping_type': 'Pull',
-		'condition': '{"subject": "Data migration event"}',
+		'condition': '{"subject": "data migration event" }',
 		'fields': [
 			{ 'remote_fieldname': 'subject', 'local_fieldname': 'description' }
 		]

frappe/email/email_body.py

@@ -420,7 +420,6 @@ def get_filecontent_from_path(path):
 
 		return filecontent
 	else:
-		print(full_path + ' doesn\'t exists')
 		return None
 
 

frappe/utils/data.py

@@ -14,6 +14,7 @@ from num2words import num2words
 from six.moves import html_parser as HTMLParser
 from six.moves.urllib.parse import quote, urljoin
 from html2text import html2text
+from markdown2 import markdown, MarkdownError
 from six import iteritems, text_type, string_types, integer_types
 
 DATE_FORMAT = "%Y-%m-%d"
@@ -844,3 +845,19 @@ def to_markdown(html):
 		pass
 
 	return text
+
+def to_html(markdown_text):
+	html = None
+	try:
+		html = markdown(markdown_text)
+	except MarkdownError:
+		pass
+
+	return html
+
+def get_source_value(source, key):
+	'''Get value from source (object or dict) based on key'''
+	if isinstance(source, dict):
+		return source.get(key)
+	else:
+		return getattr(source, key)

requirements.txt

@@ -50,4 +50,3 @@ pyqrcode
 pypng
 premailer
 psycopg2
-