refactor: AssignmentRule.apply

frappe/__init__.py

@@ -292,6 +292,7 @@ def init(site: str, sites_path: str = ".", new_site: bool = False, force=False)
  local.role_permissions = {}
  local.valid_columns = {}
  local.new_doc_templates = {}
+ local.system_notifications_queue = []
 
  local.jenv = None
  local.jloader = None

frappe/automation/doctype/assignment_rule/assignment_rule.py

@@ -62,34 +62,31 @@ def validate_assignment_days(self):
  )
  )
 
- def apply_unassign(self, doc, assignments):
+ def apply_unassign(self, doc: Document, assignments: list[frappe._dict]):
  if self.unassign_condition and self.name in [d.assignment_rule for d in assignments]:
  return self.clear_assignment(doc)
 
  return False
 
- def apply_assign(self, doc):
+ def apply_assign(self, doc: Document):
  if self.safe_eval("assign_condition", doc):
  return self.do_assignment(doc)
 
- def do_assignment(self, doc):
+ def do_assignment(self, doc: Document):
  # clear existing assignment, to reassign
- assign_to.clear(doc.get("doctype"), doc.get("name"), ignore_permissions=True)
+ assign_to.clear(doc.doctype, doc.name, ignore_permissions=True)
 
- user = self.get_user(doc)
-
- if user:
+ if user := self.get_user(doc):
  assign_to.add(
  dict(
  assign_to=[user],
- doctype=doc.get("doctype"),
- name=doc.get("name"),
- description=frappe.render_template(self.description, doc),
+ description=frappe.render_template(self.description, doc.as_dict()),
  assignment_rule=self.name,
  notify=True,
- date=doc.get(self.due_date_based_on) if self.due_date_based_on else None,
+ date=getattr(doc, self.due_date_based_on) if self.due_date_based_on else None,
  ),
  ignore_permissions=True,
+ doc=doc,
  )
 
  # set for reference in round robin
@@ -98,17 +95,15 @@ def do_assignment(self, doc):
 
  return False
 
- def clear_assignment(self, doc):
+ def clear_assignment(self, doc: Document):
  """Clear assignments"""
  if self.safe_eval("unassign_condition", doc):
- return assign_to.clear(doc.get("doctype"), doc.get("name"), ignore_permissions=True)
+ return assign_to.clear(doc.doctype, doc.name, ignore_permissions=True)
 
- def close_assignments(self, doc):
+ def close_assignments(self, doc: Document):
  """Close assignments"""
  if self.safe_eval("close_condition", doc):
- return assign_to.close_all_assignments(
- doc.get("doctype"), doc.get("name"), ignore_permissions=True
- )
+ return assign_to.close_all_assignments(doc.doctype, doc.name, ignore_permissions=True)
 
  def get_user(self, doc):
  """
@@ -165,10 +160,10 @@ def get_user_based_on_field(self, doc):
  if frappe.db.exists("User", val):
  return val
 
- def safe_eval(self, fieldname, doc):
+ def safe_eval(self, fieldname, doc: Document):
  try:
- if self.get(fieldname):
- return frappe.safe_eval(self.get(fieldname), None, doc)
+ if evaluation_condition := getattr(self, fieldname, None):
+ return frappe.safe_eval(evaluation_condition, None, doc.as_dict())
  except Exception as e:
  # when assignment fails, don't block the document as it may be
  # a part of the email pulling
@@ -232,7 +227,7 @@ def reopen_closed_assignment(doc):
  return bool(todo_list)
 
 
-def apply(doc=None, method=None, doctype=None, name=None):
+def apply(doc: Document, method: str | None = None, doctype: str | None = None, name: str | None = None):
  doctype = doctype or doc.doctype
 
  skip_assignment_rules = (
@@ -263,7 +258,6 @@ def apply(doc=None, method=None, doctype=None, name=None):
  if not assignment_rule_docs:
  return
 
- doc = doc.as_dict()
  assignments = get_assignments(doc)
 
  clear = True # are all assignments cleared

frappe/desk/doctype/notification_log/notification_log.py

@@ -88,24 +88,33 @@ def enqueue_create_notification(users: list[str] | str, doc: dict):
  users = [user.strip() for user in users.split(",") if user.strip()]
  users = list(set(users))
 
- frappe.enqueue(
- "frappe.desk.doctype.notification_log.notification_log.make_notification_logs",
- doc=doc,
- users=users,
- now=frappe.flags.in_test,
- )
+ if frappe.flags.in_test:
+ make_notification_log(doc, users)
+ else:
+ frappe.local.system_notifications_queue.append((doc, users))
+ if flush_system_notifications not in frappe.db.after_commit._functions:
+ frappe.db.after_commit(flush_system_notifications)
+
+
+def flush_system_notifications():
+ if frappe.local.system_notifications_queue:
+ frappe.enqueue(
+ "frappe.desk.doctype.notification_log.notification_log.make_notification_logs",
+ notifications=frappe.local.system_notifications_queue,
+ )
+ frappe.local.system_notifications_queue.clear()
+
+
+def make_notification_logs(notifications: list[tuple[str, list[str]]]):
+ for doc, users in notifications:
+ make_notification_log(doc, users)
 
 
-def make_notification_logs(doc, users):
+def make_notification_log(doc: dict, users: list[str]):
  for user in _get_user_ids(users):
- notification = frappe.new_doc("Notification Log")
- notification.update(doc)
- notification.for_user = user
- if (
- notification.for_user != notification.from_user
- or doc.type == "Energy Point"
- or doc.type == "Alert"
- ):
+ if user != doc.from_user or doc.type == "Energy Point" or doc.type == "Alert":
+ notification = frappe.new_doc("Notification Log").update(doc)
+ notification.for_user = user
  notification.insert(ignore_permissions=True)
 
 

frappe/desk/doctype/todo/todo.py

@@ -89,54 +89,42 @@ def update_in_reference(self):
  if not (self.reference_type and self.reference_name):
  return
 
- try:
- assignments = frappe.get_all(
- "ToDo",
- filters={
- "reference_type": self.reference_type,
- "reference_name": self.reference_name,
- "status": ("not in", ("Cancelled", "Closed")),
- "allocated_to": ("is", "set"),
- },
- pluck="allocated_to",
- )
- assignments.reverse()
-
- if frappe.get_meta(self.reference_type).issingle:
- frappe.db.set_single_value(
- self.reference_type,
- "_assign",
- json.dumps(assignments),
- update_modified=False,
- )
- else:
- frappe.db.set_value(
- self.reference_type,
- self.reference_name,
- "_assign",
- json.dumps(assignments),
- update_modified=False,
- )
-
- except Exception as e:
- if frappe.db.is_table_missing(e) and frappe.flags.in_install:
- # no table
- return
-
- elif frappe.db.is_missing_column(e):
- from frappe.database.schema import add_column
-
- add_column(self.reference_type, "_assign", "Text")
- self.update_in_reference()
+ assignments = frappe.get_all(
+ "ToDo",
+ filters={
+ "reference_type": self.reference_type,
+ "reference_name": self.reference_name,
+ "status": ("not in", ("Cancelled", "Closed")),
+ "allocated_to": ("is", "set"),
+ },
+ pluck="allocated_to",
+ order_by="creation",
+ )
 
- else:
- raise
+ if frappe.get_meta(self.reference_type).issingle:
+ frappe.db.set_single_value(
+ self.reference_type,
+ "_assign",
+ json.dumps(assignments),
+ update_modified=False,
+ )
+ else:
+ frappe.db.set_value(
+ self.reference_type,
+ self.reference_name,
+ "_assign",
+ json.dumps(assignments),
+ update_modified=False,
+ )
 
  @classmethod
  def get_owners(cls, filters=None):
- """Return list of owners after applying filters on ToDos."""
- rows = frappe.get_all(cls.DocType, filters=filters or {}, fields=["allocated_to"])
- return [parse_addr(row.allocated_to)[1] for row in rows if row.allocated_to]
+ """Returns list of owners after applying filters on todo's."""
+ return [
+ parse_addr(allocated_to)[1]
+ for allocated_to in frappe.get_all(cls.DocType, filters=filters or {}, pluck="allocated_to")
+ if allocated_to
+ ]
 
 
 # NOTE: todo is viewable if a user is an owner, or set as assigned_to value, or has any role that is allowed to access ToDo doctype.

frappe/desk/form/assign_to.py

@@ -4,6 +4,7 @@
 """assign/unassign to ToDo"""
 
 import json
+from typing import TYPE_CHECKING
 
 import frappe
 import frappe.share
@@ -15,6 +16,10 @@
  get_title_html,
 )
 from frappe.desk.form.document_follow import follow_document
+from frappe.utils import nowdate
+
+if TYPE_CHECKING:
+ from frappe.model.document import Document
 
 
 class DuplicateToDoError(frappe.ValidationError):
@@ -39,7 +44,7 @@ def get(args=None):
 
 
 @frappe.whitelist()
-def add(args=None, *, ignore_permissions=False):
+def add(args=None, *, ignore_permissions=False, doc=None):
  """add in someone's to do list
  args = {
  "assign_to": [],
@@ -55,63 +60,64 @@ def add(args=None, *, ignore_permissions=False):
 
  users_with_duplicate_todo = []
  shared_with_users = []
+ context_doc: "Document" = doc or frappe.get_cached_doc(args["doctype"], args["name"])
+ disable_document_sharing = frappe.get_system_settings("disable_document_sharing")
+
+ if not ignore_permissions:
+ context_doc.check_permission()
 
  for assign_to in frappe.parse_json(args.get("assign_to")):
  filters = {
- "reference_type": args["doctype"],
- "reference_name": args["name"],
- "status": "Open",
+ "reference_type": context_doc.doctype,
+ "reference_name": context_doc.name,
  "allocated_to": assign_to,
+ "status": "Open",
  }
- if not ignore_permissions:
- frappe.get_doc(args["doctype"], args["name"]).check_permission()
 
- if frappe.get_all("ToDo", filters=filters):
+ if frappe.db.exists("ToDo", filters):
  users_with_duplicate_todo.append(assign_to)
  else:
- from frappe.utils import nowdate
-
- if not args.get("description"):
- args["description"] = _("Assignment for {0} {1}").format(args["doctype"], args["name"])
+ todo_description = args.get("description") or _("Assignment for {0} {1}").format(
+ context_doc.doctype, context_doc.name
+ )
 
  d = frappe.get_doc(
  {
  "doctype": "ToDo",
  "allocated_to": assign_to,
- "reference_type": args["doctype"],
- "reference_name": args["name"],
- "description": args.get("description"),
+ "reference_type": context_doc.doctype,
+ "reference_name": context_doc.name,
+ "description": todo_description,
  "priority": args.get("priority", "Medium"),
  "status": "Open",
- "date": args.get("date", nowdate()),
+ "date": args.get("date") or nowdate(),
  "assigned_by": args.get("assigned_by", frappe.session.user),
  "assignment_rule": args.get("assignment_rule"),
  }
  ).insert(ignore_permissions=True)
 
  # set assigned_to if field exists
- if frappe.get_meta(args["doctype"]).get_field("assigned_to"):
- frappe.db.set_value(args["doctype"], args["name"], "assigned_to", assign_to)
-
- doc = frappe.get_doc(args["doctype"], args["name"])
+ if context_doc.meta.get_field("assigned_to"):
+ frappe.db.set_value(context_doc.doctype, context_doc.name, "assigned_to", assign_to)
 
  # if assignee does not have permissions, share or inform
- if not frappe.has_permission(doc=doc, user=assign_to):
- if frappe.get_system_settings("disable_document_sharing"):
- msg = _("User {0} is not permitted to access this document.").format(
- frappe.bold(assign_to)
- )
- msg += "<br>" + _(
- "As document sharing is disabled, please give them the required permissions before assigning."
+ if not context_doc.has_permission(user=assign_to):
+ if disable_document_sharing:
+ msg = (
+ _("User {0} is not permitted to access this document.").format(frappe.bold(assign_to))
+ + "<br>"
+ + _(
+ "As document sharing is disabled, please give them the required permissions before assigning."
+ )
  )
  frappe.throw(msg, title=_("Missing Permission"))
- else:
-  frappe.share.add(doc.doctype, doc.name, assign_to)
-  shared_with_users.append(assign_to)
+
+ frappe.share.add(context_doc.doctype, context_doc.name, assign_to)
+ shared_with_users.append(assign_to)
 
  # make this document followed by assigned user
  if frappe.get_cached_value("User", assign_to, "follow_assigned_documents"):
- follow_document(args["doctype"], args["name"], assign_to)
+ follow_document(context_doc.doctype, context_doc.name, assign_to)
 
  # notify
  notify_assignment(
@@ -133,7 +139,7 @@ def add(args=None, *, ignore_permissions=False):
  user_list = format_message_for_assign_to(users_with_duplicate_todo)
  frappe.msgprint(_("Already in the following Users ToDo list:{0}").format(user_list, alert=True))
 
- return get(args)
+ return get(args | {"doctype": context_doc.doctype, "name": context_doc.name})
 
 
 @frappe.whitelist()
@@ -223,7 +229,7 @@ def set_status(doctype, name, todo=None, assign_to=None, status="Cancelled", ign
  pass
 
  # clear assigned_to if field exists
- if frappe.get_meta(doctype).get_field("assigned_to") and status in ("Cancelled", "Closed"):
+ if status in ("Cancelled", "Closed") and frappe.get_meta(doctype).get_field("assigned_to"):
  frappe.db.set_value(doctype, name, "assigned_to", None)
 
  return get({"doctype": doctype, "name": name})
@@ -261,7 +267,7 @@ def notify_assignment(assigned_by, allocated_to, doc_type, doc_name, action="CLO
  if not (assigned_by and allocated_to and doc_type and doc_name):
  return
 
- assigned_user = frappe.db.get_value("User", allocated_to, ["language", "enabled"], as_dict=True)
+ assigned_user = frappe.get_cached_value("User", allocated_to, ["language", "enabled"], as_dict=True)
 
  # return if self assigned or user disabled
  if assigned_by == allocated_to or not assigned_user.enabled: