6418546956

Committed 05 Oct 2023 11:43AM UTC coverage: 72.382% (-0.004%) from 72.386%

Build # 6418546956

Build Type

Pull #1889

github

Committed by

pre-commit-ci[bot]

Commit Message

[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Pull Request Pull Request #1889: Remove mandatory email requirement from new participant form

Run Details

1 of 8 new or added lines in 2 files covered. (12.5%)

40 existing lines in 3 files now uncovered.

13269 of 18332 relevant lines covered (72.38%)

0.72 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.33

/funnel/models/sync_ticket.py

"""Models for syncing tickets to a project from an external ticketing provider."""

from __future__ import annotations

import base64
import os
from collections.abc import Iterable, Sequence
from typing import Any

from coaster.sqlalchemy import LazyRoleSet

from . import (
    BaseMixin,
    BaseScopedNameMixin,
    DynamicMapped,
    Mapped,
    Model,
    Query,
    UuidMixin,
    backref,
    db,
    relationship,
    sa,
    with_roles,
)
from .account import Account, AccountEmail
from .email_address import EmailAddress, EmailAddressMixin
from .helpers import reopen
from .project import Project
from .project_membership import project_child_role_map

__all__ = [
    'SyncTicket',
    'TicketClient',
    'TicketEvent',
    'TicketEventParticipant',
    'TicketParticipant',
    'TicketType',
]


def make_key():
    return base64.urlsafe_b64encode(os.urandom(128)).decode('utf-8')


def make_public_key():
    return make_key()[:8]


def make_private_key():
    return make_key()[:8]


ticket_event_ticket_type = sa.Table(
    'ticket_event_ticket_type',
    Model.metadata,
    sa.Column(
        'ticket_event_id',
        sa.Integer,
        sa.ForeignKey('ticket_event.id'),
        primary_key=True,
    ),
    sa.Column(
        'ticket_type_id', sa.Integer, sa.ForeignKey('ticket_type.id'), primary_key=True
    ),
    sa.Column(
        'created_at',
        sa.TIMESTAMP(timezone=True),
        default=sa.func.utcnow(),
        nullable=False,
    ),
)


class GetTitleMixin(BaseScopedNameMixin):
    @classmethod
    def get(
        cls, parent: Any, name: str | None = None, title: str | None = None
    ) -> GetTitleMixin | None:
        if not bool(name) ^ bool(title):
            raise TypeError("Expects name xor title")
        if name:
            return cls.query.filter_by(parent=parent, name=name).one_or_none()
        return cls.query.filter_by(parent=parent, title=title).one_or_none()

    @classmethod
    def upsert(  # type: ignore[override]  # pylint: disable=arguments-renamed
        cls,
        parent: Any,
        current_name: str | None = None,
        current_title: str | None = None,
        **fields,
    ) -> GetTitleMixin:
        instance = cls.get(parent, current_name, current_title)
        if instance is not None:
            instance._set_fields(fields)  # pylint: disable=protected-access
        else:
            fields.pop('title', None)
            instance = cls(parent=parent, title=current_title, **fields)
            db.session.add(instance)
        return instance


class TicketEvent(GetTitleMixin, Model):
    """
    A discrete event under a project that a ticket grants access to.

    A project may have multiple events, such as a workshop and a two-day conference.
    The workshop is one discrete event, as is each day of the two-day conference.
    Tickets and events have a many-to-many relationship within a project. A ticket type
    may grant access to multiple events and a different ticket type may grant an
    overlapping set of events.
    """

    __tablename__ = 'ticket_event'

    project_id = sa.orm.mapped_column(
        sa.Integer, sa.ForeignKey('project.id'), nullable=False
    )
    project: Mapped[Project] = with_roles(
        relationship(Project, backref=backref('ticket_events', cascade='all')),
        rw={'project_promoter'},
        grants_via={None: project_child_role_map},
    )
    parent: Mapped[Project] = sa.orm.synonym('project')
    ticket_types: Mapped[list[TicketType]] = with_roles(
        relationship(
            'TicketType',
            secondary=ticket_event_ticket_type,
            back_populates='ticket_events',
        ),
        rw={'project_promoter'},
    )
    ticket_participants: DynamicMapped[TicketParticipant] = with_roles(
        relationship(
            'TicketParticipant',
            secondary='ticket_event_participant',
            backref='ticket_events',
            lazy='dynamic',
        ),
        rw={'project_promoter'},
    )
    badge_template = with_roles(
        sa.orm.mapped_column(sa.Unicode(250), nullable=True), rw={'project_promoter'}
    )

    __table_args__ = (
        sa.UniqueConstraint('project_id', 'name'),
        sa.UniqueConstraint('project_id', 'title'),
    )

    __roles__ = {
        'all': {
            'call': {'url_for'},
        },
        'project_promoter': {
            'read': {'name', 'title'},
            'write': {'name', 'title'},
        },
    }


class TicketType(GetTitleMixin, Model):
    """
    A ticket type that can grant access to multiple events within a project.

    Eg: Early Geek, Super Early Geek, Workshop A, B, C.
    """

    __tablename__ = 'ticket_type'

    project_id = sa.orm.mapped_column(
        sa.Integer, sa.ForeignKey('project.id'), nullable=False
    )
    project: Mapped[Project] = with_roles(
        relationship(Project, backref=backref('ticket_types', cascade='all')),
        rw={'project_promoter'},
        grants_via={None: project_child_role_map},
    )
    parent: Mapped[Project] = sa.orm.synonym('project')
    ticket_events: Mapped[list[TicketEvent]] = with_roles(
        relationship(
            TicketEvent,
            secondary=ticket_event_ticket_type,
            back_populates='ticket_types',
        ),
        rw={'project_promoter'},
    )

    __table_args__ = (
        sa.UniqueConstraint('project_id', 'name'),
        sa.UniqueConstraint('project_id', 'title'),
    )

    __roles__ = {
        'all': {
            'call': {'url_for'},
        },
        'project_promoter': {
            'read': {'name', 'title'},
            'write': {'name', 'title'},
        },
    }


class TicketParticipant(EmailAddressMixin, UuidMixin, BaseMixin, Model):
    """A participant in one or more events, synced from an external ticket source."""

    __tablename__ = 'ticket_participant'
    __email_optional__ = True
    __email_for__ = 'participant'

    fullname = with_roles(
        sa.orm.mapped_column(sa.Unicode(80), nullable=False),
        read={'promoter', 'member', 'scanner'},
    )
    #: Unvalidated phone number
    phone = with_roles(
        sa.orm.mapped_column(sa.Unicode(80), nullable=True),
        read={'promoter', 'member', 'scanner'},
    )
    #: Unvalidated Twitter id
    twitter = with_roles(
        sa.orm.mapped_column(sa.Unicode(80), nullable=True),
        read={'promoter', 'member', 'scanner'},
    )
    #: Job title
    job_title = with_roles(
        sa.orm.mapped_column(sa.Unicode(80), nullable=True),
        read={'promoter', 'member', 'scanner'},
    )
    #: Company
    company = with_roles(
        sa.orm.mapped_column(sa.Unicode(80), nullable=True),
        read={'promoter', 'member', 'scanner'},
    )
    #: Participant's city
    city = with_roles(
        sa.orm.mapped_column(sa.Unicode(80), nullable=True),
        read={'promoter', 'member', 'scanner'},
    )
    # public key
    puk = sa.orm.mapped_column(
        sa.Unicode(44), nullable=False, default=make_public_key, unique=True
    )
    key = sa.orm.mapped_column(
        sa.Unicode(44), nullable=False, default=make_private_key, unique=True
    )
    badge_printed = sa.orm.mapped_column(sa.Boolean, default=False, nullable=False)
    participant_id: Mapped[int | None] = sa.orm.mapped_column(
        sa.ForeignKey('account.id'), nullable=True
    )
    participant: Mapped[Account | None] = relationship(
        Account, backref=backref('ticket_participants', cascade='all')
    )
    project_id = sa.orm.mapped_column(
        sa.Integer, sa.ForeignKey('project.id'), nullable=False
    )
    project: Mapped[Project] = with_roles(
        relationship(Project, back_populates='ticket_participants'),
        read={'promoter', 'member', 'scanner'},
        grants_via={None: project_child_role_map},
    )

    __table_args__ = (sa.UniqueConstraint('project_id', 'email_address_id'),)

    # Since 'email' comes from the mixin, it's not available to be annotated using
    # `with_roles`. Instead, we have to specify the roles that can access it in here:
    __roles__ = {
        'promoter': {'read': {'email'}},
        'member': {'read': {'email'}},
        'scanner': {'read': {'email'}},
    }

    def roles_for(
        self, actor: Account | None = None, anchors: Sequence = ()
    ) -> LazyRoleSet:
        roles = super().roles_for(actor, anchors)
        if actor is not None:
            if actor == self.participant:
                roles.add('member')
            cx = ContactExchange.query.get((actor.id, self.id))
            if cx is not None:
                roles.add('scanner')
        return roles

    @property
    def avatar(self):
        return self.participant.logo_url if self.participant else ''

    with_roles(avatar, read={'all'})

    @property
    def has_public_profile(self) -> bool:
        return self.participant.has_public_profile if self.participant else False

    with_roles(has_public_profile, read={'all'})

    @property
    def profile_url(self) -> str | None:
        return self.participant.profile_url if self.participant else None

    with_roles(profile_url, read={'all'})

    @classmethod
    def get(
        cls, current_project: Project, current_email: str
    ) -> TicketParticipant | None:
        return cls.query.filter_by(
            project=current_project, email_address=EmailAddress.get(current_email)
        ).one_or_none()

    @classmethod
    def upsert(
        cls, current_project: Project, current_email: str, **fields
    ) -> TicketParticipant:
        ticket_participant = cls.get(current_project, current_email)
        accountemail = AccountEmail.get(current_email)
        if accountemail is not None:
            participant = accountemail.account
        else:
            participant = None
        if ticket_participant is not None:
            ticket_participant.participant = participant
            ticket_participant._set_fields(fields)  # pylint: disable=protected-access
        else:
            with db.session.no_autoflush:
                ticket_participant = cls(
                    project=current_project,
                    participant=participant,
                    email=current_email,
                    **fields,
                )
            db.session.add(ticket_participant)
        return ticket_participant

    def add_events(self, ticket_events: Iterable[TicketEvent]) -> None:
        for ticket_event in ticket_events:
            if ticket_event not in self.ticket_events:
                self.ticket_events.append(ticket_event)

    def remove_events(self, ticket_events: Iterable[TicketEvent]) -> None:
        for ticket_event in ticket_events:
            if ticket_event in self.ticket_events:
                self.ticket_events.remove(ticket_event)

    @classmethod
    def checkin_list(cls, ticket_event: TicketEvent) -> list:  # TODO: List type?
        """
        Return ticket participant details as a comma separated string.

        Also includes associated ticket types.

        FIXME: This is bad design and should be replaced with a saner mechanism.
        """
        query = (
            db.session.query(
                sa.func.distinct(cls.uuid).label('uuid'),
                cls.fullname.label('fullname'),
                EmailAddress.email.label('email'),
                cls.company.label('company'),
                cls.twitter.label('twitter'),
                cls.puk.label('puk'),
                cls.key.label('key'),
                TicketEventParticipant.checked_in.label('checked_in'),
                cls.badge_printed.label('badge_printed'),
                db.session.query(sa.func.string_agg(TicketType.title, ','))
                .select_from(SyncTicket)
                .join(TicketType, SyncTicket.ticket_type_id == TicketType.id)
                .filter(SyncTicket.ticket_participant_id == TicketParticipant.id)
                .label('ticket_type_titles'),
                cls.participant_id.is_not(None).label('has_user'),
            )
            .select_from(TicketParticipant)
            .join(
                TicketEventParticipant,
                TicketParticipant.id == TicketEventParticipant.ticket_participant_id,
            )
            .join(EmailAddress, EmailAddress.id == TicketParticipant.email_address_id)
            .outerjoin(
                SyncTicket, TicketParticipant.id == SyncTicket.ticket_participant_id
            )
            .filter(TicketEventParticipant.ticket_event_id == ticket_event.id)
            .order_by(TicketParticipant.fullname)
        )
        return query.all()


class TicketEventParticipant(BaseMixin, Model):
    """Join model between :class:`TicketParticipant` and :class:`TicketEvent`."""

    __tablename__ = 'ticket_event_participant'

    ticket_participant_id = sa.orm.mapped_column(
        sa.Integer, sa.ForeignKey('ticket_participant.id'), nullable=False
    )
    ticket_participant: Mapped[TicketParticipant] = relationship(
        TicketParticipant,
        backref=backref(
            'ticket_event_participants',
            cascade='all',
            overlaps='ticket_events,ticket_participants',
        ),
        overlaps='ticket_events,ticket_participants',
    )
    ticket_event_id = sa.orm.mapped_column(
        sa.Integer, sa.ForeignKey('ticket_event.id'), nullable=False
    )
    ticket_event: Mapped[TicketEvent] = relationship(
        TicketEvent,
        backref=backref(
            'ticket_event_participants',
            cascade='all',
            overlaps='ticket_events,ticket_participants',
        ),
        overlaps='ticket_events,ticket_participants',
    )
    checked_in = sa.orm.mapped_column(sa.Boolean, default=False, nullable=False)

    __table_args__ = (
        # Uses a custom name that is not as per convention because the default name is
        # too long for PostgreSQL
        sa.UniqueConstraint(
            'ticket_event_id',
            'ticket_participant_id',
            name='ticket_event_participant_event_id_participant_id_key',
        ),
    )

    @classmethod
    def get(
        cls, ticket_event: TicketEvent, participant_uuid_b58: str
    ) -> TicketEventParticipant | None:
        return (
            cls.query.join(TicketParticipant)
            .filter(
                TicketEventParticipant.ticket_event == ticket_event,
                TicketParticipant.uuid_b58 == participant_uuid_b58,
            )
            .one_or_none()
        )


class TicketClient(BaseMixin, Model):
    __tablename__ = 'ticket_client'
    name = with_roles(
        sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'}
    )
    client_eventid = with_roles(
        sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'}
    )
    clientid = with_roles(
        sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'}
    )
    client_secret = with_roles(
        sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'}
    )
    client_access_token = with_roles(
        sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'}
    )
    project_id = sa.orm.mapped_column(
        sa.Integer, sa.ForeignKey('project.id'), nullable=False
    )
    project = with_roles(
        relationship(Project, backref=backref('ticket_clients', cascade='all')),
        rw={'project_promoter'},
        grants_via={None: project_child_role_map},
    )

    __roles__ = {'all': {'call': {'url_for'}}}

    def import_from_list(self, ticket_list):
        """Batch upsert tickets and their associated ticket types and participants."""
        for ticket_dict in ticket_list:
            ticket_type = TicketType.upsert(
                self.project, current_title=ticket_dict['ticket_type']
            )

            ticket_participant = TicketParticipant.upsert(
                self.project,
                ticket_dict['email'],
                fullname=ticket_dict['fullname'],
                phone=ticket_dict['phone'],
                twitter=ticket_dict['twitter'],
                company=ticket_dict['company'],
                job_title=ticket_dict['job_title'],
                city=ticket_dict['city'],
            )

            ticket = SyncTicket.get(
                self, ticket_dict.get('order_no'), ticket_dict.get('ticket_no')
            )
            if ticket and (
                ticket.ticket_participant != ticket_participant
                or ticket_dict.get('status') == 'cancelled'
            ):
                # Ensure that the participant of a transferred or cancelled ticket does
                # not have access to this ticket's events
                ticket.ticket_participant.remove_events(ticket_type.ticket_events)

            if ticket_dict.get('status') == 'confirmed':
                ticket = SyncTicket.upsert(
                    self,
                    ticket_dict.get('order_no'),
                    ticket_dict.get('ticket_no'),
                    ticket_participant=ticket_participant,
                    ticket_type=ticket_type,
                )
                # Ensure that the new or updated participant has access to events
                ticket.ticket_participant.add_events(ticket_type.ticket_events)


class SyncTicket(BaseMixin, Model):
    """Model for a ticket that was bought elsewhere, like Boxoffice or Explara."""

    __tablename__ = 'sync_ticket'

    ticket_no = sa.orm.mapped_column(sa.Unicode(80), nullable=False)
    order_no = sa.orm.mapped_column(sa.Unicode(80), nullable=False)
    ticket_type_id = sa.orm.mapped_column(
        sa.Integer, sa.ForeignKey('ticket_type.id'), nullable=False
    )
    ticket_type: Mapped[TicketType] = relationship(
        TicketType, backref=backref('sync_tickets', cascade='all')
    )
    ticket_participant_id = sa.orm.mapped_column(
        sa.Integer, sa.ForeignKey('ticket_participant.id'), nullable=False
    )
    ticket_participant: Mapped[TicketParticipant] = relationship(
        TicketParticipant,
        backref=backref('sync_tickets', cascade='all'),
    )
    ticket_client_id = sa.orm.mapped_column(
        sa.Integer, sa.ForeignKey('ticket_client.id'), nullable=False
    )
    ticket_client: Mapped[TicketClient] = relationship(
        TicketClient, backref=backref('sync_tickets', cascade='all')
    )
    __table_args__ = (sa.UniqueConstraint('ticket_client_id', 'order_no', 'ticket_no'),)

    @classmethod
    def get(
        cls, ticket_client: TicketClient, order_no: str, ticket_no: str
    ) -> SyncTicket | None:
        return cls.query.filter_by(
            ticket_client=ticket_client, order_no=order_no, ticket_no=ticket_no
        ).one_or_none()

    @classmethod
    def upsert(
        cls, ticket_client: TicketClient, order_no: str, ticket_no: str, **fields
    ) -> SyncTicket:
        """
        Update or insert ticket details.

        Returns a tuple containing the upserted ticket, and the participant the ticket
        was previously associated with or None if there was no earlier participant.
        """
        ticket = cls.get(ticket_client, order_no, ticket_no)
        if ticket is not None:
            ticket._set_fields(fields)  # pylint: disable=protected-access
        else:
            fields.pop('ticket_client', None)
            fields.pop('order_no', None)
            fields.pop('ticket_no', None)
            ticket = cls(
                ticket_client=ticket_client,
                order_no=order_no,
                ticket_no=ticket_no,
                **fields,
            )

            db.session.add(ticket)

        return ticket


@reopen(Project)
class __Project:
    # XXX: This relationship exposes an edge case in RoleMixin. It previously expected
    # TicketParticipant.participant to be unique per project, meaning one user could
    # have one participant ticket only. This is not guaranteed by the model as tickets
    # are unique per email address per ticket type, and one user can have (a) two email
    # addresses with tickets, or (b) tickets of different types. RoleMixin has since
    # been patched to look for the first matching record (.first() instead of .one()).
    # This may expose a new edge case in future in case the TicketParticipant model adds
    # an `offered_roles` method, as only the first matching record's method will be
    # called
    ticket_participants: DynamicMapped[TicketParticipant] = with_roles(
        relationship(
            TicketParticipant, lazy='dynamic', cascade='all', back_populates='project'
        ),
        grants_via={
            'participant': {'participant', 'project_participant', 'ticket_participant'}
        },
    )


@reopen(Account)
class __Account:
    @property
    def ticket_followers(self) -> Query[Account]:
        """All users with a ticket in a project."""
        return (
            Account.query.filter(Account.state.ACTIVE)
            .join(TicketParticipant, TicketParticipant.participant_id == Account.id)
            .join(Project, TicketParticipant.project_id == Project.id)
            .filter(Project.state.PUBLISHED, Project.account == self)
        )

    with_roles(ticket_followers, grants={'follower'})


# Tail imports to avoid cyclic dependency errors, for symbols used only in methods
# pylint: disable=wrong-import-position
from .contact_exchange import ContactExchange  # isort:skip

1	"""Models for syncing tickets to a project from an external ticketing provider."""	1✔
2
3	from __future__ import annotations	1✔
4
5	import base64	1✔
6	import os	1✔
7	from collections.abc import Iterable, Sequence	1✔
8	from typing import Any	1✔
9
10	from coaster.sqlalchemy import LazyRoleSet	1✔
11
12	from . import (	1✔
13	BaseMixin,
14	BaseScopedNameMixin,
15	DynamicMapped,
16	Mapped,
17	Model,
18	Query,
19	UuidMixin,
20	backref,
21	db,
22	relationship,
23	sa,
24	with_roles,
25	)
26	from .account import Account, AccountEmail	1✔
27	from .email_address import EmailAddress, EmailAddressMixin	1✔
28	from .helpers import reopen	1✔
29	from .project import Project	1✔
30	from .project_membership import project_child_role_map	1✔
31
32	__all__ = [	1✔
33	'SyncTicket',
34	'TicketClient',
35	'TicketEvent',
36	'TicketEventParticipant',
37	'TicketParticipant',
38	'TicketType',
39	]
40
41
42	def make_key():	1✔
43	return base64.urlsafe_b64encode(os.urandom(128)).decode('utf-8')	1✔
44
45
46	def make_public_key():	1✔
47	return make_key()[:8]	1✔
48
49
50	def make_private_key():	1✔
51	return make_key()[:8]	1✔
52
53
54	ticket_event_ticket_type = sa.Table(	1✔
55	'ticket_event_ticket_type',
56	Model.metadata,
57	sa.Column(
58	'ticket_event_id',
59	sa.Integer,
60	sa.ForeignKey('ticket_event.id'),
61	primary_key=True,
62	),
63	sa.Column(
64	'ticket_type_id', sa.Integer, sa.ForeignKey('ticket_type.id'), primary_key=True
65	),
66	sa.Column(
67	'created_at',
68	sa.TIMESTAMP(timezone=True),
69	default=sa.func.utcnow(),
70	nullable=False,
71	),
72	)
73
74
75	class GetTitleMixin(BaseScopedNameMixin):	1✔
76	@classmethod	1✔
77	def get(	1✔
78	cls, parent: Any, name: str \| None = None, title: str \| None = None
79	) -> GetTitleMixin \| None:
80	if not bool(name) ^ bool(title):	1✔
81	raise TypeError("Expects name xor title")	×
82	if name:	1✔
83	return cls.query.filter_by(parent=parent, name=name).one_or_none()	×
84	return cls.query.filter_by(parent=parent, title=title).one_or_none()	1✔
85
86	@classmethod	1✔
87	def upsert( # type: ignore[override] # pylint: disable=arguments-renamed	1✔
88	cls,
89	parent: Any,
90	current_name: str \| None = None,
91	current_title: str \| None = None,
92	**fields,
93	) -> GetTitleMixin:
94	instance = cls.get(parent, current_name, current_title)	1✔
95	if instance is not None:	1✔
96	instance._set_fields(fields) # pylint: disable=protected-access	1✔
97	else:
98	fields.pop('title', None)	1✔
99	instance = cls(parent=parent, title=current_title, **fields)	1✔
100	db.session.add(instance)	1✔
101	return instance	1✔
102
103
104	class TicketEvent(GetTitleMixin, Model):	1✔
105	"""	1✔
106	A discrete event under a project that a ticket grants access to.
107
108	A project may have multiple events, such as a workshop and a two-day conference.
109	The workshop is one discrete event, as is each day of the two-day conference.
110	Tickets and events have a many-to-many relationship within a project. A ticket type
111	may grant access to multiple events and a different ticket type may grant an
112	overlapping set of events.
113	"""
114
115	__tablename__ = 'ticket_event'	1✔
116
117	project_id = sa.orm.mapped_column(	1✔
118	sa.Integer, sa.ForeignKey('project.id'), nullable=False
119	)
120	project: Mapped[Project] = with_roles(	1✔
121	relationship(Project, backref=backref('ticket_events', cascade='all')),
122	rw={'project_promoter'},
123	grants_via={None: project_child_role_map},
124	)
125	parent: Mapped[Project] = sa.orm.synonym('project')	1✔
126	ticket_types: Mapped[list[TicketType]] = with_roles(	1✔
127	relationship(
128	'TicketType',
129	secondary=ticket_event_ticket_type,
130	back_populates='ticket_events',
131	),
132	rw={'project_promoter'},
133	)
134	ticket_participants: DynamicMapped[TicketParticipant] = with_roles(	1✔
135	relationship(
136	'TicketParticipant',
137	secondary='ticket_event_participant',
138	backref='ticket_events',
139	lazy='dynamic',
140	),
141	rw={'project_promoter'},
142	)
143	badge_template = with_roles(	1✔
144	sa.orm.mapped_column(sa.Unicode(250), nullable=True), rw={'project_promoter'}
145	)
146
147	__table_args__ = (	1✔
148	sa.UniqueConstraint('project_id', 'name'),
149	sa.UniqueConstraint('project_id', 'title'),
150	)
151
152	__roles__ = {	1✔
153	'all': {
154	'call': {'url_for'},
155	},
156	'project_promoter': {
157	'read': {'name', 'title'},
158	'write': {'name', 'title'},
159	},
160	}
161
162
163	class TicketType(GetTitleMixin, Model):	1✔
164	"""	1✔
165	A ticket type that can grant access to multiple events within a project.
166
167	Eg: Early Geek, Super Early Geek, Workshop A, B, C.
168	"""
169
170	__tablename__ = 'ticket_type'	1✔
171
172	project_id = sa.orm.mapped_column(	1✔
173	sa.Integer, sa.ForeignKey('project.id'), nullable=False
174	)
175	project: Mapped[Project] = with_roles(	1✔
176	relationship(Project, backref=backref('ticket_types', cascade='all')),
177	rw={'project_promoter'},
178	grants_via={None: project_child_role_map},
179	)
180	parent: Mapped[Project] = sa.orm.synonym('project')	1✔
181	ticket_events: Mapped[list[TicketEvent]] = with_roles(	1✔
182	relationship(
183	TicketEvent,
184	secondary=ticket_event_ticket_type,
185	back_populates='ticket_types',
186	),
187	rw={'project_promoter'},
188	)
189
190	__table_args__ = (	1✔
191	sa.UniqueConstraint('project_id', 'name'),
192	sa.UniqueConstraint('project_id', 'title'),
193	)
194
195	__roles__ = {	1✔
196	'all': {
197	'call': {'url_for'},
198	},
199	'project_promoter': {
200	'read': {'name', 'title'},
201	'write': {'name', 'title'},
202	},
203	}
204
205
206	class TicketParticipant(EmailAddressMixin, UuidMixin, BaseMixin, Model):	1✔
207	"""A participant in one or more events, synced from an external ticket source."""	1✔
208
209	__tablename__ = 'ticket_participant'	1✔
210	__email_optional__ = True	1✔
211	__email_for__ = 'participant'	1✔
212
213	fullname = with_roles(	1✔
214	sa.orm.mapped_column(sa.Unicode(80), nullable=False),
215	read={'promoter', 'member', 'scanner'},
216	)
217	#: Unvalidated phone number
218	phone = with_roles(	1✔
219	sa.orm.mapped_column(sa.Unicode(80), nullable=True),
220	read={'promoter', 'member', 'scanner'},
221	)
222	#: Unvalidated Twitter id
223	twitter = with_roles(	1✔
224	sa.orm.mapped_column(sa.Unicode(80), nullable=True),
225	read={'promoter', 'member', 'scanner'},
226	)
227	#: Job title
228	job_title = with_roles(	1✔
229	sa.orm.mapped_column(sa.Unicode(80), nullable=True),
230	read={'promoter', 'member', 'scanner'},
231	)
232	#: Company
233	company = with_roles(	1✔
234	sa.orm.mapped_column(sa.Unicode(80), nullable=True),
235	read={'promoter', 'member', 'scanner'},
236	)
237	#: Participant's city
238	city = with_roles(	1✔
239	sa.orm.mapped_column(sa.Unicode(80), nullable=True),
240	read={'promoter', 'member', 'scanner'},
241	)
242	# public key
243	puk = sa.orm.mapped_column(	1✔
244	sa.Unicode(44), nullable=False, default=make_public_key, unique=True
245	)
246	key = sa.orm.mapped_column(	1✔
247	sa.Unicode(44), nullable=False, default=make_private_key, unique=True
248	)
249	badge_printed = sa.orm.mapped_column(sa.Boolean, default=False, nullable=False)	1✔
250	participant_id: Mapped[int \| None] = sa.orm.mapped_column(	1✔
251	sa.ForeignKey('account.id'), nullable=True
252	)
253	participant: Mapped[Account \| None] = relationship(	1✔
254	Account, backref=backref('ticket_participants', cascade='all')
255	)
256	project_id = sa.orm.mapped_column(	1✔
257	sa.Integer, sa.ForeignKey('project.id'), nullable=False
258	)
259	project: Mapped[Project] = with_roles(	1✔
260	relationship(Project, back_populates='ticket_participants'),
261	read={'promoter', 'member', 'scanner'},
262	grants_via={None: project_child_role_map},
263	)
264
265	__table_args__ = (sa.UniqueConstraint('project_id', 'email_address_id'),)	1✔
266
267	# Since 'email' comes from the mixin, it's not available to be annotated using
268	# `with_roles`. Instead, we have to specify the roles that can access it in here:
269	__roles__ = {	1✔
270	'promoter': {'read': {'email'}},
271	'member': {'read': {'email'}},
272	'scanner': {'read': {'email'}},
273	}
274
275	def roles_for(	1✔
276	self, actor: Account \| None = None, anchors: Sequence = ()
277	) -> LazyRoleSet:
278	roles = super().roles_for(actor, anchors)	×
279	if actor is not None:	×
280	if actor == self.participant:	×
281	roles.add('member')	×
282	cx = ContactExchange.query.get((actor.id, self.id))	×
283	if cx is not None:	×
284	roles.add('scanner')	×
285	return roles	×
286
287	@property	1✔
288	def avatar(self):	1✔
289	return self.participant.logo_url if self.participant else ''	×
290
291	with_roles(avatar, read={'all'})	1✔
292
293	@property	1✔
294	def has_public_profile(self) -> bool:	1✔
295	return self.participant.has_public_profile if self.participant else False	×
296
297	with_roles(has_public_profile, read={'all'})	1✔
298
299	@property	1✔
300	def profile_url(self) -> str \| None:	1✔
301	return self.participant.profile_url if self.participant else None	×
302
303	with_roles(profile_url, read={'all'})	1✔
304
305	@classmethod	1✔
306	def get(	1✔
307	cls, current_project: Project, current_email: str
308	) -> TicketParticipant \| None:
309	return cls.query.filter_by(	1✔
310	project=current_project, email_address=EmailAddress.get(current_email)
311	).one_or_none()
312
313	@classmethod	1✔
314	def upsert(	1✔
315	cls, current_project: Project, current_email: str, **fields
316	) -> TicketParticipant:
317	ticket_participant = cls.get(current_project, current_email)	1✔
318	accountemail = AccountEmail.get(current_email)	1✔
319	if accountemail is not None:	1✔
320	participant = accountemail.account	×
321	else:
322	participant = None	1✔
323	if ticket_participant is not None:	1✔
324	ticket_participant.participant = participant	1✔
325	ticket_participant._set_fields(fields) # pylint: disable=protected-access	1✔
326	else:
327	with db.session.no_autoflush:	1✔
328	ticket_participant = cls(	1✔
329	project=current_project,
330	participant=participant,
331	email=current_email,
332	**fields,
333	)
334	db.session.add(ticket_participant)	1✔
335	return ticket_participant	1✔
336
337	def add_events(self, ticket_events: Iterable[TicketEvent]) -> None:	1✔
338	for ticket_event in ticket_events:	1✔
339	if ticket_event not in self.ticket_events:	1✔
340	self.ticket_events.append(ticket_event)	1✔
341
342	def remove_events(self, ticket_events: Iterable[TicketEvent]) -> None:	1✔
343	for ticket_event in ticket_events:	1✔
344	if ticket_event in self.ticket_events:	1✔
345	self.ticket_events.remove(ticket_event)	1✔
346
347	@classmethod	1✔
348	def checkin_list(cls, ticket_event: TicketEvent) -> list: # TODO: List type?	1✔
349	"""
350	Return ticket participant details as a comma separated string.
351
352	Also includes associated ticket types.
353
354	FIXME: This is bad design and should be replaced with a saner mechanism.
355	"""
356	query = (	×
357	db.session.query(
358	sa.func.distinct(cls.uuid).label('uuid'),
359	cls.fullname.label('fullname'),
360	EmailAddress.email.label('email'),
361	cls.company.label('company'),
362	cls.twitter.label('twitter'),
363	cls.puk.label('puk'),
364	cls.key.label('key'),
365	TicketEventParticipant.checked_in.label('checked_in'),
366	cls.badge_printed.label('badge_printed'),
367	db.session.query(sa.func.string_agg(TicketType.title, ','))
368	.select_from(SyncTicket)
369	.join(TicketType, SyncTicket.ticket_type_id == TicketType.id)
370	.filter(SyncTicket.ticket_participant_id == TicketParticipant.id)
371	.label('ticket_type_titles'),
372	cls.participant_id.is_not(None).label('has_user'),
373	)
374	.select_from(TicketParticipant)
375	.join(
376	TicketEventParticipant,
377	TicketParticipant.id == TicketEventParticipant.ticket_participant_id,
378	)
379	.join(EmailAddress, EmailAddress.id == TicketParticipant.email_address_id)
380	.outerjoin(
381	SyncTicket, TicketParticipant.id == SyncTicket.ticket_participant_id
382	)
383	.filter(TicketEventParticipant.ticket_event_id == ticket_event.id)
384	.order_by(TicketParticipant.fullname)
385	)
UNCOV 386	return query.all()	×
387
388
389	class TicketEventParticipant(BaseMixin, Model):	1✔
390	"""Join model between :class:`TicketParticipant` and :class:`TicketEvent`."""	1✔
391
392	__tablename__ = 'ticket_event_participant'	1✔
393
394	ticket_participant_id = sa.orm.mapped_column(	1✔
395	sa.Integer, sa.ForeignKey('ticket_participant.id'), nullable=False
396	)
397	ticket_participant: Mapped[TicketParticipant] = relationship(	1✔
398	TicketParticipant,
399	backref=backref(
400	'ticket_event_participants',
401	cascade='all',
402	overlaps='ticket_events,ticket_participants',
403	),
404	overlaps='ticket_events,ticket_participants',
405	)
406	ticket_event_id = sa.orm.mapped_column(	1✔
407	sa.Integer, sa.ForeignKey('ticket_event.id'), nullable=False
408	)
409	ticket_event: Mapped[TicketEvent] = relationship(	1✔
410	TicketEvent,
411	backref=backref(
412	'ticket_event_participants',
413	cascade='all',
414	overlaps='ticket_events,ticket_participants',
415	),
416	overlaps='ticket_events,ticket_participants',
417	)
418	checked_in = sa.orm.mapped_column(sa.Boolean, default=False, nullable=False)	1✔
419
420	__table_args__ = (	1✔
421	# Uses a custom name that is not as per convention because the default name is
422	# too long for PostgreSQL
423	sa.UniqueConstraint(
424	'ticket_event_id',
425	'ticket_participant_id',
426	name='ticket_event_participant_event_id_participant_id_key',
427	),
428	)
429
430	@classmethod	1✔
431	def get(	1✔
432	cls, ticket_event: TicketEvent, participant_uuid_b58: str
433	) -> TicketEventParticipant \| None:
UNCOV 434	return (	×
435	cls.query.join(TicketParticipant)
436	.filter(
437	TicketEventParticipant.ticket_event == ticket_event,
438	TicketParticipant.uuid_b58 == participant_uuid_b58,
439	)
440	.one_or_none()
441	)
442
443
444	class TicketClient(BaseMixin, Model):	1✔
445	__tablename__ = 'ticket_client'	1✔
446	name = with_roles(	1✔
447	sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'}
448	)
449	client_eventid = with_roles(	1✔
450	sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'}
451	)
452	clientid = with_roles(	1✔
453	sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'}
454	)
455	client_secret = with_roles(	1✔
456	sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'}
457	)
458	client_access_token = with_roles(	1✔
459	sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'}
460	)
461	project_id = sa.orm.mapped_column(	1✔
462	sa.Integer, sa.ForeignKey('project.id'), nullable=False
463	)
464	project = with_roles(	1✔
465	relationship(Project, backref=backref('ticket_clients', cascade='all')),
466	rw={'project_promoter'},
467	grants_via={None: project_child_role_map},
468	)
469
470	__roles__ = {'all': {'call': {'url_for'}}}	1✔
471
472	def import_from_list(self, ticket_list):	1✔
473	"""Batch upsert tickets and their associated ticket types and participants."""
474	for ticket_dict in ticket_list:	1✔
475	ticket_type = TicketType.upsert(	1✔
476	self.project, current_title=ticket_dict['ticket_type']
477	)
478
479	ticket_participant = TicketParticipant.upsert(	1✔
480	self.project,
481	ticket_dict['email'],
482	fullname=ticket_dict['fullname'],
483	phone=ticket_dict['phone'],
484	twitter=ticket_dict['twitter'],
485	company=ticket_dict['company'],
486	job_title=ticket_dict['job_title'],
487	city=ticket_dict['city'],
488	)
489
490	ticket = SyncTicket.get(	1✔
491	self, ticket_dict.get('order_no'), ticket_dict.get('ticket_no')
492	)
493	if ticket and (	1✔
494	ticket.ticket_participant != ticket_participant
495	or ticket_dict.get('status') == 'cancelled'
496	):
497	# Ensure that the participant of a transferred or cancelled ticket does
498	# not have access to this ticket's events
499	ticket.ticket_participant.remove_events(ticket_type.ticket_events)	1✔
500
501	if ticket_dict.get('status') == 'confirmed':	1✔
502	ticket = SyncTicket.upsert(	1✔
503	self,
504	ticket_dict.get('order_no'),
505	ticket_dict.get('ticket_no'),
506	ticket_participant=ticket_participant,
507	ticket_type=ticket_type,
508	)
509	# Ensure that the new or updated participant has access to events
510	ticket.ticket_participant.add_events(ticket_type.ticket_events)	1✔
511
512
513	class SyncTicket(BaseMixin, Model):	1✔
514	"""Model for a ticket that was bought elsewhere, like Boxoffice or Explara."""	1✔
515
516	__tablename__ = 'sync_ticket'	1✔
517
518	ticket_no = sa.orm.mapped_column(sa.Unicode(80), nullable=False)	1✔
519	order_no = sa.orm.mapped_column(sa.Unicode(80), nullable=False)	1✔
520	ticket_type_id = sa.orm.mapped_column(	1✔
521	sa.Integer, sa.ForeignKey('ticket_type.id'), nullable=False
522	)
523	ticket_type: Mapped[TicketType] = relationship(	1✔
524	TicketType, backref=backref('sync_tickets', cascade='all')
525	)
526	ticket_participant_id = sa.orm.mapped_column(	1✔
527	sa.Integer, sa.ForeignKey('ticket_participant.id'), nullable=False
528	)
529	ticket_participant: Mapped[TicketParticipant] = relationship(	1✔
530	TicketParticipant,
531	backref=backref('sync_tickets', cascade='all'),
532	)
533	ticket_client_id = sa.orm.mapped_column(	1✔
534	sa.Integer, sa.ForeignKey('ticket_client.id'), nullable=False
535	)
536	ticket_client: Mapped[TicketClient] = relationship(	1✔
537	TicketClient, backref=backref('sync_tickets', cascade='all')
538	)
539	__table_args__ = (sa.UniqueConstraint('ticket_client_id', 'order_no', 'ticket_no'),)	1✔
540
541	@classmethod	1✔
542	def get(	1✔
543	cls, ticket_client: TicketClient, order_no: str, ticket_no: str
544	) -> SyncTicket \| None:
545	return cls.query.filter_by(	1✔
546	ticket_client=ticket_client, order_no=order_no, ticket_no=ticket_no
547	).one_or_none()
548
549	@classmethod	1✔
550	def upsert(	1✔
551	cls, ticket_client: TicketClient, order_no: str, ticket_no: str, **fields
552	) -> SyncTicket:
553	"""
554	Update or insert ticket details.
555
556	Returns a tuple containing the upserted ticket, and the participant the ticket
557	was previously associated with or None if there was no earlier participant.
558	"""
559	ticket = cls.get(ticket_client, order_no, ticket_no)	1✔
560	if ticket is not None:	1✔
561	ticket._set_fields(fields) # pylint: disable=protected-access	1✔
562	else:
563	fields.pop('ticket_client', None)	1✔
564	fields.pop('order_no', None)	1✔
565	fields.pop('ticket_no', None)	1✔
566	ticket = cls(	1✔
567	ticket_client=ticket_client,
568	order_no=order_no,
569	ticket_no=ticket_no,
570	**fields,
571	)
572
573	db.session.add(ticket)	1✔
574
575	return ticket	1✔
576
577
578	@reopen(Project)	1✔
579	class __Project:	1✔
580	# XXX: This relationship exposes an edge case in RoleMixin. It previously expected
581	# TicketParticipant.participant to be unique per project, meaning one user could
582	# have one participant ticket only. This is not guaranteed by the model as tickets
583	# are unique per email address per ticket type, and one user can have (a) two email
584	# addresses with tickets, or (b) tickets of different types. RoleMixin has since
585	# been patched to look for the first matching record (.first() instead of .one()).
586	# This may expose a new edge case in future in case the TicketParticipant model adds
587	# an `offered_roles` method, as only the first matching record's method will be
588	# called
589	ticket_participants: DynamicMapped[TicketParticipant] = with_roles(	1✔
590	relationship(
591	TicketParticipant, lazy='dynamic', cascade='all', back_populates='project'
592	),
593	grants_via={
594	'participant': {'participant', 'project_participant', 'ticket_participant'}
595	},
596	)
597
598
599	@reopen(Account)	1✔
600	class __Account:	1✔
601	@property	1✔
602	def ticket_followers(self) -> Query[Account]:	1✔
603	"""All users with a ticket in a project."""
604	return (	1✔
605	Account.query.filter(Account.state.ACTIVE)
606	.join(TicketParticipant, TicketParticipant.participant_id == Account.id)
607	.join(Project, TicketParticipant.project_id == Project.id)
608	.filter(Project.state.PUBLISHED, Project.account == self)
609	)
610
611	with_roles(ticket_followers, grants={'follower'})	1✔
612
613
614	# Tail imports to avoid cyclic dependency errors, for symbols used only in methods
615	# pylint: disable=wrong-import-position
616	from .contact_exchange import ContactExchange # isort:skip	1✔

hasgeek / funnel / 6418546956

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous