8456388393

Committed 27 Mar 2024 05:54PM UTC coverage: 72.154% (-0.05%) from 72.204%

Build # 8456388393

Build Type

Pull #2005

github

Committed by

jace

Commit Message

Migrate project participants to account followers

Pull Request Pull Request #2005: Support account follow

Run Details

61 of 99 new or added lines in 6 files covered. (61.62%)

1 existing line in 1 file now uncovered.

13549 of 18778 relevant lines covered (72.15%)

1.44 hits per line

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

90.39

/funnel/models/account.py

"""Account model with subtypes, and account-linked personal data models."""

# pylint: disable=unnecessary-lambda,invalid-unary-operand-type
# pyright: reportGeneralTypeIssues=false,reportAttributeAccessIssue=false

from __future__ import annotations

import hashlib
import itertools
from collections.abc import Iterable, Iterator, Sequence
from datetime import datetime
from typing import (
    TYPE_CHECKING,
    Any,
    ClassVar,
    Literal,
    Self,
    TypeAlias,
    cast,
    overload,
)
from uuid import UUID

import phonenumbers
from babel import Locale
from furl import furl
from passlib.hash import argon2, bcrypt
from pytz.tzinfo import BaseTzInfo
from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy
from sqlalchemy.ext.hybrid import Comparator
from sqlalchemy.sql.expression import ColumnElement
from werkzeug.utils import cached_property
from zbase32 import decode as zbase32_decode, encode as zbase32_encode

from baseframe import __
from coaster.sqlalchemy import (
    DynamicAssociationProxy,
    RoleMixin,
    StateManager,
    add_primary_relationship,
    auto_init_default,
    failsafe_add,
    immutable,
    role_check,
    with_roles,
)
from coaster.utils import LabeledEnum, newsecret, require_one_of, utcnow

from ..typing import OptionalMigratedTables
from .base import (
    BaseMixin,
    DynamicMapped,
    LocaleType,
    Mapped,
    Model,
    Query,
    TimezoneType,
    TSVectorType,
    UrlType,
    UuidMixin,
    db,
    hybrid_property,
    relationship,
    sa,
    sa_exc,
    sa_orm,
)
from .email_address import EmailAddress, EmailAddressMixin
from .helpers import (
    RESERVED_NAMES,
    ImgeeType,
    MarkdownCompositeDocument,
    add_search_trigger,
    quote_autocomplete_like,
    quote_autocomplete_tsquery,
    valid_account_name,
    visual_field_delimiter,
)
from .phone_number import PhoneNumber, PhoneNumberMixin

__all__ = [
    'ACCOUNT_STATE',
    'Account',
    'AccountEmail',
    'AccountEmailClaim',
    'AccountExternalId',
    'AccountOldId',
    'AccountPhone',
    'Anchor',
    'Community',
    'deleted_account',
    'DuckTypeAccount',
    'Organization',
    'Placeholder',
    'removed_account',
    'Team',
    'unknown_account',
    'User',
]


class ACCOUNT_STATE(LabeledEnum):  # noqa: N801
    """State codes for accounts."""

    #: Regular, active account
    ACTIVE = (1, __("Active"))
    #: Suspended account (cause and explanation not included here)
    SUSPENDED = (2, __("Suspended"))
    #: Merged into another account
    MERGED = (3, __("Merged"))
    #: Permanently deleted account
    DELETED = (5, __("Deleted"))

    #: This account is gone
    GONE = {MERGED, DELETED}


class PROFILE_STATE(LabeledEnum):  # noqa: N801
    """The visibility state of an account (auto/public/private)."""

    AUTO = (1, 'auto', __("Autogenerated"))
    PUBLIC = (2, 'public', __("Public"))
    PRIVATE = (3, 'private', __("Private"))

    NOT_PUBLIC = {AUTO, PRIVATE}
    NOT_PRIVATE = {PUBLIC}


class ZBase32Comparator(Comparator[str]):  # pylint: disable=abstract-method
    """Comparator to allow lookup by Account.uuid_zbase32."""

    def __eq__(self, other: object) -> sa.ColumnElement[bool]:  # type: ignore[override]
        """Return an expression for column == other."""
        try:
            return self.__clause_element__() == UUID(  # type: ignore[return-value]
                bytes=zbase32_decode(str(other))
            )
        except ValueError:  # zbase32 call failed, so it's not a valid string
            return sa.false()


# --- Tables ---------------------------------------------------------------------------

team_membership = sa.Table(
    'team_membership',
    Model.metadata,
    sa.Column(
        'account_id',
        sa.Integer,
        sa.ForeignKey('account.id'),
        nullable=False,
        primary_key=True,
    ),
    sa.Column(
        'team_id',
        sa.Integer,
        sa.ForeignKey('team.id'),
        nullable=False,
        primary_key=True,
    ),
    sa.Column(
        'created_at',
        sa.TIMESTAMP(timezone=True),
        nullable=False,
        default=sa.func.utcnow(),
    ),
)


# --- Models ---------------------------------------------------------------------------


