Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions buzz/api/teams/__init__.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import frappe

from buzz.api.teams import services
from buzz.api.teams.schemas import TeamOption, TeamOverview
from buzz.api.teams import invitations, services
from buzz.api.teams.schemas import InviteOutcome, TeamOption, TeamOverview


@frappe.whitelist()
Expand Down Expand Up @@ -29,3 +29,13 @@ def get_my_teams() -> list[TeamOption]:
@frappe.whitelist()
def get_team_overview(team: str) -> TeamOverview:
return services.team_overview(team)


@frappe.whitelist(methods=["POST"])
def remove_member(team: str, user: str) -> None:
services.remove_member(team, user)


@frappe.whitelist(methods=["POST"])
def invite_members(team: str, invites: list[dict]) -> list[InviteOutcome]:
return invitations.invite_members(team, invites)
17 changes: 16 additions & 1 deletion buzz/api/teams/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,22 @@
from frappe import _lt

from buzz.api.exceptions import NotPermitted
from buzz.api.exceptions import BuzzAPIError, NotPermitted


class NotATeamMember(NotPermitted):
message = _lt("You are not a member of this team.")


class CannotManageMembers(NotPermitted):
title = _lt("Not Permitted")
message = _lt("You cannot manage members of this team.")


class CannotGrantOwnership(NotPermitted):
title = _lt("Not Permitted")
message = _lt("Ownership of a team cannot be granted.")


class UnknownTeamRole(BuzzAPIError):
title = _lt("Invalid Role")
message = _lt("{team_role} is not a team role.")
68 changes: 68 additions & 0 deletions buzz/api/teams/invitations.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
import frappe
from frappe.core.api.user_invitation import invite_by_email

from buzz.api.teams.exceptions import CannotGrantOwnership, CannotManageMembers, UnknownTeamRole
from buzz.api.teams.schemas import InviteOutcome
from buzz.events.doctype.buzz_team_membership.buzz_team_membership import upsert_membership
from buzz.permissions import can_manage_members

# Where an invitee lands once they accept. The router sends a team member on to their events.
INVITE_REDIRECT_PATH = "/b/manage"


def invite_members(team: str, invites: list[dict]) -> list[InviteOutcome]:
"""Put people on a team, by whichever route each of them needs.

Someone who already has a User joins straight away; only a stranger gets an emailed
invitation. Splitting the two is not a nicety: core's `invite_by_email` skips any
address that ever accepted a buzz invitation, without regard to which team it was
for, so inviting a colleague to a second team would silently do nothing.
"""
if not can_manage_members(team):
CannotManageMembers.throw()

# Every row is checked before any of them is acted on, so one bad role cannot leave
# half a batch invited.
for invite in invites:
validate_role(invite["team_role"])

return [invite_one(team, invite) for invite in invites]


def validate_role(team_role: str) -> None:
if team_role == "Owner":
CannotGrantOwnership.throw()
if team_role not in assignable_roles():
UnknownTeamRole.throw(team_role=team_role)


def assignable_roles() -> list[str]:
"""The membership doctype's own options, minus the one nobody may be given."""
options = frappe.get_meta("Buzz Team Membership").get_field("team_role").options
return [role for role in options.split("\n") if role and role != "Owner"]


def invite_one(team: str, invite: dict) -> InviteOutcome:
# User names are lowercased emails, so the lookup has to be too.
email = invite["email"].strip().lower()
team_role = invite["team_role"]

# None means no such User; an existing one with a blank name still reads as "".
full_name = frappe.db.get_value("User", email, "full_name")
if full_name is not None:
added = upsert_membership(team, email, team_role)
return InviteOutcome(
email=email,
status="added" if added else "already_a_member",
full_name=full_name,
)

invite_by_email(
emails=email,
roles=["Buzz User"],
redirect_to_path=INVITE_REDIRECT_PATH,
app_name="buzz",
buzz_team=team,
buzz_team_role=team_role,
)
return InviteOutcome(email=email, status="invited")
14 changes: 14 additions & 0 deletions buzz/api/teams/schemas.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,24 @@ class TeamMember(APIResponse):
team_role: str


class InviteOutcome(APIResponse):
email: str
status: str
# Only someone who already has a User has a name to show; a stranger is still an address.
full_name: str | None = None


class TeamInvite(APIResponse):
email: str
team_role: str


class TeamOverview(APIResponse):
name: str
team_name: str
slug: str | None
logo: str | None
my_role: str
members: list[TeamMember]
# Kept apart from members: these people cannot do anything on the team yet.
invites: list[TeamInvite]
49 changes: 46 additions & 3 deletions buzz/api/teams/services.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
import frappe
from frappe.query_builder import Case

from buzz.api.teams.exceptions import NotATeamMember
from buzz.api.teams.schemas import TeamMember, TeamOverview
from buzz.permissions import team_role_of
from buzz.api.teams.exceptions import CannotManageMembers, NotATeamMember
from buzz.api.teams.schemas import TeamInvite, TeamMember, TeamOverview
from buzz.permissions import can_manage_members, team_role_of

TEAM_FIELDS = ("name", "team_name", "slug", "logo")

Expand All @@ -26,20 +27,62 @@ def team_overview(team: str) -> TeamOverview:
**details,
my_role=role,
members=members_of(team),
invites=pending_invites_for(team),
)


def members_of(team: str) -> list[TeamMember]:
membership = frappe.qb.DocType("Buzz Team Membership")
user = frappe.qb.DocType("User")

# A team can hold several Owners, so this is a role bucket rather than a single row.
owner_first = Case().when(membership.team_role == "Owner", 0).else_(1)

rows = (
frappe.qb.from_(membership)
.inner_join(user)
.on(user.name == membership.user)
.select(membership.user, membership.team_role, user.full_name, user.user_image)
.where((membership.team == team) & (membership.enabled == 1))
.orderby(owner_first)
.orderby(user.full_name)
).run(as_dict=True)

return [TeamMember(**row) for row in rows]


def pending_invites_for(team: str) -> list[TeamInvite]:
"""Invitations still waiting on their recipient.

App-scoped as well as team-scoped: `User Invitation` is shared with every other
installed app, and only buzz's own rows carry a team.
"""
rows = frappe.db.get_all(
"User Invitation",
filters={"buzz_team": team, "status": "Pending", "app_name": "buzz"},
fields=["email", "buzz_team_role as team_role"],
order_by="creation asc",
)

return [TeamInvite(**row) for row in rows]


def remove_member(team: str, user: str) -> None:
"""Take a member off a team by disabling their membership.

Disabled rather than deleted: `upsert_membership` re-enables the same row if they are
ever invited back, and the history survives. An Owner row refuses to be disabled in the
membership controller, so ownership needs no check here.
"""
if not can_manage_members(team):
CannotManageMembers.throw()

name = frappe.db.exists("Buzz Team Membership", {"team": team, "user": user, "enabled": 1})
if not name:
NotATeamMember.throw()

membership = frappe.get_doc("Buzz Team Membership", name)
membership.enabled = 0
# Event Manager holds no write permission on the membership doctype, so the guard above
# is the authorization — the same shape as the Desk add-members flow.
membership.save(ignore_permissions=True)
Loading
Loading