| 103 | |
| 104 | |
| 105 | class TeamMemberAddForm(StarletteForm): |
| 106 | email = StringField( |
| 107 | _l("Email"), |
| 108 | validators=[ |
| 109 | DataRequired(), |
| 110 | Email(message=_l("Invalid email address")), |
| 111 | ], |
| 112 | ) |
| 113 | role = SelectField( |
| 114 | _l("Role"), |
| 115 | choices=[ |
| 116 | ("admin", "Admin"), |
| 117 | ("member", "Member"), |
| 118 | ], |
| 119 | default="member", |
| 120 | ) |
| 121 | submit = SubmitField(_l("Invite")) |
| 122 | |
| 123 | def __init__(self, *args, team: Team, db: AsyncSession, **kwargs): |
| 124 | super().__init__(*args, **kwargs) |
| 125 | self.team = team |
| 126 | self.db = db |
| 127 | |
| 128 | async def async_validate_email(self, field): |
| 129 | if not self.db or not self.team: |
| 130 | return |
| 131 | |
| 132 | email = field.data.strip().lower() |
| 133 | existing_member = await self.db.scalar( |
| 134 | select(TeamMember) |
| 135 | .join(User) |
| 136 | .where(TeamMember.team_id == self.team.id, func.lower(User.email) == email) |
| 137 | ) |
| 138 | if existing_member: |
| 139 | raise ValidationError( |
| 140 | _("%(email)s is already a member of the team.", email=field.data) |
| 141 | ) |
| 142 | |
| 143 | pending_invite = await self.db.scalar( |
| 144 | select(TeamInvite).where( |
| 145 | TeamInvite.team_id == self.team.id, |
| 146 | func.lower(TeamInvite.email) == email, |
| 147 | TeamInvite.status == "pending", |
| 148 | ) |
| 149 | ) |
| 150 | if pending_invite and pending_invite.expires_at > utc_now(): |
| 151 | raise ValidationError( |
| 152 | _("Invitation already sent to %(email)s.", email=field.data) |
| 153 | ) |
| 154 | |
| 155 | |
| 156 | class TeamMemberRemoveForm(StarletteForm): |
nothing calls this directly
no outgoing calls
no test coverage detected