Source code for org.models.message

from functools import cached_property
from onegov.chat import Message
from onegov.core.elements import Link, Confirm, Intercooler
from onegov.core.utils import paragraphify, linkify
from onegov.event import Event
from onegov.org import _
from onegov.org.utils import hashtag_elements
from onegov.ticket import Ticket, TicketCollection
from sqlalchemy.orm import object_session


from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
    from collections.abc import Iterable, Iterator
    from libres.db.models import Reservation
    from onegov.chat.collections import MessageCollection
    from onegov.directory import Directory
    from onegov.file import File
    from onegov.org.layout import DefaultLayout
    from onegov.org.request import OrgRequest
    from onegov.pay import Payment
    from sqlalchemy import Column
    from sqlalchemy.orm import Session
    from typing import Self

# 👉 when adding new ticket messages be sure to evaluate if they should
# be added to the ticket status page through the org.public_ticket_messages
# setting


[docs] class TicketMessageMixin: if TYPE_CHECKING:
[docs] meta: Column[dict[str, Any]]
@classmethod def bound_messages(cls, session: Session) -> MessageCollection[Any]: ... @cached_property
[docs] def ticket(self) -> Ticket | None: return TicketCollection(object_session(self)).by_id( self.meta['id'], self.meta['handler_code'] )
@classmethod
[docs] def create( cls, ticket: Ticket, request: 'OrgRequest', text: str | None = None, owner: str | None = None, **extra_meta: Any ) -> 'Self': meta = { 'id': ticket.id.hex, 'handler_code': ticket.handler_code, 'group': ticket.group } meta.update(extra_meta) # force a change of the ticket to make sure that it gets reindexed ticket.force_update() # the owner can be forced to a specific value owner = owner or request.current_username or ticket.ticket_email return cls.bound_messages(request.session).add( channel_id=ticket.number, owner=owner, text=text, meta=meta )
[docs] class TicketNote(Message, TicketMessageMixin):
[docs] __mapper_args__ = { 'polymorphic_identity': 'ticket_note' }
if TYPE_CHECKING: # text is not optional for TicketNote
[docs] text: Column[str] # type:ignore[assignment]
@classmethod
[docs] def create( # type:ignore[override] cls, ticket: Ticket, request: 'OrgRequest', text: str, file: 'File | None' = None, owner: str | None = None ) -> 'Self': note = super().create(ticket, request, text=text, owner=owner) note.file = file return note
[docs] def formatted_text(self, layout: 'DefaultLayout') -> str: return hashtag_elements( layout.request, paragraphify(linkify(self.text)))
[docs] class TicketChatMessage(Message, TicketMessageMixin): """ Chat messages sent between the person in charge of the ticket and the submitter of the ticket. Paramters of note: - origin: 'external' or 'internal', to differentiate between the messages sent from the organisation to someone outside or from someone outside to someone inside. - notify: only relevant for messages originating from 'internal' - if the last sent message with origin 'internal' has this flag, a notification is sent to the owner of that message, whenever a new external reply comes in. """
[docs] __mapper_args__ = { 'polymorphic_identity': 'ticket_chat' }
@classmethod
[docs] def create( # type:ignore[override] cls, ticket: Ticket, request: 'OrgRequest', text: str, owner: str, origin: str, notify: bool = False, recipient: str | None = None ) -> 'Self': return super().create( ticket, request, text=text, owner=owner, origin=origin, notify=notify, recipient=recipient)
[docs] def formatted_text(self, layout: 'DefaultLayout') -> str: return self.text and hashtag_elements( layout.request, paragraphify(linkify(self.text))) or ''
@property
[docs] def subtype(self) -> str | None: return self.meta.get('origin', None)
[docs] class TicketMessage(Message, TicketMessageMixin):
[docs] __mapper_args__ = { 'polymorphic_identity': 'ticket' }
@classmethod
[docs] def create( # type:ignore[override] cls, ticket: Ticket, request: 'OrgRequest', change: str, **extra_meta: Any ) -> 'Self': return super().create(ticket, request, change=change, **extra_meta)
[docs] class ReservationMessage(Message, TicketMessageMixin):
[docs] __mapper_args__ = { 'polymorphic_identity': 'reservation' }
@classmethod
[docs] def create( # type:ignore[override] cls, reservations: 'Iterable[Reservation]', ticket: Ticket, request: 'OrgRequest', change: str ) -> 'Self': return super().create(ticket, request, change=change, reservations=[ r.id for r in reservations ])
[docs] class SubmissionMessage(Message, TicketMessageMixin):
[docs] __mapper_args__ = { 'polymorphic_identity': 'submission' }
@classmethod
[docs] def create( # type:ignore[override] cls, ticket: Ticket, request: 'OrgRequest', change: str ) -> 'Self': return super().create(ticket, request, change=change)
[docs] class EventMessage(Message, TicketMessageMixin):
[docs] __mapper_args__ = { 'polymorphic_identity': 'event' }
@classmethod
[docs] def create( # type:ignore[override] cls, event: Event, ticket: Ticket, request: 'OrgRequest', change: str ) -> 'Self': return super().create( ticket, request, change=change, event_name=event.name)
[docs] class PaymentMessage(Message, TicketMessageMixin):
[docs] __mapper_args__ = { 'polymorphic_identity': 'payment' }
@classmethod
[docs] def create( # type:ignore[override] cls, payment: 'Payment', ticket: Ticket, request: 'OrgRequest', change: str ) -> 'Self': assert payment.amount is not None return super().create( ticket, request, change=change, payment_id=payment.id.hex, amount=float(payment.amount), currency=payment.currency )
[docs] class DirectoryMessage(Message, TicketMessageMixin):
[docs] __mapper_args__ = { 'polymorphic_identity': 'directory' }
@classmethod
[docs] def create( # type:ignore[override] cls, directory: 'Directory', ticket: Ticket, request: 'OrgRequest', action: str ) -> 'Self': return super().create( ticket, request, directory_id=directory.id.hex, action=action )