diff options
Diffstat (limited to 'odoo/workflow')
| -rw-r--r-- | odoo/workflow/__init__.py | 83 | ||||
| -rw-r--r-- | odoo/workflow/helpers.py | 21 | ||||
| -rw-r--r-- | odoo/workflow/instance.py | 117 | ||||
| -rw-r--r-- | odoo/workflow/service.py | 96 | ||||
| -rw-r--r-- | odoo/workflow/workitem.py | 314 |
5 files changed, 631 insertions, 0 deletions
diff --git a/odoo/workflow/__init__.py b/odoo/workflow/__init__.py new file mode 100644 index 00000000000..752f7e54bea --- /dev/null +++ b/odoo/workflow/__init__.py @@ -0,0 +1,83 @@ +# -*- coding: utf-8 -*- +# Part of Odoo. See LICENSE file for full copyright and licensing details. + +from openerp.workflow.service import WorkflowService + +# The new API is in openerp.workflow.workflow_service +# OLD API of the Workflow + +def clear_cache(cr, uid): + WorkflowService.clear_cache(cr.dbname) + +def trg_write(uid, res_type, res_id, cr): + """ + Reevaluates the specified workflow instance. Thus if any condition for + a transition have been changed in the backend, then running ``trg_write`` + will move the workflow over that transition. + + :param res_type: the model name + :param res_id: the model instance id the workflow belongs to + :param cr: a database cursor + """ + return WorkflowService.new(cr, uid, res_type, res_id).write() + +def trg_trigger(uid, res_type, res_id, cr): + """ + Activate a trigger. + + If a workflow instance is waiting for a trigger from another model, then this + trigger can be activated if its conditions are met. + + :param res_type: the model name + :param res_id: the model instance id the workflow belongs to + :param cr: a database cursor + """ + return WorkflowService.new(cr, uid, res_type, res_id).trigger() + +def trg_delete(uid, res_type, res_id, cr): + """ + Delete a workflow instance + + :param res_type: the model name + :param res_id: the model instance id the workflow belongs to + :param cr: a database cursor + """ + return WorkflowService.new(cr, uid, res_type, res_id).delete() + +def trg_create(uid, res_type, res_id, cr): + """ + Create a new workflow instance + + :param res_type: the model name + :param res_id: the model instance id to own the created worfklow instance + :param cr: a database cursor + """ + return WorkflowService.new(cr, uid, res_type, res_id).create() + +def trg_validate(uid, res_type, res_id, signal, cr): + """ + Fire a signal on a given workflow instance + + :param res_type: the model name + :param res_id: the model instance id the workflow belongs to + :signal: the signal name to be fired + :param cr: a database cursor + """ + assert isinstance(signal, basestring) + return WorkflowService.new(cr, uid, res_type, res_id).validate(signal) + +def trg_redirect(uid, res_type, res_id, new_rid, cr): + """ + Re-bind a workflow instance to another instance of the same model. + + Make all workitems which are waiting for a (subflow) workflow instance + for the old resource point to the (first active) workflow instance for + the new resource. + + :param res_type: the model name + :param res_id: the model instance id the workflow belongs to + :param new_rid: the model instance id to own the worfklow instance + :param cr: a database cursor + """ + assert isinstance(new_rid, (long, int)) + return WorkflowService.new(cr, uid, res_type, res_id).redirect(new_rid) diff --git a/odoo/workflow/helpers.py b/odoo/workflow/helpers.py new file mode 100644 index 00000000000..8350259c452 --- /dev/null +++ b/odoo/workflow/helpers.py @@ -0,0 +1,21 @@ +import openerp.sql_db + +class Session(object): + def __init__(self, cr, uid): + assert isinstance(cr, openerp.sql_db.Cursor) + assert isinstance(uid, (int, long)) + self.cr = cr + self.uid = uid + +class Record(object): + def __init__(self, model, record_id): + assert isinstance(model, basestring) + assert isinstance(record_id, (int, long)) + self.model = model + self.id = record_id + +class WorkflowActivity(object): + KIND_FUNCTION = 'function' + KIND_DUMMY = 'dummy' + KIND_STOPALL = 'stopall' + KIND_SUBFLOW = 'subflow' diff --git a/odoo/workflow/instance.py b/odoo/workflow/instance.py new file mode 100644 index 00000000000..7baf6fddf88 --- /dev/null +++ b/odoo/workflow/instance.py @@ -0,0 +1,117 @@ +# -*- coding: utf-8 -*- +# Part of Odoo. See LICENSE file for full copyright and licensing details. +import workitem +from openerp.workflow.helpers import Session +from openerp.workflow.helpers import Record +from openerp.workflow.workitem import WorkflowItem + +class WorkflowInstance(object): + def __init__(self, session, record, values): + assert isinstance(session, Session) + assert isinstance(record, Record) + self.session = session + self.record = record + + if not values: + values = {} + + assert isinstance(values, dict) + self.instance = values + + @classmethod + def create(cls, session, record, workflow_id): + assert isinstance(session, Session) + assert isinstance(record, Record) + assert isinstance(workflow_id, (int, long)) + + cr = session.cr + cr.execute('insert into wkf_instance (res_type,res_id,uid,wkf_id,state) values (%s,%s,%s,%s,%s) RETURNING id', (record.model, record.id, session.uid, workflow_id, 'active')) + instance_id = cr.fetchone()[0] + + cr.execute('select * from wkf_activity where flow_start=True and wkf_id=%s', (workflow_id,)) + stack = [] + + activities = cr.dictfetchall() + for activity in activities: + WorkflowItem.create(session, record, activity, instance_id, stack) + + cr.execute('SELECT * FROM wkf_instance WHERE id = %s', (instance_id,)) + values = cr.dictfetchone() + wi = WorkflowInstance(session, record, values) + wi.update() + + return wi + + def delete(self): + self.session.cr.execute('delete from wkf_instance where res_id=%s and res_type=%s', (self.record.id, self.record.model)) + + def validate(self, signal, force_running=False): + assert isinstance(signal, basestring) + assert isinstance(force_running, bool) + + cr = self.session.cr + cr.execute("select * from wkf_workitem where inst_id=%s", (self.instance['id'],)) + stack = [] + for i, work_item_values in enumerate(cr.dictfetchall()): + if i > 0: + # test if previous workitem has already processed this one + cr.execute("select id from wkf_workitem where id=%s", (work_item_values['id'],)) + if not cr.fetchone(): + continue + wi = WorkflowItem(self.session, self.record, work_item_values) + wi.process(signal=signal, force_running=force_running, stack=stack) + # An action is returned + self._update_end() + return stack and stack[0] or False + + def update(self): + cr = self.session.cr + + cr.execute("select * from wkf_workitem where inst_id=%s", (self.instance['id'],)) + for work_item_values in cr.dictfetchall(): + stack = [] + WorkflowItem(self.session, self.record, work_item_values).process(stack=stack) + return self._update_end() + + def _update_end(self): + cr = self.session.cr + instance_id = self.instance['id'] + cr.execute('select wkf_id from wkf_instance where id=%s', (instance_id,)) + wkf_id = cr.fetchone()[0] + cr.execute('select state,flow_stop from wkf_workitem w left join wkf_activity a on (a.id=w.act_id) where w.inst_id=%s', (instance_id,)) + ok=True + for r in cr.fetchall(): + if (r[0]<>'complete') or not r[1]: + ok=False + break + if ok: + cr.execute('select distinct a.name from wkf_activity a left join wkf_workitem w on (a.id=w.act_id) where w.inst_id=%s', (instance_id,)) + act_names = cr.fetchall() + cr.execute("update wkf_instance set state='complete' where id=%s", (instance_id,)) + cr.execute("update wkf_workitem set state='complete' where subflow_id=%s", (instance_id,)) + cr.execute("select i.id,w.osv,i.res_id from wkf_instance i left join wkf w on (i.wkf_id=w.id) where i.id IN (select inst_id from wkf_workitem where subflow_id=%s)", (instance_id,)) + for cur_instance_id, cur_model_name, cur_record_id in cr.fetchall(): + cur_record = Record(cur_model_name, cur_record_id) + for act_name in act_names: + WorkflowInstance(self.session, cur_record, {'id':cur_instance_id}).validate('subflow.%s' % act_name[0]) + + return ok + + + + + +def create(session, record, workflow_id): + return WorkflowInstance(session, record).create(workflow_id) + +def delete(session, record): + return WorkflowInstance(session, record).delete() + +def validate(session, record, instance_id, signal, force_running=False): + return WorkflowInstance(session, record).validate(instance_id, signal, force_running) + +def update(session, record, instance_id): + return WorkflowInstance(session, record).update(instance_id) + +def _update_end(session, record, instance_id): + return WorkflowInstance(session, record)._update_end(instance_id) diff --git a/odoo/workflow/service.py b/odoo/workflow/service.py new file mode 100644 index 00000000000..ea64f8103b6 --- /dev/null +++ b/odoo/workflow/service.py @@ -0,0 +1,96 @@ +# -*- coding: utf-8 -*- +# Part of Odoo. See LICENSE file for full copyright and licensing details. +from helpers import Session +from helpers import Record + +from openerp.workflow.instance import WorkflowInstance +# import instance + + +class WorkflowService(object): + CACHE = {} + + @classmethod + def clear_cache(cls, dbname): + cls.CACHE[dbname] = {} + + @classmethod + def new(cls, cr, uid, model_name, record_id): + return cls(Session(cr, uid), Record(model_name, record_id)) + + def __init__(self, session, record): + assert isinstance(session, Session) + assert isinstance(record, Record) + + self.session = session + self.record = record + + self.cr = self.session.cr + + def write(self): + self.cr.execute('select id from wkf_instance where res_id=%s and res_type=%s and state=%s', + (self.record.id or None, self.record.model or None, 'active') + ) + for (instance_id,) in self.cr.fetchall(): + WorkflowInstance(self.session, self.record, {'id': instance_id}).update() + + def trigger(self): + self.cr.execute('select instance_id from wkf_triggers where res_id=%s and model=%s', (self.record.id, self.record.model)) + res = self.cr.fetchall() + for (instance_id,) in res: + self.cr.execute('select %s,res_type,res_id from wkf_instance where id=%s', (self.session.uid, instance_id,)) + current_uid, current_model_name, current_record_id = self.cr.fetchone() + + current_session = Session(self.session.cr, current_uid) + current_record = Record(current_model_name, current_record_id) + + WorkflowInstance(current_session, current_record, {'id': instance_id}).update() + + def delete(self): + WorkflowInstance(self.session, self.record, {}).delete() + + def create(self): + WorkflowService.CACHE.setdefault(self.cr.dbname, {}) + + wkf_ids = WorkflowService.CACHE[self.cr.dbname].get(self.record.model, None) + + if not wkf_ids: + self.cr.execute('select id from wkf where osv=%s and on_create=True', (self.record.model,)) + wkf_ids = self.cr.fetchall() + WorkflowService.CACHE[self.cr.dbname][self.record.model] = wkf_ids + + for (wkf_id, ) in wkf_ids: + WorkflowInstance.create(self.session, self.record, wkf_id) + + def validate(self, signal): + result = False + # ids of all active workflow instances for a corresponding resource (id, model_nam) + self.cr.execute('select id from wkf_instance where res_id=%s and res_type=%s and state=%s', (self.record.id, self.record.model, 'active')) + # TODO: Refactor the workflow instance object + for (instance_id,) in self.cr.fetchall(): + wi = WorkflowInstance(self.session, self.record, {'id': instance_id}) + + res2 = wi.validate(signal) + + result = result or res2 + return result + + def redirect(self, new_rid): + # get ids of wkf instances for the old resource (res_id) + # CHECKME: shouldn't we get only active instances? + self.cr.execute('select id, wkf_id from wkf_instance where res_id=%s and res_type=%s', (self.record.id, self.record.model)) + + for old_inst_id, workflow_id in self.cr.fetchall(): + # first active instance for new resource (new_rid), using same wkf + self.cr.execute( + 'SELECT id '\ + 'FROM wkf_instance '\ + 'WHERE res_id=%s AND res_type=%s AND wkf_id=%s AND state=%s', + (new_rid, self.record.model, workflow_id, 'active')) + new_id = self.cr.fetchone() + if new_id: + # select all workitems which "wait" for the old instance + self.cr.execute('select id from wkf_workitem where subflow_id=%s', (old_inst_id,)) + for (item_id,) in self.cr.fetchall(): + # redirect all those workitems to the wkf instance of the new resource + self.cr.execute('update wkf_workitem set subflow_id=%s where id=%s', (new_id[0], item_id)) diff --git a/odoo/workflow/workitem.py b/odoo/workflow/workitem.py new file mode 100644 index 00000000000..8affdd7a8a1 --- /dev/null +++ b/odoo/workflow/workitem.py @@ -0,0 +1,314 @@ + +# Part of Odoo. See LICENSE file for full copyright and licensing details. + +# +# TODO: +# cr.execute('delete from wkf_triggers where model=%s and res_id=%s', (res_type,res_id)) +# +import logging +import instance + +from openerp.workflow.helpers import Session +from openerp.workflow.helpers import Record +from openerp.workflow.helpers import WorkflowActivity + +logger = logging.getLogger(__name__) + +import openerp +from openerp.tools.safe_eval import safe_eval + +class Environment(dict): + """ + Dictionary class used as an environment to evaluate workflow code (such as + the condition on transitions). + + This environment provides sybmols for cr, uid, id, model name, model + instance, column names, and all the record (the one obtained by browsing + the provided ID) attributes. + """ + def __init__(self, session, record): + self.cr = session.cr + self.uid = session.uid + self.model = record.model + self.id = record.id + self.ids = [record.id] + + def __getitem__(self, key): + env = openerp.api.Environment(self.cr, self.uid, {}) + records = env[self.model].browse(self.ids) + if hasattr(records, key): + return getattr(records, key) + else: + return super(Environment, self).__getitem__(key) + + +class WorkflowItem(object): + def __init__(self, session, record, work_item_values): + assert isinstance(session, Session) + assert isinstance(record, Record) + self.session = session + self.record = record + + if not work_item_values: + work_item_values = {} + + assert isinstance(work_item_values, dict) + self.workitem = work_item_values + + @classmethod + def create(cls, session, record, activity, instance_id, stack): + assert isinstance(session, Session) + assert isinstance(record, Record) + assert isinstance(activity, dict) + assert isinstance(instance_id, (long, int)) + assert isinstance(stack, list) + + cr = session.cr + cr.execute("select nextval('wkf_workitem_id_seq')") + id_new = cr.fetchone()[0] + cr.execute("insert into wkf_workitem (id,act_id,inst_id,state) values (%s,%s,%s,'active')", (id_new, activity['id'], instance_id)) + cr.execute('select * from wkf_workitem where id=%s',(id_new,)) + work_item_values = cr.dictfetchone() + logger.info('Created workflow item in activity %s', + activity['id'], + extra={'ident': (session.uid, record.model, record.id)}) + + workflow_item = WorkflowItem(session, record, work_item_values) + workflow_item.process(stack=stack) + + @classmethod + def create_all(cls, session, record, activities, instance_id, stack): + assert isinstance(activities, list) + + for activity in activities: + cls.create(session, record, activity, instance_id, stack) + + def process(self, signal=None, force_running=False, stack=None): + assert isinstance(force_running, bool) + assert stack is not None + + cr = self.session.cr + + cr.execute('select * from wkf_activity where id=%s', (self.workitem['act_id'],)) + activity = cr.dictfetchone() + + triggers = False + if self.workitem['state'] == 'active': + triggers = True + if not self._execute(activity, stack): + return False + + if force_running or self.workitem['state'] == 'complete': + ok = self._split_test(activity['split_mode'], signal, stack) + triggers = triggers and not ok + + if triggers: + cr.execute('select * from wkf_transition where act_from=%s ORDER BY sequence,id', (self.workitem['act_id'],)) + for trans in cr.dictfetchall(): + if trans['trigger_model']: + ids = self.wkf_expr_eval_expr(trans['trigger_expr_id']) + for res_id in ids: + cr.execute('select nextval(\'wkf_triggers_id_seq\')') + id =cr.fetchone()[0] + cr.execute('insert into wkf_triggers (model,res_id,instance_id,workitem_id,id) values (%s,%s,%s,%s,%s)', (trans['trigger_model'],res_id, self.workitem['inst_id'], self.workitem['id'], id)) + + return True + + def _execute(self, activity, stack): + """Send a signal to parenrt workflow (signal: subflow.signal_name)""" + result = True + cr = self.session.cr + signal_todo = [] + + if (self.workitem['state']=='active') and activity['signal_send']: + # signal_send']: + cr.execute("select i.id,w.osv,i.res_id from wkf_instance i left join wkf w on (i.wkf_id=w.id) where i.id IN (select inst_id from wkf_workitem where subflow_id=%s)", (self.workitem['inst_id'],)) + for instance_id, model_name, record_id in cr.fetchall(): + record = Record(model_name, record_id) + signal_todo.append((instance_id, record, activity['signal_send'])) + + + if activity['kind'] == WorkflowActivity.KIND_DUMMY: + if self.workitem['state']=='active': + self._state_set(activity, 'complete') + if activity['action_id']: + res2 = self.wkf_expr_execute_action(activity) + if res2: + stack.append(res2) + result=res2 + + elif activity['kind'] == WorkflowActivity.KIND_FUNCTION: + + if self.workitem['state']=='active': + self._state_set(activity, 'running') + returned_action = self.wkf_expr_execute(activity) + if type(returned_action) in (dict,): + stack.append(returned_action) + if activity['action_id']: + res2 = self.wkf_expr_execute_action(activity) + # A client action has been returned + if res2: + stack.append(res2) + result=res2 + self._state_set(activity, 'complete') + + elif activity['kind'] == WorkflowActivity.KIND_STOPALL: + if self.workitem['state']=='active': + self._state_set(activity, 'running') + cr.execute('delete from wkf_workitem where inst_id=%s and id<>%s', (self.workitem['inst_id'], self.workitem['id'])) + if activity['action']: + self.wkf_expr_execute(activity) + self._state_set(activity, 'complete') + + elif activity['kind'] == WorkflowActivity.KIND_SUBFLOW: + + if self.workitem['state']=='active': + + self._state_set(activity, 'running') + if activity.get('action', False): + id_new = self.wkf_expr_execute(activity) + if not id_new: + cr.execute('delete from wkf_workitem where id=%s', (self.workitem['id'],)) + return False + assert type(id_new)==type(1) or type(id_new)==type(1L), 'Wrong return value: '+str(id_new)+' '+str(type(id_new)) + cr.execute('select id from wkf_instance where res_id=%s and wkf_id=%s', (id_new, activity['subflow_id'])) + id_new = cr.fetchone()[0] + else: + inst = instance.WorkflowInstance(self.session, self.record) + id_new = inst.create(activity['subflow_id']) + + cr.execute('update wkf_workitem set subflow_id=%s where id=%s', (id_new, self.workitem['id'])) + self.workitem['subflow_id'] = id_new + + if self.workitem['state']=='running': + cr.execute("select state from wkf_instance where id=%s", (self.workitem['subflow_id'],)) + state = cr.fetchone()[0] + if state=='complete': + self._state_set(activity, 'complete') + + for instance_id, record, signal_send in signal_todo: + wi = instance.WorkflowInstance(self.session, record, {'id': instance_id}) + wi.validate(signal_send, force_running=True) + + return result + + def _state_set(self, activity, state): + self.session.cr.execute('update wkf_workitem set state=%s where id=%s', (state, self.workitem['id'])) + self.workitem['state'] = state + logger.info('Changed state of work item %s to "%s" in activity %s', + self.workitem['id'], state, activity['id'], + extra={'ident': (self.session.uid, self.record.model, self.record.id)}) + + def _split_test(self, split_mode, signal, stack): + cr = self.session.cr + cr.execute('select * from wkf_transition where act_from=%s ORDER BY sequence,id', (self.workitem['act_id'],)) + test = False + transitions = [] + alltrans = cr.dictfetchall() + + if split_mode in ('XOR', 'OR'): + for transition in alltrans: + if self.wkf_expr_check(transition,signal): + test = True + transitions.append((transition['id'], self.workitem['inst_id'])) + if split_mode=='XOR': + break + else: + test = True + for transition in alltrans: + if not self.wkf_expr_check(transition, signal): + test = False + break + cr.execute('select count(*) from wkf_witm_trans where trans_id=%s and inst_id=%s', (transition['id'], self.workitem['inst_id'])) + if not cr.fetchone()[0]: + transitions.append((transition['id'], self.workitem['inst_id'])) + + if test and transitions: + cr.executemany('insert into wkf_witm_trans (trans_id,inst_id) values (%s,%s)', transitions) + cr.execute('delete from wkf_workitem where id=%s', (self.workitem['id'],)) + for t in transitions: + self._join_test(t[0], t[1], stack) + return True + return False + + def _join_test(self, trans_id, inst_id, stack): + cr = self.session.cr + cr.execute('select * from wkf_activity where id=(select act_to from wkf_transition where id=%s)', (trans_id,)) + activity = cr.dictfetchone() + if activity['join_mode']=='XOR': + WorkflowItem.create(self.session, self.record, activity, inst_id, stack=stack) + cr.execute('delete from wkf_witm_trans where inst_id=%s and trans_id=%s', (inst_id,trans_id)) + else: + cr.execute('select id from wkf_transition where act_to=%s ORDER BY sequence,id', (activity['id'],)) + trans_ids = cr.fetchall() + ok = True + for (id,) in trans_ids: + cr.execute('select count(*) from wkf_witm_trans where trans_id=%s and inst_id=%s', (id,inst_id)) + res = cr.fetchone()[0] + if not res: + ok = False + break + if ok: + for (id,) in trans_ids: + cr.execute('delete from wkf_witm_trans where trans_id=%s and inst_id=%s', (id,inst_id)) + WorkflowItem.create(self.session, self.record, activity, inst_id, stack=stack) + + def wkf_expr_eval_expr(self, lines): + """ + Evaluate each line of ``lines`` with the ``Environment`` environment, returning + the value of the last line. + """ + assert lines, 'You used a NULL action in a workflow, use dummy node instead.' + result = False + for line in lines.split('\n'): + line = line.strip() + if not line: + continue + if line == 'True': + result = True + elif line == 'False': + result = False + else: + env = Environment(self.session, self.record) + result = safe_eval(line, env, nocopy=True) + return result + + def wkf_expr_execute_action(self, activity): + """ + Evaluate the ir.actions.server action specified in the activity. + """ + context = { + 'active_model': self.record.model, + 'active_id': self.record.id, + 'active_ids': [self.record.id] + } + env = openerp.api.Environment(self.session.cr, self.session.uid, context) + action = env['ir.actions.server'].browse(activity['action_id']) + result = action.run() + return result + + def wkf_expr_execute(self, activity): + """ + Evaluate the action specified in the activity. + """ + return self.wkf_expr_eval_expr(activity['action']) + + def wkf_expr_check(self, transition, signal): + """ + Test if a transition can be taken. The transition can be taken if: + + - the signal name matches, + - the uid is SUPERUSER_ID or the user groups contains the transition's + group, + - the condition evaluates to a truish value. + """ + if transition['signal'] and signal != transition['signal']: + return False + + if self.session.uid != openerp.SUPERUSER_ID and transition['group_id']: + env = openerp.api.Environment(self.session.cr, self.session.uid, {}) + user_groups = env.user.groups_id + if transition['group_id'] not in user_groups.ids: + return False + + return self.wkf_expr_eval_expr(transition['condition']) |