0003-misc-use-a-single-sql-query-to-get-item-ids-65511.patch

Frédéric Péters, 30 mai 2022 09:53

Voir les différences: en ligne côte à côte

Subject: [PATCH 3/4] misc: use a single sql query to get item ids (#65511)

 wcs/formdata.py         |  12 +++--
 wcs/forms/backoffice.py | 108 ++++++++++++++++++++++++++++------------
 2 files changed, 85 insertions(+), 35 deletions(-)

     from .qommon import _, misc
     from .qommon.evalutils import make_datetime
     from .qommon.publisher import get_cfg
     from .qommon.storage import Contains, Intersects, Null, StorableObject
     from .qommon.storage import And, Contains, Intersects, Null, StorableObject
     from .qommon.substitution import CompatibilityNamesDict, Substitutions, invalidate_substitution_cache
     from .qommon.template import Template
-...
                 return len(cls.get_actionable_ids(user_roles))
         @classmethod
         def get_actionable_ids(cls, user_roles):
         def get_actionable_ids_criteria(cls, user_roles):
             statuses = ['wf-%s' % x.id for x in cls._formdef.workflow.get_not_endpoint_status()]
             return And([Intersects('actions_roles_array', user_roles), Contains('status', statuses)])
         @classmethod
         def get_actionable_ids(cls, user_roles):
             if get_publisher().is_using_postgresql():
                 criterias = [Intersects('actions_roles_array', user_roles), Contains('status', statuses)]
                 return cls.keys(criterias)
                 return cls.keys([cls.get_actionable_ids_criteria(user_roles)])
             else:
                 statuses = ['wf-%s' % x.id for x in cls._formdef.workflow.get_not_endpoint_status()]
                 actions_ids = set()
                 for role in user_roles:
                     actions_ids |= set(cls.get_ids_with_indexed_value('actions_roles', str(role)))

     from quixote import get_publisher, get_request, get_session, redirect
     from quixote.html import TemplateIO, htmltext
     from wcs.qommon.storage import Contains, FtsMatch, NotContains, Null, StrictNotEqual
     from wcs.qommon.storage import Contains, FtsMatch, Intersects, Not, NotContains, Null, StrictNotEqual
     from wcs.roles import logged_users_role
     from ..qommon import _, misc
-...
             criterias=None,
             anonymise=False,
         ):
             def get_all_except_drafts(**clause_kwargs):
                 if get_publisher().is_using_postgresql():
                     return formdata_class.keys(**clause_kwargs)
             if get_publisher().is_using_postgresql():
                 return self.get_listing_item_ids_sql(
                     selected_filter, selected_filter_operator, query, order_by, user, criterias, anonymise
+                )
             def get_all_except_drafts():
                 item_ids = formdata_class.keys()
                 drafts = formdata_class.get_ids_with_indexed_value('status', 'draft')
                 return [x for x in item_ids if x not in drafts]
             formdata_class = self.formdef.data_class()
             # used for postgresql mode only
             clause_kwargs = {'clause': criterias + [StrictNotEqual('status', 'draft')]}
             if selected_filter == 'all':
                 if selected_filter_operator == 'ne':
                     # nothing
                     item_ids = []
                 else:
                     # all except drafts
                     item_ids = get_all_except_drafts(**clause_kwargs)
                     item_ids = get_all_except_drafts()
             elif selected_filter == 'waiting':
                 user_roles = [logged_users_role().id] + user.get_roles()
                 waiting_item_ids = formdata_class.get_actionable_ids(user_roles)
                 if selected_filter_operator == 'ne':
                     # select all ids except drafts
                     item_ids = get_all_except_drafts(**clause_kwargs)
                     item_ids = get_all_except_drafts()
                     # exclude waiting ids
                     item_ids = [x for x in item_ids if x not in waiting_item_ids]
                 else:
-...
                 else:
                     applied_filters = ['wf-%s' % selected_filter]
                 if get_publisher().is_using_postgresql():
                     if selected_filter_operator == 'ne':
                         # exclude selected status list
                         clause_kwargs['clause'].append(NotContains('status', applied_filters))
                     else:
                         # only selected status list
                         clause_kwargs['clause'].append(Contains('status', applied_filters))
                     item_ids = formdata_class.keys(**clause_kwargs)
                 else:
                     filtered_item_ids = []
                     for status_id in applied_filters:
                         filtered_item_ids.extend(
                             formdata_class.get_ids_with_indexed_value(
                                 'status',
                                 status_id,
+                            )
                 filtered_item_ids = []
                 for status_id in applied_filters:
                     filtered_item_ids.extend(
                         formdata_class.get_ids_with_indexed_value(
                             'status',
                             status_id,
+                        )
                     if selected_filter_operator == 'ne':
                         # select all ids except drafts
                         item_ids = get_all_except_drafts()
                         # exclude selected status list
                         item_ids = [x for x in item_ids if x not in filtered_item_ids]
                     else:
                         # only selected status list
                         item_ids = filtered_item_ids
+                    )
                 if selected_filter_operator == 'ne':
                     # select all ids except drafts
                     item_ids = get_all_except_drafts()
                     # exclude selected status list
                     item_ids = [x for x in item_ids if x not in filtered_item_ids]
                 else:
                     # only selected status list
                     item_ids = filtered_item_ids
             if query:
                 query_ids = formdata_class.get_ids_from_query(query)
-...
             return item_ids
         def get_listing_item_ids_sql(
             self,
             selected_filter,
             selected_filter_operator,
             query,
             order_by,
             user,
             criterias,
             anonymise,
         ):
             formdata_class = self.formdef.data_class()
             criterias = [] or criterias[:]
             criterias.append(StrictNotEqual('status', 'draft'))
             if selected_filter == 'all':
                 if selected_filter_operator == 'ne':
                     # nothing
                     return []
             elif selected_filter == 'waiting':
                 user_roles = [logged_users_role().id] + user.get_roles()
                 actionable_criteria = formdata_class.get_actionable_ids_criteria(user_roles)
                 if selected_filter_operator == 'ne':
                     criterias.append(Not(actionable_criteria))
                 else:
                     criterias.append(actionable_criteria)
             else:
                 # build selected status list
                 applied_filters = []
                 if selected_filter == 'pending':
                     applied_filters = ['wf-%s' % x.id for x in self.formdef.workflow.get_not_endpoint_status()]
                 elif selected_filter == 'done':
                     applied_filters = ['wf-%s' % x.id for x in self.formdef.workflow.get_endpoint_status()]
                 else:
                     applied_filters = ['wf-%s' % selected_filter]
                 if selected_filter_operator == 'ne':
                     # exclude selected status list
                     criterias.append(NotContains('status', applied_filters))
                 else:
                     # only selected status list
                     criterias.append(Contains('status', applied_filters))
             if query:
                 criterias.append(FtsMatch(query))
             if not anonymise:
                 # as we are in the backoffice, we don't have to care about the
                 # situation where the user is the submitter, and we limit ourselves
                 # to consider treating roles.
                 if not user.is_admin:
                     user_roles = [str(x) for x in user.get_roles()]
                     criterias.append(Intersects('concerned_roles_array', user_roles))
             return list(formdata_class.keys(criterias))
         def get_listing_items(
             self,
             fields=None,
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0003-misc-use-a-single-sql-query-to-get-item-ids-65511.patch