/usr/lib/python3/dist-packages/trytond/ir/note.py is in tryton-server 4.6.3-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | # This file is part of Tryton. The COPYRIGHT file at the top level of
# this repository contains the full copyright notices and license terms.
from textwrap import TextWrapper
from sql import Null
from sql.conditionals import Case
from ..model import ModelView, ModelSQL, ModelStorage, fields
from ..pool import Pool
from ..transaction import Transaction
from ..tools import grouped_slice, reduce_ids
from ..pyson import Eval
from .resource import ResourceMixin
__all__ = ['Note', 'NoteRead']
class Note(ResourceMixin, ModelSQL, ModelView):
"Note"
__name__ = 'ir.note'
message = fields.Text('Message', states={
'readonly': Eval('id', 0) > 0,
})
message_wrapped = fields.Function(fields.Text('Message'),
'on_change_with_message_wrapped')
unread = fields.Function(fields.Boolean('Unread'), 'get_unread',
searcher='search_unread', setter='set_unread')
@staticmethod
def default_unread():
return False
@classmethod
def get_wrapper(cls):
return TextWrapper(width=79)
@fields.depends('message')
def on_change_with_message_wrapped(self, name=None):
wrapper = self.get_wrapper()
message = self.message or ''
return '\n'.join(map(wrapper.fill, message.splitlines()))
@classmethod
def get_unread(cls, ids, name):
pool = Pool()
Read = pool.get('ir.note.read')
cursor = Transaction().connection.cursor()
user_id = Transaction().user
table = cls.__table__()
read = Read.__table__()
unread = {}
for sub_ids in grouped_slice(ids):
where = reduce_ids(table.id, sub_ids)
query = table.join(read, 'LEFT',
condition=(table.id == read.note)
& (read.user == user_id)
).select(table.id,
Case((read.user != Null, False), else_=True),
where=where)
cursor.execute(*query)
unread.update(cursor.fetchall())
return unread
@classmethod
def search_unread(cls, name, clause):
pool = Pool()
Read = pool.get('ir.note.read')
user_id = Transaction().user
table = cls.__table__()
read = Read.__table__()
_, operator, value = clause
assert operator in ['=', '!=']
Operator = fields.SQL_OPERATORS[operator]
where = Operator(Case((read.user != Null, False), else_=True), value)
query = table.join(read, 'LEFT',
condition=(table.id == read.note)
& (read.user == user_id)
).select(table.id, where=where)
return [('id', 'in', query)]
@classmethod
def set_unread(cls, notes, name, value):
pool = Pool()
Read = pool.get('ir.note.read')
user_id = Transaction().user
if not value:
Read.create([{'note': n.id, 'user': user_id} for n in notes])
else:
reads = []
for sub_notes in grouped_slice(notes):
reads += Read.search([
('note', 'in', [n.id for n in sub_notes]),
('user', '=', user_id),
])
Read.delete(reads)
@classmethod
def write(cls, notes, values, *args):
# Avoid changing write meta data if only unread is set
if args or list(values.keys()) != ['unread']:
super(Note, cls).write(notes, values, *args)
else:
# Check access write and clean cache
ModelStorage.write(notes, values)
cls.set_unread(notes, 'unread', values['unread'])
class NoteRead(ModelSQL):
"Note Read"
__name__ = 'ir.note.read'
note = fields.Many2One('ir.note', 'Note', required=True,
ondelete='CASCADE')
user = fields.Many2One('res.user', 'User', required=True,
ondelete='CASCADE')
|