Hello,
I use bulk_insert to insert many records
def _document_values_generator(self,
documents: Iterable["Document"],
columns: list[str],
) -> Generator[tuple[Any], None, None]:
for doc in documents:
doc.creation = doc.modified = frappe.utils.get_datetime()
doc.owner = doc.created_by = doc.modified_by = frappe.session.user
doc_values = doc.get_valid_dict(
convert_dates_to_str=True,
ignore_nulls=True,
ignore_virtual=True,
)
yield tuple(doc_values.get(col) for col in columns)
def bulk_insert(self,
doctype: str,
documents: Iterable["Document"],
ignore_duplicates: bool = False,
chunk_size=10_000):
columns = frappe.get_meta(doctype).get_valid_columns()
values = self._document_values_generator(documents, columns)
frappe.db.bulk_insert(
doctype, columns, values, ignore_duplicates=ignore_duplicates, chunk_size=chunk_size
)
when i have name field it work but when I dont have name . document without field name or name is assigned with autoname return error:
File "env/lib/python3.10/site-packages/pymysql/err.py", line 143, in raise_mysql_exception raise errorclass(errno, errval) pymysql.err.IntegrityError: (1048, "Column 'name' cannot be null")
I know that name is a primary key and will be unique but with insert one record use
if not doc.name and (meta.autoname or "").lower() != "prompt":
# name can only be set directly if autoname is prompt
new_doc.set("name", None)
the whole function to insert one record
meta = frappe.get_meta(self.doctype)
new_doc = frappe.new_doc(self.doctype)
new_doc.update(doc)
if not doc.name and (meta.autoname or "").lower() != "prompt":
# name can only be set directly if autoname is prompt
new_doc.set("name", None)
new_doc.flags.updater_reference = {
"doctype": self.data_import.doctype,
"docname": self.data_import.name,
"label": _("via Data Import"),
}
new_doc.insert()
if meta.is_submittable and self.data_import.submit_after_import:
new_doc.submit()
i optimise the function to use same thing like with one record
def _document_values_generator(self,
documents: Iterable["Document"],
columns: list[str],
) -> Generator[tuple[Any], None, None]:
for doc in documents:
meta = frappe.get_meta(self.doctype)
doc.creation = doc.modified = frappe.utils.get_datetime()
doc.owner = doc.created_by = doc.modified_by = frappe.session.user
doc_values = doc.get_valid_dict(
convert_dates_to_str=True,
ignore_nulls=True,
ignore_virtual=True,
)
new_doc = frappe.new_doc(self.doctype)
new_doc.update(doc_values)
if not doc.get('name') and (meta.autoname or "").lower() != "prompt":
new_doc.set("name", None)
new_doc.flags.updater_reference = {
"doctype": self.data_import.doctype,
"docname": self.data_import.name,
"label": _("via Data Import"),
}
yield tuple(new_doc.get(col) for col in columns)
def bulk_insert(self,
doctype: str,
documents: Iterable["Document"],
ignore_duplicates: bool = False,
chunk_size=10_000):
columns = frappe.get_meta(doctype).get_valid_columns()
values = self._document_values_generator(documents, columns)
meta = frappe.get_meta(self.doctype)
doc_list = []
for value in values:
#doc_values = {col: val for col, val in zip(columns, value)} # Creating a dictionary for the row
doc_list.append(value)
test = tuple(doc_list)
first_tuple_data = next(values, None) # Get the first tuple or None if empty
if first_tuple_data:
frappe.throw(f"First tuple data: {first_tuple_data}")
frappe.db.bulk_insert(
doctype, columns, doc_list, ignore_duplicates=ignore_duplicates, chunk_size=chunk_size
)
frappe.db.commit()