class Account(UuidMixin, BaseMixin[int, 'Account'], Model):
    """Account model."""

    __tablename__ = 'account'
    # Name has a length limit 63 to fit DNS label limit
    __name_length__ = 63
    # Titles can be longer
    __title_length__ = 80

    __active_membership_attrs__: ClassVar[set[str]] = set()
    __noninvite_membership_attrs__: ClassVar[set[str]] = set()

    # Helper flags (see subclasses)
    is_user_profile: ClassVar[bool] = False
    is_organization_profile: ClassVar[bool] = False
    is_community_profile: ClassVar[bool] = False
    is_placeholder_profile: ClassVar[bool] = False

    reserved_names: ClassVar[set[str]] = RESERVED_NAMES

    type_: Mapped[str] = sa_orm.mapped_column('type', sa.CHAR(1), nullable=False)

    #: Join date for users and organizations (skipped for placeholders)
    joined_at: Mapped[datetime | None] = sa_orm.mapped_column(
        sa.TIMESTAMP(timezone=True), nullable=True
    )

    #: The optional "username", used in the URL stub, with a unique constraint on the
    #: lowercase value (defined in __table_args__ below)
    name: Mapped[str | None] = with_roles(
        sa_orm.mapped_column(
            sa.Unicode(__name_length__),
            sa.CheckConstraint("name <> ''"),
            nullable=True,
        ),
        read={'all'},
    )

    #: The account's title (user's fullname)
    title: Mapped[str] = with_roles(
        sa_orm.mapped_column(sa.Unicode(__title_length__), default='', nullable=False),
        read={'all'},
    )
    #: Alias title as user's fullname
    fullname: Mapped[str] = sa_orm.synonym('title')
    #: Alias name as user's username
    username: Mapped[str | None] = sa_orm.synonym('name')

    #: Argon2 or Bcrypt hash of the user's password
    pw_hash: Mapped[str | None] = sa_orm.mapped_column()
    #: Timestamp for when the user's password last changed
    pw_set_at: Mapped[datetime | None] = sa_orm.mapped_column(
        sa.TIMESTAMP(timezone=True), nullable=True
    )
    #: Expiry date for the password (to prompt user to reset it)
    pw_expires_at: Mapped[datetime | None] = sa_orm.mapped_column(
        sa.TIMESTAMP(timezone=True), nullable=True
    )
    #: User's preferred/last known timezone
    timezone: Mapped[BaseTzInfo | None] = with_roles(
        sa_orm.mapped_column(TimezoneType(backend='pytz'), nullable=True),
        read={'owner'},
    )
    #: Update timezone automatically from browser activity
    auto_timezone: Mapped[bool] = sa_orm.mapped_column(default=True)
    #: User's preferred/last known locale
    locale: Mapped[Locale | None] = with_roles(
        sa_orm.mapped_column(LocaleType, nullable=True), read={'owner'}
    )
    #: Update locale automatically from browser activity
    auto_locale: Mapped[bool] = sa_orm.mapped_column(default=True)
    #: User's state code (active, suspended, merged, deleted)
    _state: Mapped[int] = sa_orm.mapped_column(
        'state',
        sa.SmallInteger,
        StateManager.check_constraint('state', ACCOUNT_STATE, sa.SmallInteger),
        nullable=False,
        default=ACCOUNT_STATE.ACTIVE,
    )
    #: Account state manager
    state = StateManager['Account']('_state', ACCOUNT_STATE, doc="Account state")
    #: Other accounts that were merged into this account
    old_accounts: AssociationProxy[list[Account]] = association_proxy(
        'oldids', 'old_account'
    )

    _profile_state: Mapped[int] = sa_orm.mapped_column(
        'profile_state',
        sa.SmallInteger,
        StateManager.check_constraint('profile_state', PROFILE_STATE, sa.SmallInteger),
        nullable=False,
        default=PROFILE_STATE.AUTO,
    )
    profile_state = StateManager['Account'](
        '_profile_state', PROFILE_STATE, doc="Current state of the account profile"
    )

    tagline: Mapped[str | None] = sa_orm.mapped_column(
        sa.CheckConstraint("tagline <> ''")
    )
    description, description_text, description_html = MarkdownCompositeDocument.create(
        'description', default='', nullable=False
    )
    website: Mapped[furl | None] = sa_orm.mapped_column(
        UrlType, sa.CheckConstraint("website <> ''"), nullable=True
    )
    logo_url: Mapped[furl | None] = sa_orm.mapped_column(
        ImgeeType, sa.CheckConstraint("logo_url <> ''"), nullable=True
    )
    banner_image_url: Mapped[furl | None] = sa_orm.mapped_column(
        ImgeeType, sa.CheckConstraint("banner_image_url <> ''"), nullable=True
    )

    #: Protected accounts cannot be deleted
    is_protected: Mapped[bool] = with_roles(
        immutable(sa_orm.mapped_column(default=False)),
        read={'owner', 'admin'},
    )
    #: Verified accounts get listed on the home page and are not considered throwaway
    #: accounts for spam control. There are no other privileges at this time
    is_verified: Mapped[bool] = with_roles(
        sa_orm.mapped_column(default=False, index=True),
        read={'all'},
    )

    #: Revision number maintained by SQLAlchemy, starting at 1
    revisionid: Mapped[int] = with_roles(sa_orm.mapped_column(), read={'all'})

    search_vector: Mapped[str] = sa_orm.mapped_column(
        TSVectorType(
            'title',
            'name',
            'tagline',
            'description_text',
            weights={
                'title': 'A',
                'name': 'A',
                'tagline': 'B',
                'description_text': 'B',
            },
            regconfig='english',
            hltext=lambda: sa.func.concat_ws(
                visual_field_delimiter,
                Account.title,
                Account.name,
                Account.tagline,
                Account.description_html,
            ),
        ),
        nullable=False,
        deferred=True,
    )

    name_vector: Mapped[str] = sa_orm.mapped_column(
        TSVectorType(
            'title',
            'name',
            regconfig='simple',
            hltext=lambda: sa.func.concat_ws(' @', Account.title, Account.name),
        ),
        nullable=False,
        deferred=True,
    )

    # --- Backrefs

    # account.py:
    oldid: Mapped[AccountOldId] = relationship(
        primaryjoin=lambda: sa_orm.foreign(AccountOldId.id) == Account.uuid,
        uselist=False,
        back_populates='old_account',
    )
    oldids: Mapped[list[AccountOldId]] = relationship(
        foreign_keys=lambda: AccountOldId.account_id, back_populates='account'
    )
    teams: Mapped[list[Team]] = relationship(
        foreign_keys=lambda: Team.account_id,
        order_by=lambda: sa.func.lower(Team.title),
        back_populates='account',
    )
    member_teams: Mapped[list[Team]] = relationship(
        secondary='team_membership', back_populates='users'
    )
    emails: Mapped[list[AccountEmail]] = relationship(back_populates='account')
    emailclaims: Mapped[list[AccountEmailClaim]] = relationship(
        back_populates='account'
    )
    phones: Mapped[list[AccountPhone]] = relationship(back_populates='account')
    externalids: Mapped[list[AccountExternalId]] = relationship(
        back_populates='account'
    )

    # account_membership.py
    memberships: DynamicMapped[AccountMembership] = relationship(
        foreign_keys=lambda: AccountMembership.account_id,
        lazy='dynamic',
        passive_deletes=True,
        back_populates='account',
    )
    active_follower_memberships: DynamicMapped[AccountMembership] = with_roles(
        relationship(
            lazy='dynamic',
            primaryjoin=lambda: sa.and_(
                sa_orm.remote(AccountMembership.account_id) == Account.id,
                AccountMembership.is_active,
            ),
            order_by=lambda: AccountMembership.granted_at.asc(),
            viewonly=True,
        ),
        read={'reader'},
        # Use offered_roles to determine which roles the user gets
        grants_via={
            'member': {
                'follower': 'follower',
                'member': 'member',
                'admin': 'admin',
                'owner': 'owner',
            }
        },
    )
    active_admin_memberships: DynamicMapped[AccountMembership] = relationship(
        lazy='dynamic',
        primaryjoin=lambda: sa.and_(
            sa_orm.remote(AccountMembership.account_id) == Account.id,
            AccountMembership.is_active,
            AccountMembership.is_admin.is_(True),
        ),
        order_by=lambda: AccountMembership.granted_at.asc(),
        viewonly=True,
    )

    active_owner_memberships: DynamicMapped[AccountMembership] = relationship(
        lazy='dynamic',
        primaryjoin=lambda: sa.and_(
            sa_orm.remote(AccountMembership.account_id) == Account.id,
            AccountMembership.is_active,
            AccountMembership.is_owner.is_(True),
        ),
        viewonly=True,
    )

    active_invitations: DynamicMapped[AccountMembership] = relationship(
        lazy='dynamic',
        primaryjoin=lambda: sa.and_(
            sa_orm.remote(AccountMembership.account_id) == Account.id,
            AccountMembership.is_invite,
            AccountMembership.revoked_at.is_(None),
        ),
        viewonly=True,
    )

    owner_users = with_roles(
        DynamicAssociationProxy['Account']('active_owner_memberships', 'member'),
        read={'all'},
    )
    admin_users = with_roles(
        DynamicAssociationProxy['Account']('active_admin_memberships', 'member'),
        read={'all'},
    )
    followers = with_roles(
        DynamicAssociationProxy['Account']('active_follower_memberships', 'member'),
        read={'reader'},
    )

    organization_admin_memberships: DynamicMapped[AccountMembership] = relationship(
        lazy='dynamic',
        foreign_keys=lambda: AccountMembership.member_id,
        viewonly=True,
    )

    noninvite_organization_admin_memberships: DynamicMapped[AccountMembership] = (
        relationship(
            lazy='dynamic',
            foreign_keys=lambda: AccountMembership.member_id,
            primaryjoin=lambda: sa.and_(
                sa_orm.remote(AccountMembership.member_id) == Account.id,
                ~AccountMembership.is_invite,
            ),
            viewonly=True,
        )
    )
    active_organization_admin_memberships: DynamicMapped[AccountMembership] = (
        relationship(
            lazy='dynamic',
            foreign_keys=lambda: AccountMembership.member_id,
            primaryjoin=lambda: sa.and_(
                sa_orm.remote(AccountMembership.member_id) == Account.id,
                AccountMembership.is_active,
            ),
            viewonly=True,
        )
    )
    active_organization_owner_memberships: DynamicMapped[AccountMembership] = (
        relationship(
            lazy='dynamic',
            foreign_keys=lambda: AccountMembership.member_id,
            primaryjoin=lambda: sa.and_(
                sa_orm.remote(AccountMembership.member_id) == Account.id,
                AccountMembership.is_active,
                AccountMembership.is_owner.is_(True),
            ),
            viewonly=True,
        )
    )
    active_organization_invitations: DynamicMapped[AccountMembership] = relationship(
        lazy='dynamic',
        foreign_keys=lambda: AccountMembership.member_id,
        primaryjoin=lambda: sa.and_(
            sa_orm.remote(AccountMembership.member_id) == Account.id,
            AccountMembership.is_invite,
            AccountMembership.revoked_at.is_(None),
        ),
        viewonly=True,
    )
    active_following_memberships: DynamicMapped[AccountMembership] = relationship(
        lazy='dynamic',
        primaryjoin=lambda: sa.and_(
            sa_orm.remote(AccountMembership.member_id) == Account.id,
            AccountMembership.is_active,
        ),
        order_by=lambda: AccountMembership.granted_at.asc(),
        viewonly=True,
    )

    organizations_as_owner = DynamicAssociationProxy['Account'](
        'active_organization_owner_memberships', 'account'
    )
    organizations_as_admin = DynamicAssociationProxy['Account'](
        'active_organization_admin_memberships', 'account'
    )
    accounts_following = with_roles(
        DynamicAssociationProxy['Account']('active_following_memberships', 'account'),
        read={'all'},
    )

    # auth_client.py
    clients: Mapped[AuthClient] = relationship(back_populates='account')
    authtokens: DynamicMapped[AuthToken] = relationship(
        lazy='dynamic', back_populates='account'
    )
    client_permissions: Mapped[list[AuthClientPermissions]] = relationship(
        back_populates='account'
    )

    # comment.py
    comments: DynamicMapped[Comment] = relationship(
        lazy='dynamic', back_populates='_posted_by'
    )
    # commentset_membership.py
    active_commentset_memberships: DynamicMapped[CommentsetMembership] = relationship(
        lazy='dynamic',
        primaryjoin='''and_(
            CommentsetMembership.member_id == Account.id,
            CommentsetMembership.is_active,
        )''',
        viewonly=True,
    )
    subscribed_commentsets = DynamicAssociationProxy['Commentset'](
        'active_commentset_memberships', 'commentset'
    )

    # contact_exchange.py
    scanned_contacts: DynamicMapped[ContactExchange] = relationship(
        lazy='dynamic',
        order_by='ContactExchange.scanned_at.desc()',
        passive_deletes=True,
        back_populates='account',
    )

    # login_session.py
    all_login_sessions: DynamicMapped[LoginSession] = relationship(
        lazy='dynamic', back_populates='account'
    )
    active_login_sessions: DynamicMapped[LoginSession] = relationship(
        lazy='dynamic',
        primaryjoin=lambda: sa.and_(
            LoginSession.account_id == Account.id,
            LoginSession.accessed_at > sa.func.utcnow() - LOGIN_SESSION_VALIDITY_PERIOD,
            LoginSession.revoked_at.is_(None),
        ),
        order_by=lambda: LoginSession.accessed_at.desc(),
        viewonly=True,
    )

    # mailer.py
    mailers: Mapped[list[Mailer]] = relationship(
        back_populates='user', order_by=lambda: Mailer.updated_at.desc()
    )

    # moderation.py
    moderator_reports: DynamicMapped[CommentModeratorReport] = relationship(
        lazy='dynamic', back_populates='reported_by'
    )

    # notification.py
    all_notifications: DynamicMapped[NotificationRecipient] = with_roles(
        relationship(
            lazy='dynamic',
            order_by=lambda: NotificationRecipient.created_at.desc(),
            viewonly=True,
        ),
        read={'owner'},
    )

    notification_preferences: Mapped[dict[str, NotificationPreferences]] = relationship(
        collection_class=sa_orm.attribute_keyed_dict('notification_type'),
        back_populates='account',
    )

    # This relationship is wrapped in a property that creates it on first access
    _main_notification_preferences: Mapped[NotificationPreferences] = relationship(
        primaryjoin=lambda: sa.and_(
            NotificationPreferences.account_id == Account.id,
            NotificationPreferences.notification_type == '',
        ),
        uselist=False,
        viewonly=True,
    )

    @cached_property
    def main_notification_preferences(self) -> NotificationPreferences:
        """Return user's main notification preferences, toggling transports on/off."""
        if not self._main_notification_preferences:
            main = NotificationPreferences(
                notification_type='',
                account=self,
                by_email=True,
                by_sms=True,
                by_webpush=False,
                by_telegram=False,
                by_whatsapp=False,
            )
            db.session.add(main)
            return main
        return self._main_notification_preferences

    # project_membership.py
    projects_as_crew_memberships: DynamicMapped[ProjectMembership] = relationship(
        lazy='dynamic',
        foreign_keys=lambda: ProjectMembership.member_id,
        viewonly=True,
    )

    # This is used to determine if it is safe to purge the subject's database record
    projects_as_crew_noninvite_memberships: DynamicMapped[ProjectMembership] = (
        relationship(
            lazy='dynamic',
            primaryjoin=lambda: sa.and_(
                ProjectMembership.member_id == Account.id,
                ~ProjectMembership.is_invite,
            ),
            viewonly=True,
        )
    )
    projects_as_crew_active_memberships: DynamicMapped[ProjectMembership] = (
        relationship(
            lazy='dynamic',
            primaryjoin=lambda: sa.and_(
                ProjectMembership.member_id == Account.id,
                ProjectMembership.is_active,
            ),
            viewonly=True,
        )
    )

    projects_as_crew = DynamicAssociationProxy['Project'](
        'projects_as_crew_active_memberships', 'project'
    )

    projects_as_editor_active_memberships: DynamicMapped[ProjectMembership] = (
        relationship(
            lazy='dynamic',
            primaryjoin=lambda: sa.and_(
                ProjectMembership.member_id == Account.id,
                ProjectMembership.is_active,
                ProjectMembership.is_editor.is_(True),
            ),
            viewonly=True,
        )
    )

    projects_as_editor = DynamicAssociationProxy['Project'](
        'projects_as_editor_active_memberships', 'project'
    )

    # project.py
    projects: DynamicMapped[Project] = relationship(
        lazy='dynamic',
        foreign_keys=lambda: Project.account_id,
        back_populates='account',
    )
    project_redirects: DynamicMapped[ProjectRedirect] = relationship(
        lazy='dynamic', back_populates='account'
    )

    listed_projects: DynamicMapped[Project] = relationship(
        lazy='dynamic',
        primaryjoin=lambda: sa.and_(
            Account.id == Project.account_id,
            Project.state.PUBLISHED,
        ),
        viewonly=True,
    )
    draft_projects: DynamicMapped[Project] = relationship(
        lazy='dynamic',
        primaryjoin=lambda: sa.and_(
            Account.id == Project.account_id,
            sa.or_(Project.state.DRAFT, Project.cfp_state.DRAFT),
        ),
        viewonly=True,
    )
    projects_by_name: Mapped[dict[str, Project]] = with_roles(
        relationship(
            foreign_keys=lambda: Project.account_id,
            collection_class=sa_orm.attribute_keyed_dict('name'),
            viewonly=True,
        ),
        read={'all'},
    )

    # proposal_membership.py
    all_proposal_memberships: DynamicMapped[ProposalMembership] = relationship(
        lazy='dynamic',
        foreign_keys=lambda: ProposalMembership.member_id,
        viewonly=True,
    )

    noninvite_proposal_memberships: DynamicMapped[ProposalMembership] = relationship(
        lazy='dynamic',
        primaryjoin=lambda: sa.and_(
            ProposalMembership.member_id == Account.id,
            ~ProposalMembership.is_invite,
        ),
        viewonly=True,
    )

    proposal_memberships: DynamicMapped[ProposalMembership] = relationship(
        lazy='dynamic',
        primaryjoin=lambda: sa.and_(
            ProposalMembership.member_id == Account.id,
            ProposalMembership.is_active,
        ),
        viewonly=True,
    )

    # This is a User property of the proposals the user account is a collaborator in
    proposals = DynamicAssociationProxy['Proposal']('proposal_memberships', 'proposal')

    @property
    def public_proposal_memberships(self) -> Query[ProposalMembership]:
        """Query for all proposal memberships to proposals that are public."""
        # TODO: Include proposal state filter (pending proposal workflow fix)
        return (
            self.proposal_memberships.join(Proposal, ProposalMembership.proposal)
            .join(Project, Proposal.project)
            .filter(ProposalMembership.is_uncredited.is_(False))
        )

    public_proposals = DynamicAssociationProxy['Proposal'](
        'public_proposal_memberships', 'proposal'
    )

    # proposal.py
    created_proposals: DynamicMapped[Proposal] = relationship(
        lazy='dynamic', back_populates='created_by'
    )

    # rsvp.py
    rsvps: DynamicMapped[Rsvp] = relationship(
        lazy='dynamic', back_populates='participant'
    )

    @property
    def rsvp_followers(self) -> Query[Account]:
        """All users with an active RSVP in a project."""
        return (
            Account.query.filter(Account.state.ACTIVE)
            .join(Rsvp, Rsvp.participant_id == Account.id)
            .join(Project, Rsvp.project_id == Project.id)
            .filter(Rsvp.state.YES, Project.state.PUBLISHED, Project.account == self)
        )

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

    # saved.py
    saved_projects: DynamicMapped[SavedProject] = relationship(
        lazy='dynamic', passive_deletes=True, back_populates='account'
    )
    saved_sessions: DynamicMapped[SavedSession] = relationship(
        lazy='dynamic', passive_deletes=True, back_populates='account'
    )

    def saved_sessions_in(self, project: Project) -> Query[SavedSession]:
        return self.saved_sessions.join(Session).filter(Session.project == project)

    # site_membership.py
    # Singular, as only one can be active
    active_site_membership: Mapped[SiteMembership] = relationship(
        lazy='select',
        primaryjoin=lambda: sa.and_(
            SiteMembership.member_id == Account.id, SiteMembership.is_active
        ),
        viewonly=True,
        uselist=False,
    )

    @cached_property
    def is_comment_moderator(self) -> bool:
        """Test if this user is a comment moderator."""
        return (
            self.active_site_membership is not None
            and self.active_site_membership.is_comment_moderator
        )

    @cached_property
    def is_user_moderator(self) -> bool:
        """Test if this user is an account moderator."""
        return (
            self.active_site_membership is not None
            and self.active_site_membership.is_user_moderator
        )

    @cached_property
    def is_site_editor(self) -> bool:
        """Test if this user is a site editor."""
        return (
            self.active_site_membership is not None
            and self.active_site_membership.is_site_editor
        )

    @cached_property
    def is_sysadmin(self) -> bool:
        """Test if this user is a sysadmin."""
        return (
            self.active_site_membership is not None
            and self.active_site_membership.is_sysadmin
        )

    # site_admin means user has one or more of above roles
    @cached_property
    def is_site_admin(self) -> bool:
        """Test if this user has any site-level admin rights."""
        return self.active_site_membership is not None

    # sponsor_membership.py
    noninvite_project_sponsor_memberships: DynamicMapped[ProjectSponsorMembership] = (
        relationship(
            lazy='dynamic',
            primaryjoin=lambda: sa.and_(
                ProjectSponsorMembership.member_id == Account.id,
                ~ProjectSponsorMembership.is_invite,
            ),
            order_by=lambda: ProjectSponsorMembership.granted_at.desc(),
            viewonly=True,
        )
    )

    project_sponsor_memberships: DynamicMapped[ProjectSponsorMembership] = relationship(
        lazy='dynamic',
        primaryjoin=lambda: sa.and_(
            ProjectSponsorMembership.member_id == Account.id,
            ProjectSponsorMembership.is_active,
        ),
        order_by=lambda: ProjectSponsorMembership.granted_at.desc(),
        viewonly=True,
    )

    project_sponsor_membership_invites: DynamicMapped[ProjectSponsorMembership] = (
        with_roles(
            relationship(
                lazy='dynamic',
                primaryjoin=lambda: sa.and_(
                    ProjectSponsorMembership.member_id == Account.id,
                    ProjectSponsorMembership.is_invite,
                    ProjectSponsorMembership.revoked_at.is_(None),
                ),
                order_by=lambda: ProjectSponsorMembership.granted_at.desc(),
                viewonly=True,
            ),
            read={'admin'},
        )
    )

    noninvite_proposal_sponsor_memberships: DynamicMapped[ProposalSponsorMembership] = (
        relationship(
            lazy='dynamic',
            primaryjoin=lambda: sa.and_(
                ProposalSponsorMembership.member_id == Account.id,
                ~ProposalSponsorMembership.is_invite,
            ),
            order_by=lambda: ProposalSponsorMembership.granted_at.desc(),
            viewonly=True,
        )
    )

    proposal_sponsor_memberships: DynamicMapped[ProposalSponsorMembership] = (
        relationship(
            lazy='dynamic',
            primaryjoin=lambda: sa.and_(
                ProposalSponsorMembership.member_id == Account.id,
                ProposalSponsorMembership.is_active,
            ),
            order_by=lambda: ProposalSponsorMembership.granted_at.desc(),
            viewonly=True,
        )
    )

    proposal_sponsor_membership_invites: DynamicMapped[ProposalSponsorMembership] = (
        with_roles(
            relationship(
                lazy='dynamic',
                primaryjoin=lambda: sa.and_(
                    ProposalSponsorMembership.member_id == Account.id,
                    ProposalSponsorMembership.is_invite,
                    ProposalSponsorMembership.revoked_at.is_(None),
                ),
                order_by=lambda: ProposalSponsorMembership.granted_at.desc(),
                viewonly=True,
            ),
            read={'admin'},
        )
    )

    sponsored_projects = DynamicAssociationProxy['Project'](
        'project_sponsor_memberships', 'project'
    )

    sponsored_proposals = DynamicAssociationProxy['Project'](
        'proposal_sponsor_memberships', 'proposal'
    )

    # sync_ticket.py:
    ticket_participants: Mapped[list[TicketParticipant]] = relationship(
        back_populates='participant'
    )

    @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'})

    # update.py
    created_updates: DynamicMapped[Update] = relationship(
        lazy='dynamic',
        foreign_keys=lambda: Update.created_by_id,
        back_populates='created_by',
    )
    published_updates: DynamicMapped[Update] = relationship(
        lazy='dynamic',
        foreign_keys=lambda: Update.published_by_id,
        back_populates='published_by',
    )
    deleted_updates: DynamicMapped[Update] = relationship(
        lazy='dynamic',
        foreign_keys=lambda: Update.deleted_by_id,
        back_populates='deleted_by',
    )

    __table_args__ = (
        sa.Index(
            'ix_account_name_lower',
            sa.func.lower(name).label('name_lower'),
            unique=True,
            postgresql_ops={'name_lower': 'varchar_pattern_ops'},
        ),
        sa.Index(
            'ix_account_title_lower',
            sa.func.lower(title).label('title_lower'),
            postgresql_ops={'title_lower': 'varchar_pattern_ops'},
        ),
        sa.Index('ix_account_search_vector', 'search_vector', postgresql_using='gin'),
        sa.Index('ix_account_name_vector', 'name_vector', postgresql_using='gin'),
    )

    __mapper_args__ = {
        # 'polymorphic_identity' from subclasses is stored in the type column
        'polymorphic_on': type_,
        # When querying the Account model, cast automatically to all subclasses
        'with_polymorphic': '*',
        # Store a version id in this column to prevent edits to obsolete data
        'version_id_col': revisionid,
    }

    __roles__ = {
        'all': {
            'read': {
                'uuid',
                'name',
                'urlname',
                'title',
                'fullname',
                'username',
                'pickername',
                'timezone',
                'description',
                'website',
                'logo_url',
                'banner_image_url',
                'joined_at',
                'absolute_url',
                'urls',
                'is_user_profile',
                'is_organization_profile',
                'is_placeholder_profile',
            },
            'call': {'views', 'forms', 'features', 'url_for', 'state', 'profile_state'},
        }
    }

    __datasets__ = {
        'primary': {
            'urls',
            'uuid_b58',
            'name',
            'urlname',
            'title',
            'fullname',
            'username',
            'pickername',
            'timezone',
            'description',
            'logo_url',
            'website',
            'joined_at',
            'absolute_url',
            'is_verified',
        },
        'related': {
            'urls',
            'uuid_b58',
            'name',
            'urlname',
            'title',
            'fullname',
            'username',
            'pickername',
            'timezone',
            'description',
            'logo_url',
            'joined_at',
            'absolute_url',
            'is_verified',
        },
    }
    __json_datasets__ = ('primary', 'related')

    profile_state.add_conditional_state(
        'ACTIVE_AND_PUBLIC',
        profile_state.PUBLIC,
        lambda account: bool(account.state.ACTIVE),
    )

    @classmethod
    def _defercols(cls) -> list[sa_orm.interfaces.LoaderOption]:
        """Return columns that are typically deferred when loading a user."""
        defer = sa_orm.defer
        return [
            defer(cls.created_at),
            defer(cls.updated_at),
            defer(cls.pw_hash),
            defer(cls.pw_set_at),
            defer(cls.pw_expires_at),
            defer(cls.timezone),
        ]

    @classmethod
    def type_filter(cls) -> sa.ColumnElement[bool]:
        """Return filter for the subclass's type."""
        return cls.type_ == cls.__mapper_args__.get('polymorphic_identity')

    if TYPE_CHECKING:
        # These are added via add_primary_relationship
        primary_email: Mapped[AccountEmail | None] = relationship()
        primary_phone: Mapped[AccountPhone | None] = relationship()

    def __repr__(self) -> str:
        if self.name:
            return f'<{self.__class__.__name__} {self.title} @{self.name}>'
        return f'<{self.__class__.__name__} {self.title}>'

    def __str__(self) -> str:
        """Return picker name for account."""
        return self.pickername

    def __format__(self, format_spec: str) -> str:
        if not format_spec:
            return self.pickername
        return format(self.pickername, format_spec)

    @property
    def pickername(self) -> str:
        """Return title and @name in a format suitable for identification."""
        if self.name:
            return f'{self.title} (@{self.name})'
        return self.title

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

    @role_check('reader')
    def has_reader_role(
        self, _actor: Account | None, _anchors: Sequence[Any] = ()
    ) -> bool:
        """Grant 'reader' role to all if the profile state is active and public."""
        return bool(self.profile_state.ACTIVE_AND_PUBLIC)

    @cached_property
    def verified_contact_count(self) -> int:
        """Count of verified contact details."""
        return len(self.emails) + len(self.phones)

    @property
    def has_verified_contact_info(self) -> bool:
        """User has any verified contact info (email or phone)."""
        return bool(self.emails) or bool(self.phones)

    @property
    def has_contact_info(self) -> bool:
        """User has any contact information (including unverified)."""
        return self.has_verified_contact_info or bool(self.emailclaims)

    def merged_account(self) -> Account:
        """Return the account that this account was merged into (default: self)."""
        if self.state.MERGED:
            # If our state is MERGED, there _must_ be a corresponding AccountOldId
            # record
            return cast(AccountOldId, AccountOldId.get(self.uuid)).account
        return self

    def _set_password(self, password: str | None) -> None:
        """Set a password (write-only property)."""
        if password is None:
            self.pw_hash = None
        else:
            self.pw_hash = argon2.hash(password)
            # Also see :meth:`password_is` for transparent upgrade
        self.pw_set_at = sa.func.utcnow()
        # Expire passwords after one year. TODO: make this configurable
        self.pw_expires_at = sa.func.utcnow() + sa.cast('1 year', sa.Interval)

    #: Write-only property (passwords cannot be read back in plain text)
    password = property(fset=_set_password, doc=_set_password.__doc__)

    def password_has_expired(self) -> bool:
        """Verify if password expiry timestamp has passed."""
        return (
            self.pw_hash is not None
            and self.pw_expires_at is not None
            and self.pw_expires_at <= utcnow()
        )

    def password_is(self, password: str, upgrade_hash: bool = False) -> bool:
        """Test if the candidate password matches saved hash."""
        if self.pw_hash is None:
            return False

        # Passwords may use the current Argon2 scheme or the older Bcrypt scheme.
        # Bcrypt passwords are transparently upgraded if requested.
        if argon2.identify(self.pw_hash):
            return argon2.verify(password, self.pw_hash)
        if bcrypt.identify(self.pw_hash):
            verified = bcrypt.verify(password, self.pw_hash)
            if verified and upgrade_hash:
                self.pw_hash = argon2.hash(password)
            return verified
        return False

    def add_email(
        self,
        email: str,
        primary: bool | None = None,
        private: bool = False,
    ) -> AccountEmail:
        """
        Add an email address (assumed to be verified).

        :param email: Email address as a string
        :param primary: Mark this email address as primary (default: auto-assign)
        :param private: Mark as private (currently unused)
        """
        accountemail = AccountEmail(account=self, email=email, private=private)
        accountemail = failsafe_add(
            db.session,
            accountemail,
            account=self,
            email_address=accountemail.email_address,
        )
        if (primary is None and self.primary_email is None) or primary is True:
            self.primary_email = accountemail
        return accountemail
        # FIXME: This should remove competing instances of AccountEmailClaim

    def del_email(self, email: str) -> None:
        """Remove an email address from the user's account."""
        accountemail = AccountEmail.get_for(account=self, email=email)
        if accountemail is not None:
            if self.primary_email in (accountemail, None):
                self.primary_email = (
                    AccountEmail.query.filter(
                        AccountEmail.account == self, AccountEmail.id != accountemail.id
                    )
                    .order_by(AccountEmail.created_at.desc())
                    .first()
                )
            db.session.delete(accountemail)

    @property
    def email(self) -> Literal[''] | AccountEmail:
        """Return primary email address for user."""
        # Look for a primary address
        accountemail = self.primary_email
        if accountemail is not None:
            return accountemail
        # No primary? Maybe there's one that's not set as primary?
        if self.emails:
            accountemail = self.emails[0]
            # XXX: Mark as primary. This may or may not be saved depending on
            # whether the request ended in a database commit.
            self.primary_email = accountemail
            return accountemail
        # This user has no email address. Return a blank string instead of None
        # to support the common use case, where the caller will use str(user.email)
        # to get the email address as a string.
        return ''

    with_roles(email, read={'owner'})

    def add_phone(
        self,
        phone: str,
        primary: bool | None = None,
        private: bool = False,
    ) -> AccountPhone:
        """
        Add a phone number (assumed to be verified).

        :param phone: Phone number as a string
        :param primary: Mark this phone number as primary (default: auto-assign)
        :param private: Mark as private (currently unused)
        """
        accountphone = AccountPhone(account=self, phone=phone, private=private)
        accountphone = failsafe_add(
            db.session,
            accountphone,
            account=self,
            phone_number=accountphone.phone_number,
        )
        if (primary is None and self.primary_phone is None) or primary is True:
            self.primary_phone = accountphone
        return accountphone

    def del_phone(self, phone: str) -> None:
        """Remove a phone number from the user's account."""
        accountphone = AccountPhone.get_for(account=self, phone=phone)
        if accountphone is not None:
            if self.primary_phone in (accountphone, None):
                self.primary_phone = (
                    AccountPhone.query.filter(
                        AccountPhone.account == self, AccountPhone.id != accountphone.id
                    )
                    .order_by(AccountPhone.created_at.desc())
                    .first()
                )
            db.session.delete(accountphone)

    @property
    def phone(self) -> Literal[''] | AccountPhone:
        """Return primary phone number for user."""
        # Look for a primary phone number
        accountphone = self.primary_phone
        if accountphone is not None:
            return accountphone
        # No primary? Maybe there's one that's not set as primary?
        if self.phones:
            accountphone = self.phones[0]
            # XXX: Mark as primary. This may or may not be saved depending on
            # whether the request ended in a database commit.
            self.primary_phone = accountphone
            return accountphone
        # This user has no phone number. Return a blank string instead of None
        # to support the common use case, where the caller will use str(user.phone)
        # to get the phone number as a string.
        return ''

    with_roles(phone, read={'owner'})

    @property
    def has_public_profile(self) -> bool:
        """Return the visibility state of an account."""
        return self.name is not None and bool(self.profile_state.ACTIVE_AND_PUBLIC)

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

    def is_profile_complete(self) -> bool:
        """Verify if profile is complete (fullname, username and contacts present)."""
        return bool(self.title and self.name and self.has_verified_contact_info)

    def active_memberships(self) -> Iterator[ImmutableMembershipMixin]:
        """Enumerate all active memberships."""
        # Each collection is cast into a list before chaining to ensure that it does not
        # change during processing (if, for example, membership is revoked or replaced).
        return itertools.chain(
            *(list(getattr(self, attr)) for attr in self.__active_membership_attrs__)
        )

    def has_any_memberships(self) -> bool:
        """
        Test for any non-invite membership records that must be preserved.

        This is used to test for whether the account is safe to purge (hard delete) from
        the database. If non-invite memberships are present, the account cannot be
        purged as immutable records must be preserved. Instead, the account must be put
        into DELETED state with all PII scrubbed.
        """
        return any(
            db.session.query(getattr(self, attr).exists()).scalar()
            for attr in self.__noninvite_membership_attrs__
        )

    # --- Transport details

    @with_roles(call={'owner'})
    def has_transport_email(self) -> bool:
        """User has an email transport address."""
        return bool(self.state.ACTIVE) and bool(self.email)

    @with_roles(call={'owner'})
    def has_transport_sms(self) -> bool:
        """User has an SMS transport address."""
        return (
            bool(self.state.ACTIVE)
            and self.phone != ''
            and self.phone.phone_number.has_sms is not False
        )

    @with_roles(call={'owner'})
    def has_transport_webpush(self) -> bool:  # TODO  # pragma: no cover
        """User has a webpush transport address."""
        return False

    @with_roles(call={'owner'})
    def has_transport_telegram(self) -> bool:  # TODO  # pragma: no cover
        """User has a Telegram transport address."""
        return False

    @with_roles(call={'owner'})
    def has_transport_whatsapp(self) -> bool:
        """User has a WhatsApp transport address."""
        return (
            bool(self.state.ACTIVE)
            and self.phone != ''
            and self.phone.phone_number.has_wa is not False
        )

    @with_roles(call={'owner'})
    def transport_for_email(self, context: Model | None = None) -> AccountEmail | None:
        """Return user's preferred email address within a context."""
        # TODO: Per-account/project customization is a future option
        if self.state.ACTIVE:
            return self.email or None
        return None

    @with_roles(call={'owner'})
    def transport_for_sms(self, context: Model | None = None) -> AccountPhone | None:
        """Return user's preferred phone number within a context."""
        # TODO: Per-account/project customization is a future option
        if (
            self.state.ACTIVE
            and self.phone != ''
            and self.phone.phone_number.has_sms is not False
        ):
            return self.phone
        return None

    @with_roles(call={'owner'})
    def transport_for_webpush(
        self, context: Model | None = None
    ) -> None:  # TODO  # pragma: no cover
        """Return user's preferred webpush transport address within a context."""
        return None

    @with_roles(call={'owner'})
    def transport_for_telegram(
        self, context: Model | None = None
    ) -> None:  # TODO  # pragma: no cover
        """Return user's preferred Telegram transport address within a context."""
        return None

    @with_roles(call={'owner'})
    def transport_for_whatsapp(
        self, context: Model | None = None
    ) -> AccountPhone | None:
        """Return user's preferred WhatsApp transport address within a context."""
        # TODO: Per-account/project customization is a future option
        if self.state.ACTIVE and self.phone != '' and self.phone.phone_number.has_wa:
            return self.phone
        return None

    @with_roles(call={'owner'})
    def has_transport(self, transport: str) -> bool:
        """
        Verify if user has a given transport address.

        Helper method to call ``self.has_transport_<transport>()``.

        ..note::
            Because this method does not accept a context, it may return True for a
            transport that has been muted in that context. This may cause an empty
            background job to be queued for a notification. Revisit this method when
            preference contexts are supported.
        """
        return getattr(self, 'has_transport_' + transport)()

    @with_roles(call={'owner'})
    def transport_for(
        self, transport: str, context: Model | None = None
    ) -> AccountEmail | AccountPhone | None:
        """
        Get transport address for a given transport and context.

        Helper method to call ``self.transport_for_<transport>(context)``.
        """
        return getattr(self, 'transport_for_' + transport)(context)

    def default_email(
        self, context: Model | None = None
    ) -> AccountEmail | AccountEmailClaim | None:
        """
        Return default email address (verified if present, else unverified).

        ..note::
            This is a temporary helper method, pending merger of
            :class:`AccountEmailClaim` into :class:`AccountEmail` with
            :attr:`~AccountEmail.verified` ``== False``. The appropriate replacement is
            :meth:`Account.transport_for_email` with a context.
        """
        email = self.transport_for_email(context=context)
        if email:
            return email
        # Fallback when ``transport_for_email`` returns None
        if self.email:
            return self.email
        if self.emailclaims:
            return self.emailclaims[0]
        # This user has no email addresses
        return None

    @property
    def _self_is_owner_of_self(self) -> Account | None:
        """
        Return self in a user account.

        Helper method for :meth:`roles_for` and :meth:`actors_with` to assert that the
        user is owner and admin of their own account.
        """
        return self if self.is_user_profile else None

    with_roles(
        _self_is_owner_of_self,
        grants={'follower', 'member', 'admin', 'owner'},
    )

    def organizations_as_owner_ids(self) -> list[int]:
        """
        Return the database ids of the organizations this user is an owner of.

        This is used for database queries.
        """
        return [
            membership.account_id
            for membership in self.active_organization_owner_memberships
        ]

    @state.transition(state.ACTIVE, state.MERGED)
    def mark_merged_into(self, other_account: Account) -> None:
        """Mark account as merged into another account."""
        db.session.add(AccountOldId(id=self.uuid, account=other_account))

    @state.transition(state.ACTIVE, state.SUSPENDED)
    def mark_suspended(self) -> None:
        """Mark account as suspended on support or moderator request."""

    @state.transition(state.SUSPENDED, state.ACTIVE)
    def mark_active(self) -> None:
        """Restore a suspended account to active state."""

    @state.transition(state.ACTIVE, state.DELETED)
    def do_delete(self) -> None:
        """Delete account."""
        # 0: Safety check
        if not self.is_safe_to_delete():
            raise ValueError("Account cannot be deleted")

        # 1. Delete contact information
        for contact_source in cast(
            list,
            (
                self.emails,
                self.emailclaims,
                self.phones,
                self.externalids,
            ),
        ):
            for contact in contact_source:
                db.session.delete(contact)

        # 2. Revoke all active memberships
        for membership in self.active_memberships():
            if callable(
                freeze := getattr(membership, 'freeze_member_attribution', None)
            ):
                membership = freeze(self)
            if membership.revoke_on_member_delete:
                membership.revoke(actor=self)
        # TODO: freeze fullname in unrevoked memberships (pending title column there)
        if (
            self.active_site_membership
            and self.active_site_membership.revoke_on_member_delete
        ):
            self.active_site_membership.revoke(actor=self)

        # 3. Drop all team memberships
        self.member_teams.clear()

        # 4. Revoke auth tokens
        AuthToken.revoke_all_for(self)
        AuthClientPermissions.revoke_all_for(self)

        # 5. Revoke all active login sessions
        for login_session in self.active_login_sessions:
            login_session.revoke()

        # 6. Clear name (username), title (fullname) and stored password hash
        self.name = None
        self.title = ''
        self.password = None

        # 7. Unassign tickets assigned to the user
        self.ticket_participants = []  # pylint: disable=attribute-defined-outside-init

    @with_roles(call={'owner'})
    @profile_state.transition(
        profile_state.NOT_PUBLIC,
        profile_state.PUBLIC,
        title=__("Make public"),
    )
    @state.requires(state.ACTIVE)
    def make_profile_public(self) -> None:
        """Make an account public if it is eligible."""

    @with_roles(call={'owner'})
    @profile_state.transition(
        profile_state.NOT_PRIVATE, profile_state.PRIVATE, title=__("Make private")
    )
    def make_profile_private(self) -> None:
        """Make an account private."""

    def is_safe_to_delete(self) -> bool:
        """Test if account is not protected and has no projects."""
        return self.is_protected is False and self.projects.count() == 0

    def is_safe_to_purge(self) -> bool:
        """Test if account is safe to delete and has no memberships (active or not)."""
        return self.is_safe_to_delete() and not self.has_any_memberships()

    @property
    def urlname(self) -> str:
        """Return :attr:`name` or ``~``-prefixed :attr:`uuid_zbase32`."""
        if self.name is not None:
            return self.name
        return f'~{self.uuid_zbase32}'

    @hybrid_property
    def uuid_zbase32(self) -> str:
        """Account UUID rendered in z-Base-32."""
        return zbase32_encode(self.uuid.bytes)

    @uuid_zbase32.inplace.comparator
    @classmethod
    def _uuid_zbase32_comparator(cls) -> ZBase32Comparator:
        """Return SQL comparator for :prop:`uuid_zbase32`."""
        return ZBase32Comparator(cls.uuid)  # type: ignore[arg-type]

    @classmethod
    def name_is(cls, name: str) -> ColumnElement:
        """Generate query filter to check if name is matching (case insensitive)."""
        if name.startswith('~'):
            return cls.uuid_zbase32 == name[1:]
        return sa.func.lower(cls.name) == sa.func.lower(sa.func.replace(name, '-', '_'))

    @classmethod
    def name_in(cls, names: Iterable[str]) -> ColumnElement:
        """Generate query filter to check if name is among candidates."""
        return sa.func.lower(cls.name).in_(
            [name.lower().replace('-', '_') for name in names]
        )

    @classmethod
    def name_like(cls, like_query: str) -> ColumnElement:
        """Generate query filter for a LIKE query on name."""
        return sa.func.lower(cls.name).like(
            sa.func.lower(sa.func.replace(like_query, '-', r'\_'))
        )

    @overload
    @classmethod
    def get(
        cls,
        *,
        name: str,
        defercols: bool = False,
    ) -> Account | None: ...

    @overload
    @classmethod
    def get(
        cls,
        *,
        buid: str,
        defercols: bool = False,
    ) -> Account | None: ...

    @overload
    @classmethod
    def get(
        cls,
        *,
        userid: str,
        defercols: bool = False,
    ) -> Account | None: ...

    @classmethod
    def get(
        cls,
        *,
        name: str | None = None,
        buid: str | None = None,
        userid: str | None = None,
        defercols: bool = False,
    ) -> Account | None:
        """
        Return an Account with the given name or buid.

        :param str name: Username to lookup
        :param str buid: Buid to lookup
        :param bool defercols: Defer loading non-critical columns
        """
        require_one_of(name=name, buid=buid, userid=userid)

        # userid parameter is temporary for Flask-Lastuser compatibility
        if userid:
            buid = userid

        if name is not None:
            query = cls.query.filter(cls.name_is(name))
        else:
            query = cls.query.filter_by(buid=buid)
        if cls is not Account:
            query = query.filter(cls.type_filter())
        if defercols:
            query = query.options(*cls._defercols())
        account = query.one_or_none()
        if account and account.state.MERGED:
            account = account.merged_account()
        if account and account.state.ACTIVE:
            return account
        return None

    @classmethod
    def all(  # noqa: A003
        cls,
        buids: Iterable[str] | None = None,
        names: Iterable[str] | None = None,
        defercols: bool = False,
    ) -> list[Account]:
        """
        Return all matching accounts.

        :param list buids: Buids to look up
        :param list names: Names (usernames) to look up
        :param bool defercols: Defer loading non-critical columns
        """
        accounts = set()
        if buids and names:
            query = cls.query.filter(sa.or_(cls.buid.in_(buids), cls.name_in(names)))
        elif buids:
            query = cls.query.filter(cls.buid.in_(buids))
        elif names:
            query = cls.query.filter(cls.name_in(names))
        else:
            return []
        if cls is not Account:
            query = query.filter(cls.type_filter())

        if defercols:
            query = query.options(*cls._defercols())
        for account in query.all():
            account = account.merged_account()
            if account.state.ACTIVE:
                accounts.add(account)
        return list(accounts)

    @classmethod
    def all_public(cls) -> Query:
        """Construct a query filtered by public profile state."""
        query = cls.query.filter(cls.profile_state.PUBLIC)
        if cls is not Account:
            query = query.filter(cls.type_filter())
        return query

    @classmethod
    def autocomplete(cls, prefix: str) -> list[Self]:
        """
        Return accounts whose names begin with the prefix, for autocomplete UI.

        Looks up accounts by title, name, external ids and email addresses.

        :param prefix: Letters to start matching with
        """
        like_query = quote_autocomplete_like(prefix)
        if not like_query or like_query == '@%':
            return []
        tsquery = quote_autocomplete_tsquery(prefix)

        # base_users is used in two of the three possible queries below
        base_users = cls.query.filter(
            cls.state.ACTIVE,
            cls.name_vector.bool_op('@@')(tsquery),
        )

        if cls is not Account:
            base_users = base_users.filter(cls.type_filter())
        base_users = (
            base_users.options(*cls._defercols()).order_by(Account.title).limit(20)
        )

        if (
            prefix != '@'
            and prefix.startswith('@')
            and AccountExternalId.__at_username_services__
        ):
            # @-prefixed, so look for usernames, including other @username-using
            # services like Twitter and GitHub. Make a union of three queries.
            users = (
                # Query 1: @query -> Account.name
                cls.query.filter(
                    cls.state.ACTIVE,
                    cls.name_like(like_query[1:]),
                )
                .options(*cls._defercols())
                .limit(20)
                # FIXME: Still broken as of SQLAlchemy 1.4.23 (also see next block)
                # .union(
                #     # Query 2: @query -> UserExternalId.username
                #     cls.query.join(UserExternalId)
                #     .filter(
                #         cls.state.ACTIVE,
                #         UserExternalId.service.in_(
                #             UserExternalId.__at_username_services__
                #         ),
                #         sa.func.lower(UserExternalId.username).like(
                #             sa.func.lower(like_query[1:])
                #         ),
                #     )
                #     .options(*cls._defercols())
                #     .limit(20),
                #     # Query 3: like_query -> Account.title
                #     cls.query.filter(
                #         cls.state.ACTIVE,
                #         sa.func.lower(cls.title).like(sa.func.lower(like_query)),
                #     )
                #     .options(*cls._defercols())
                #     .limit(20),
                # )
                .all()
            )
        elif '@' in prefix and not prefix.startswith('@'):
            # Query has an @ in the middle. Match email address (exact match only).
            # Use param `prefix` instead of `like_query` because it's not a LIKE query.
            # Combine results with regular user search
            email_filter = EmailAddress.get_filter(email=prefix)
            if email_filter is not None:
                users = (
                    cls.query.join(AccountEmail)
                    .join(EmailAddress)
                    .filter(email_filter, cls.state.ACTIVE)
                    .options(*cls._defercols())
                    .limit(20)
                    # .union(base_users)  # FIXME: Broken in SQLAlchemy 1.4.17
                    .all()
                )
            else:
                users = []
        else:
            # No '@' in the query, so do a regular autocomplete
            try:
                users = base_users.all()
            except sa_exc.ProgrammingError:
                # This can happen because the tsquery from prefix turned out to be ':*'
                users = []
        return users

    @classmethod
    def validate_name_candidate(cls, name: str) -> str | None:
        """
        Validate an account name candidate.

        Returns one of several error codes, or `None` if all is okay:

        * ``blank``: No name supplied
        * ``reserved``: Name is reserved
        * ``invalid``: Invalid characters in name
        * ``long``: Name is longer than allowed size
        * ``user``: Name is assigned to a user
        * ``org``: Name is assigned to an organization
        """
        if not name:
            return 'blank'
        if name.lower() in cls.reserved_names:
            return 'reserved'
        if not valid_account_name(name):
            return 'invalid'
        if len(name) > cls.__name_length__:
            return 'long'
        # Look for existing on the base Account model, not the subclass, as SQLAlchemy
        # will add a filter condition on subclasses to restrict the query to that type.
        existing = (
            Account.query.filter(sa.func.lower(Account.name) == sa.func.lower(name))
            .options(sa_orm.load_only(cls.id, cls.uuid, cls.type_))
            .one_or_none()
        )
        if existing is not None:
            if isinstance(existing, Placeholder):
                return 'reserved'
            if isinstance(existing, User):
                return 'user'
            if isinstance(existing, Organization):
                return 'org'
        return None

    def validate_new_name(self, name: str) -> str | None:
        """Validate a new name for this account, returning an error code or None."""
        if self.name and name.lower() == self.name.lower():
            return None
        return self.validate_name_candidate(name)

    @classmethod
    def is_available_name(cls, name: str) -> bool:
        """Test if the candidate name is available for use as an Account name."""
        return cls.validate_name_candidate(name) is None

    @sa_orm.validates('name')
    def _validate_name(self, key: str, value: str | None) -> str | None:
        """Validate the value of Account.name."""
        if value is None:
            return value

        if not isinstance(value, str):
            raise ValueError(f"Account name must be a string: {value}")

        if not value.strip():
            raise ValueError("Account name cannot be blank")

        if value.lower() in self.reserved_names or not valid_account_name(value):
            raise ValueError("Invalid account name: " + value)

        # We don't check for existence in the db since this validator only
        # checks for valid syntax. To confirm the name is actually available,
        # the caller must call :meth:`is_available_name` or attempt to commit
        # to the db and catch IntegrityError.
        return value

    @sa_orm.validates('logo_url', 'banner_image_url')
    def _validate_nullable(self, key: str, value: str | None) -> str | None:
        """Convert blank values into None."""
        return value if value else None

    @classmethod
    def active_count(cls) -> int:
        """Count of all active accounts."""
        return cls.query.filter(cls.state.ACTIVE).count()

    #: FIXME: Temporary values for Baseframe compatibility
    def organization_links(self) -> list:
        """Return list of organizations affiliated with this user (deprecated)."""
        return []

    # Project methods

    def draft_projects_for(self, user: Account | None) -> list[Project]:
        if user is not None:
            return [
                membership.project
                for membership in user.projects_as_crew_active_memberships.join(
                    Project
                ).filter(
                    # Project is attached to this account
                    Project.account_id == self.id,
                    # Project is in draft state OR has a draft call for proposals
                    sa.or_(Project.state.DRAFT, Project.cfp_state.DRAFT),
                )
            ]
        return []

    def unscheduled_projects_for(self, user: Account | None) -> list[Project]:
        if user is not None:
            return [
                membership.project
                for membership in user.projects_as_crew_active_memberships.join(
                    Project
                ).filter(
                    # Project is attached to this account
                    Project.account_id == self.id,
                    # Project is in draft state OR has a draft call for proposals
                    sa.or_(Project.state.PUBLISHED_WITHOUT_SESSIONS),
                )
            ]
        return []

    @with_roles(read={'all'}, datasets={'primary', 'without_parent', 'related'})
    @cached_property
    def published_project_count(self) -> int:
        return (
            self.listed_projects.filter(Project.state.PUBLISHED).order_by(None).count()
        )

    @with_roles(read={'all'}, grants_via={None: {'participant': 'member'}})
    @cached_property
    def membership_project(self) -> Project | None:
        """Return a project that has memberships flag enabled (temporary)."""
        return self.projects.filter(
            Project.boxoffice_data.op('@>')({'has_membership': True})
        ).first()

    # Make :attr:`type_` available under the name `type`, but declare this at the very
    # end of the class to avoid conflicts with the Python `type` global that is
    # used for type-hinting
    type: Mapped[str] = sa_orm.synonym('type_')  # noqa: A003


Account.__active_membership_attrs__.add('active_organization_admin_memberships')
Account.__noninvite_membership_attrs__.add('noninvite_organization_admin_memberships')
Account.__active_membership_attrs__.add('projects_as_crew_active_memberships')
Account.__noninvite_membership_attrs__.add('projects_as_crew_noninvite_memberships')
Account.__active_membership_attrs__.add('proposal_memberships')
Account.__noninvite_membership_attrs__.add('noninvite_proposal_memberships')
Account.__active_membership_attrs__.update(
    {'project_sponsor_memberships', 'proposal_sponsor_memberships'}
)
Account.__noninvite_membership_attrs__.update(
    {'noninvite_project_sponsor_memberships', 'noninvite_proposal_sponsor_memberships'}
)

auto_init_default(Account._state)  # pylint: disable=protected-access
auto_init_default(Account._profile_state)  # pylint: disable=protected-access
add_search_trigger(Account, 'search_vector')
add_search_trigger(Account, 'name_vector')


class AccountOldId(UuidMixin, BaseMixin[UUID, Account], Model):
    """Record of an older UUID for an account, after account merger."""

    __tablename__ = 'account_oldid'

    #: Old account, if still present
    old_account: Mapped[Account] = relationship(
        foreign_keys=lambda: AccountOldId.id,
        primaryjoin=lambda: AccountOldId.id == Account.uuid,
        back_populates='oldid',
    )
    #: User id of new user
    account_id: Mapped[int] = sa_orm.mapped_column(
        sa.ForeignKey('account.id'), default=None, nullable=False
    )
    #: New account
    account: Mapped[Account] = relationship(
        foreign_keys=[account_id], back_populates='oldids'
    )

    def __repr__(self) -> str:
        """Represent :class:`AccountOldId` as a string."""
        return f'<AccountOldId {self.buid} of {self.account!r}>'

    @classmethod
    def get(cls, uuid: UUID) -> AccountOldId | None:
        """Get an old user record given a UUID."""
        return cls.query.filter_by(id=uuid).one_or_none()


class User(Account):
    """User account."""

    __mapper_args__ = {'polymorphic_identity': 'U'}
    is_user_profile = True

    def __init__(self, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        if self.joined_at is None:
            self.joined_at = sa.func.utcnow()


# XXX: Deprecated, still here for Baseframe compatibility
Account.userid = Account.uuid_b64


# TODO: Make an Actor Protocol as the base for both -- maybe placing that in Coaster
class DuckTypeAccount(RoleMixin):
    """User singleton constructor. Duck types a regular user object."""

    id: None = None  # noqa: A003
    created_at: None = None
    updated_at: None = None
    uuid: None = None
    userid: None = None
    buid: None = None
    uuid_b58: None = None
    username: None = None
    name: None = None
    absolute_url: None = None
    email: None = None
    phone: None = None

    is_user_profile = True
    is_organization_profile = False
    is_placeholder_profile = False

    # Copy registries from Account model
    views = Account.views
    features = Account.features
    forms = Account.forms

    __roles__ = {
        'all': {
            'read': {
                'id',
                'uuid',
                'username',
                'fullname',
                'pickername',
                'absolute_url',
            },
            'call': {'views', 'forms', 'features', 'url_for'},
        }
    }

    __datasets__ = {
        'related': {
            'username',
            'fullname',
            'pickername',
            'absolute_url',
        }
    }

    #: Make obj.user/obj.posted_by from a referring object falsy
    def __bool__(self) -> bool:
        """Represent boolean state."""
        return False

    def __init__(self, representation: str) -> None:
        self.fullname = self.title = self.pickername = representation

    def __str__(self) -> str:
        return self.pickername

    def __format__(self, format_spec: str) -> str:
        if not format_spec:
            return self.pickername
        return format(self.pickername, format_spec)

    def url_for(self, *args: Any, **kwargs: Any) -> Literal['']:
        """Return blank URL for anything to do with this user."""
        return ''


deleted_account = DuckTypeAccount(__("[deleted]"))
removed_account = DuckTypeAccount(__("[removed]"))
unknown_account = DuckTypeAccount(__("[unknown]"))


# --- Organizations and teams -------------------------------------------------


class Organization(Account):
    """An organization of one or more users with distinct roles."""

    __mapper_args__ = {'polymorphic_identity': 'O'}
    is_organization_profile = True

    def __init__(self, owner: User, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        if self.joined_at is None:
            self.joined_at = sa.func.utcnow()
        db.session.add(
            AccountMembership(
                account=self, member=owner, granted_by=owner, is_owner=True
            )
        )

    def people(self) -> Query[Account]:
        """Return a list of users from across the public teams they are in."""
        return (
            Account.query.join(team_membership)
            .join(Team)
            .filter(Team.account == self, Team.is_public.is_(True))
            .options(sa_orm.joinedload(Account.member_teams))
            .order_by(sa.func.lower(Account.title))
        )


class Community(Account):
    """
    A community account.

    Communities differ from organizations in having open-ended membership.
    """

    __mapper_args__ = {'polymorphic_identity': 'C'}
    is_community_profile = True

    def __init__(self, owner: User, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        if self.joined_at is None:
            self.joined_at = sa.func.utcnow()
        db.session.add(
            AccountMembership(
                account=self, member=owner, granted_by=owner, is_owner=True
            )
        )


class Placeholder(Account):
    """A placeholder account."""

    __mapper_args__ = {'polymorphic_identity': 'P'}
    is_placeholder_profile = True


class Team(UuidMixin, BaseMixin[int, Account], Model):
    """A team of users within an organization."""

    __tablename__ = 'team'
    __title_length__ = 250
    #: Displayed name
    title: Mapped[str] = sa_orm.mapped_column(
        sa.Unicode(__title_length__), nullable=False
    )
    #: Organization
    account_id: Mapped[int] = sa_orm.mapped_column(
        sa.ForeignKey('account.id'), default=None, nullable=False, index=True
    )
    account: Mapped[Account] = with_roles(
        relationship(foreign_keys=[account_id], back_populates='teams'),
        grants_via={None: {'owner': 'owner', 'admin': 'admin'}},
    )
    users: DynamicMapped[Account] = with_roles(
        relationship(
            secondary=team_membership, lazy='dynamic', back_populates='member_teams'
        ),
        grants={'member'},
    )

    is_public: Mapped[bool] = sa_orm.mapped_column(default=False)

    # --- Backrefs
    client_permissions: Mapped[list[AuthClientTeamPermissions]] = relationship(
        back_populates='team'
    )

    def __repr__(self) -> str:
        """Represent :class:`Team` as a string."""
        return f'<Team {self.title} of {self.account!r}>'

    @property
    def pickername(self) -> str:
        """Return team's title in a format suitable for identification."""
        return self.title

    @classmethod
    def migrate_account(
        cls, old_account: Account, new_account: Account
    ) -> OptionalMigratedTables:
        """Migrate one account's data to another when merging accounts."""
        for team in list(old_account.teams):
            team.account = new_account
        for team in list(old_account.member_teams):
            if team not in new_account.member_teams:
                # FIXME: This creates new memberships, updating `created_at`.
                # Unfortunately, we can't work with model instances as in the other
                # `migrate_account` methods as team_membership is an unmapped table.
                new_account.member_teams.append(team)
            old_account.member_teams.remove(team)
        return [cls.__table__.name, team_membership.name]

    @classmethod
    def get(cls, buid: str, with_parent: bool = False) -> Team | None:
        """
        Return a Team with matching buid.

        :param str buid: Buid of the team
        """
        if with_parent:
            query = cls.query.options(sa_orm.joinedload(cls.account))
        else:
            query = cls.query
        return query.filter_by(buid=buid).one_or_none()


# --- Account email/phone and misc


class AccountEmail(EmailAddressMixin, BaseMixin[int, Account], Model):
    """An email address linked to an account."""

    __tablename__ = 'account_email'
    __email_unique__ = True
    __email_is_exclusive__ = True
    __email_for__ = 'account'

    account_id: Mapped[int] = sa_orm.mapped_column(
        sa.ForeignKey('account.id'), default=None, nullable=False
    )
    account: Mapped[Account] = relationship(back_populates='emails')
    user: Mapped[Account] = sa_orm.synonym('account')

    private: Mapped[bool] = sa_orm.mapped_column(default=False)

    __datasets__ = {
        'primary': {'member', 'email', 'private', 'type'},
        'without_parent': {'email', 'private', 'type'},
        'related': {'email', 'private', 'type'},
    }

    def __init__(self, *, account: Account, **kwargs: Any) -> None:
        email = kwargs.pop('email', None)
        if email:
            kwargs['email_address'] = EmailAddress.add_for(account, email)
        super().__init__(account=account, **kwargs)

    def __repr__(self) -> str:
        """Represent this class as a string."""
        return f'<AccountEmail {self.email} of {self.account!r}>'

    def __str__(self) -> str:  # pylint: disable=invalid-str-returned
        """Email address as a string."""
        return self.email or ''

    __json__ = __str__

    @property
    def primary(self) -> bool:
        """Check whether this email address is the user's primary."""
        return self.account.primary_email == self

    @primary.setter
    def primary(self, value: bool) -> None:
        """Set or unset this email address as primary."""
        if value:
            self.account.primary_email = self
        else:
            if self.account.primary_email == self:
                self.account.primary_email = None

    @overload
    @classmethod
    def get(
        cls,
        email: str,
    ) -> AccountEmail | None: ...

    @overload
    @classmethod
    def get(
        cls,
        *,
        blake2b160: bytes,
    ) -> AccountEmail | None: ...

    @overload
    @classmethod
    def get(
        cls,
        *,
        email_hash: str,
    ) -> AccountEmail | None: ...

    @classmethod
    def get(
        cls,
        email: str | None = None,
        *,
        blake2b160: bytes | None = None,
        email_hash: str | None = None,
    ) -> AccountEmail | None:
        """
        Return an AccountEmail with matching email or blake2b160 hash.

        :param email: Email address to look up
        :param blake2b160: 160-bit blake2b of email address to look up
        :param email_hash: blake2b hash rendered in Base58
        """
        email_filter = EmailAddress.get_filter(
            email=email, blake2b160=blake2b160, email_hash=email_hash
        )
        if email_filter is None:
            return None
        return cls.query.join(EmailAddress).filter(email_filter).one_or_none()

    @overload
    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        email: str,
    ) -> AccountEmail | None: ...

    @overload
    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        blake2b160: bytes,
    ) -> AccountEmail | None: ...

    @overload
    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        email_hash: str,
    ) -> AccountEmail | None: ...

    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        email: str | None = None,
        blake2b160: bytes | None = None,
        email_hash: str | None = None,
    ) -> AccountEmail | None:
        """
        Return instance with matching email or hash if it belongs to the given user.

        :param user: Account to look up for
        :param email: Email address to look up
        :param blake2b160: 160-bit blake2b of email address
        :param email_hash: blake2b hash rendered in Base58
        """
        email_filter = EmailAddress.get_filter(
            email=email, blake2b160=blake2b160, email_hash=email_hash
        )
        if email_filter is None:
            return None
        return (
            cls.query.join(EmailAddress)
            .filter(
                cls.account == account,
                email_filter,
            )
            .one_or_none()
        )

    @classmethod
    def migrate_account(
        cls, old_account: Account, new_account: Account
    ) -> OptionalMigratedTables:
        """Migrate one account's data to another when merging accounts."""
        primary_email = old_account.primary_email
        for accountemail in list(old_account.emails):
            accountemail.account = new_account
        if new_account.primary_email is None:
            new_account.primary_email = primary_email
        old_account.primary_email = None
        return [cls.__table__.name, account_email_primary_table.name]


class AccountEmailClaim(EmailAddressMixin, BaseMixin[int, Account], Model):
    """Claimed but unverified email address for a user."""

    __tablename__ = 'account_email_claim'
    __email_unique__ = False
    __email_for__ = 'account'
    __email_is_exclusive__ = False

    account_id: Mapped[int] = sa_orm.mapped_column(
        sa.ForeignKey('account.id'), default=None, nullable=False
    )
    account: Mapped[Account] = relationship(back_populates='emailclaims')
    user: Mapped[Account] = sa_orm.synonym('account')
    verification_code: Mapped[str] = sa_orm.mapped_column(
        sa.String(44), nullable=False, insert_default=newsecret, default=None
    )

    private: Mapped[bool] = sa_orm.mapped_column(default=False)

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

    __datasets__ = {
        'primary': {'member', 'email', 'private', 'type'},
        'without_parent': {'email', 'private', 'type'},
        'related': {'email', 'private', 'type'},
    }

    def __init__(self, *, account: Account, email: str, **kwargs: Any) -> None:
        kwargs['email_address'] = EmailAddress.add_for(account, email)
        super().__init__(account=account, **kwargs)
        self.blake2b = hashlib.blake2b(
            self.email.lower().encode(), digest_size=16
        ).digest()

    def __repr__(self) -> str:
        """Represent this class as a string."""
        return f'<AccountEmailClaim {self.email} of {self.account!r}>'

    def __str__(self) -> str:
        """Return email as a string."""
        return str(self.email)

    @classmethod
    def migrate_account(cls, old_account: Account, new_account: Account) -> None:
        """Migrate one account's data to another when merging accounts."""
        emails = {claim.email for claim in new_account.emailclaims}
        for claim in list(old_account.emailclaims):
            if claim.email not in emails:
                claim.account = new_account
            else:
                # New user also made the same claim. Delete old user's claim
                db.session.delete(claim)

    @overload
    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        email: str,
    ) -> AccountEmailClaim | None: ...

    @overload
    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        blake2b160: bytes,
    ) -> AccountEmailClaim | None: ...

    @overload
    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        email_hash: str,
    ) -> AccountEmailClaim | None: ...

    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        email: str | None = None,
        blake2b160: bytes | None = None,
        email_hash: str | None = None,
    ) -> AccountEmailClaim | None:
        """
        Return an AccountEmailClaim with matching email address for the given user.

        :param account: Account that claimed this email address
        :param email: Email address to look up
        :param blake2b160: 160-bit blake2b of email address to look up
        :param email_hash: Base58 rendering of 160-bit blake2b hash
        """
        email_filter = EmailAddress.get_filter(
            email=email, blake2b160=blake2b160, email_hash=email_hash
        )
        if email_filter is None:
            return None
        return (
            cls.query.join(EmailAddress)
            .filter(
                cls.account == account,
                email_filter,
            )
            .one_or_none()
        )

    @overload
    @classmethod
    def get_by(
        cls,
        verification_code: str,
        *,
        email: str,
    ) -> AccountEmailClaim | None: ...

    @overload
    @classmethod
    def get_by(
        cls,
        verification_code: str,
        *,
        blake2b160: bytes,
    ) -> AccountEmailClaim | None: ...

    @overload
    @classmethod
    def get_by(
        cls,
        verification_code: str,
        *,
        email_hash: str,
    ) -> AccountEmailClaim | None: ...

    @classmethod
    def get_by(
        cls,
        verification_code: str,
        *,
        email: str | None = None,
        blake2b160: bytes | None = None,
        email_hash: str | None = None,
    ) -> AccountEmailClaim | None:
        """Return an instance given verification code and email or hash."""
        email_filter = EmailAddress.get_filter(
            email=email, blake2b160=blake2b160, email_hash=email_hash
        )
        if email_filter is None:
            return None
        return (
            cls.query.join(EmailAddress)
            .filter(
                cls.verification_code == verification_code,
                email_filter,
            )
            .one_or_none()
        )

    @classmethod
    def all(cls, email: str) -> Query[Self]:  # noqa: A003
        """
        Return all instances with the matching email address.

        :param str email: Email address to lookup
        """
        email_filter = EmailAddress.get_filter(email=email)
        if email_filter is None:
            raise ValueError(email)
        return cls.query.join(EmailAddress).filter(email_filter)


auto_init_default(AccountEmailClaim.verification_code)


class AccountPhone(PhoneNumberMixin, BaseMixin[int, Account], Model):
    """A phone number linked to an account."""

    __tablename__ = 'account_phone'
    __phone_unique__ = True
    __phone_is_exclusive__ = True
    __phone_for__ = 'account'

    account_id: Mapped[int] = sa_orm.mapped_column(
        sa.ForeignKey('account.id'), default=None
    )
    account: Mapped[Account] = relationship(back_populates='phones')
    user: Mapped[Account] = sa_orm.synonym('account')

    private: Mapped[bool] = sa_orm.mapped_column(default=False)

    __datasets__ = {
        'primary': {'member', 'phone', 'private', 'type'},
        'without_parent': {'phone', 'private', 'type'},
        'related': {'phone', 'private', 'type'},
    }

    def __init__(self, *, account: Account, **kwargs: Any) -> None:
        phone = kwargs.pop('phone', None)
        if phone:
            kwargs['phone_number'] = PhoneNumber.add_for(account, phone)
        super().__init__(account=account, **kwargs)

    def __repr__(self) -> str:
        """Represent this class as a string."""
        return f'AccountPhone(phone={self.phone!r}, account={self.account!r})'

    def __str__(self) -> str:
        """Return phone number as a string."""
        return self.phone or ''

    __json__ = __str__

    @cached_property
    def parsed(self) -> phonenumbers.PhoneNumber | None:
        """Return parsed phone number using libphonenumber."""
        return self.phone_number.parsed

    @cached_property
    def formatted(self) -> str:
        """Return a phone number formatted for user display."""
        return self.phone_number.formatted

    @property
    def number(self) -> str | None:
        return self.phone_number.number

    @property
    def primary(self) -> bool:
        """Check if this is the user's primary phone number."""
        return self.account.primary_phone == self

    @primary.setter
    def primary(self, value: bool) -> None:
        if value:
            self.account.primary_phone = self
        else:
            if self.account.primary_phone == self:
                self.account.primary_phone = None

    @overload
    @classmethod
    def get(
        cls,
        phone: str,
    ) -> AccountPhone | None: ...

    @overload
    @classmethod
    def get(
        cls,
        *,
        blake2b160: bytes,
    ) -> AccountPhone | None: ...

    @overload
    @classmethod
    def get(
        cls,
        *,
        phone_hash: str,
    ) -> AccountPhone | None: ...

    @classmethod
    def get(
        cls,
        phone: str | None = None,
        *,
        blake2b160: bytes | None = None,
        phone_hash: str | None = None,
    ) -> AccountPhone | None:
        """
        Return an AccountPhone with matching phone number.

        :param phone: Phone number to lookup
        :param blake2b160: 160-bit blake2b of phone number to look up
        :param phone_hash: blake2b hash rendered in Base58
        """
        return (
            cls.query.join(PhoneNumber)
            .filter(
                PhoneNumber.get_filter(
                    phone=phone, blake2b160=blake2b160, phone_hash=phone_hash
                )
            )
            .one_or_none()
        )

    @overload
    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        phone: str,
    ) -> AccountPhone | None: ...

    @overload
    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        blake2b160: bytes,
    ) -> AccountPhone | None: ...

    @overload
    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        phone_hash: str,
    ) -> AccountPhone | None: ...

    @classmethod
    def get_for(
        cls,
        account: Account,
        *,
        phone: str | None = None,
        blake2b160: bytes | None = None,
        phone_hash: str | None = None,
    ) -> AccountPhone | None:
        """
        Return an instance with matching phone or hash if it belongs to the given user.

        :param account: Account to look up for
        :param phone: Email address to look up
        :param blake2b160: 160-bit blake2b of phone number
        :param phone_hash: blake2b hash rendered in Base58
        """
        return (
            cls.query.join(PhoneNumber)
            .filter(
                cls.account == account,
                PhoneNumber.get_filter(
                    phone=phone, blake2b160=blake2b160, phone_hash=phone_hash
                ),
            )
            .one_or_none()
        )

    @classmethod
    def migrate_account(
        cls, old_account: Account, new_account: Account
    ) -> OptionalMigratedTables:
        """Migrate one account's data to another when merging accounts."""
        primary_phone = old_account.primary_phone
        for accountphone in list(old_account.phones):
            accountphone.account = new_account
        if new_account.primary_phone is None:
            new_account.primary_phone = primary_phone
        old_account.primary_phone = None
        return [cls.__table__.name, account_phone_primary_table.name]


class AccountExternalId(BaseMixin[int, Account], Model):
    """An external connected account for a user."""

    __tablename__ = 'account_externalid'
    __at_username_services__: ClassVar[list[str]] = []
    #: Foreign key to user table
    account_id: Mapped[int] = sa_orm.mapped_column(
        sa.ForeignKey('account.id'), default=None
    )
    #: User that this connected account belongs to
    account: Mapped[Account] = relationship(back_populates='externalids')
    user: Mapped[Account] = sa_orm.synonym('account')
    #: Identity of the external service (in app's login provider registry)
    service: Mapped[str] = sa_orm.mapped_column(sa.Unicode, nullable=False)
    #: Unique user id as per external service, used for identifying related accounts
    userid: Mapped[str] = sa_orm.mapped_column(sa.Unicode, nullable=False)
    #: Optional public-facing username on the external service
    username: Mapped[str | None] = sa_orm.mapped_column(sa.Unicode, nullable=True)
    #: OAuth or OAuth2 access token
    oauth_token: Mapped[str | None] = sa_orm.mapped_column(sa.Unicode, nullable=True)
    #: Optional token secret (not used in OAuth2, used by Twitter with OAuth1a)
    oauth_token_secret: Mapped[str | None] = sa_orm.mapped_column(
        sa.Unicode, nullable=True
    )
    #: OAuth token type (typically 'bearer')
    oauth_token_type: Mapped[str | None] = sa_orm.mapped_column(
        sa.Unicode, nullable=True
    )
    #: OAuth2 refresh token
    oauth_refresh_token: Mapped[str | None] = sa_orm.mapped_column(
        sa.Unicode, nullable=True
    )
    #: OAuth2 token expiry in seconds, as sent by service provider
    oauth_expires_in: Mapped[int | None] = sa_orm.mapped_column()
    #: OAuth2 token expiry timestamp, estimate from created_at + oauth_expires_in
    oauth_expires_at: Mapped[datetime | None] = sa_orm.mapped_column(
        sa.TIMESTAMP(timezone=True), nullable=True, index=True
    )

    #: Timestamp of when this connected account was last (re-)authorised by the user
    last_used_at: Mapped[datetime] = sa_orm.mapped_column(
        sa.TIMESTAMP(timezone=True), insert_default=sa.func.utcnow(), default=None
    )

    __table_args__ = (
        sa.UniqueConstraint('service', 'userid'),
        sa.Index(
            'ix_account_externalid_username_lower',
            sa.func.lower(username).label('username_lower'),
            postgresql_ops={'username_lower': 'varchar_pattern_ops'},
        ),
    )

    def __repr__(self) -> str:
        """Represent :class:`UserExternalId` as a string."""
        return f'<UserExternalId {self.service}:{self.username} of {self.account!r}>'

    @overload
    @classmethod
    def get(
        cls,
        service: str,
        *,
        userid: str,
    ) -> AccountExternalId | None: ...

    @overload
    @classmethod
    def get(
        cls,
        service: str,
        *,
        username: str,
    ) -> AccountExternalId | None: ...

    @classmethod
    def get(
        cls,
        service: str,
        *,
        userid: str | None = None,
        username: str | None = None,
    ) -> AccountExternalId | None:
        """
        Return a UserExternalId with the given service and userid or username.

        :param str service: Service to lookup
        :param str userid: Userid to lookup
        :param str username: Username to lookup (may be non-unique)

        Usernames are not guaranteed to be unique within a service. An example is with
        Google, where the userid is a directed OpenID URL, unique but subject to change
        if the Lastuser site URL changes. The username is the email address, which will
        be the same despite different userids.
        """
        param, value = require_one_of(True, userid=userid, username=username)
        return cls.query.filter_by(**{param: value, 'service': service}).one_or_none()


account_email_primary_table = add_primary_relationship(
    Account, 'primary_email', AccountEmail, 'account', 'account_id'
)
account_phone_primary_table = add_primary_relationship(
    Account, 'primary_phone', AccountPhone, 'account', 'account_id'
)

#: Anchor type
Anchor: TypeAlias = (
    AccountEmail | AccountEmailClaim | AccountPhone | EmailAddress | PhoneNumber
)

# Tail imports
from .account_membership import AccountMembership
from .auth_client import AuthClient, AuthClientPermissions, AuthToken
from .login_session import LOGIN_SESSION_VALIDITY_PERIOD, LoginSession
from .mailer import Mailer
from .membership_mixin import ImmutableMembershipMixin
from .notification import NotificationPreferences, NotificationRecipient
from .project import Project, ProjectRedirect
from .project_membership import ProjectMembership
from .proposal import Proposal
from .proposal_membership import ProposalMembership
from .rsvp import Rsvp
from .saved import SavedProject, SavedSession
from .session import Session
from .site_membership import SiteMembership
from .sponsor_membership import ProjectSponsorMembership, ProposalSponsorMembership
from .sync_ticket import TicketParticipant
from .update import Update

if TYPE_CHECKING:
    from .auth_client import AuthClientTeamPermissions
    from .comment import Comment, Commentset  # noqa: F401
    from .commentset_membership import CommentsetMembership
    from .contact_exchange import ContactExchange
    from .moderation import CommentModeratorReport

hasgeek / funnel / 8456388393

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