"""§22 S4 (C.2) — the scope-role invitation surface. An Owner grants `{owner, contributor}` at a scope their reach covers — the project, or a single collection within it — to an existing account, looked up by email. The grant writes a `memberships` row immediately and §15-notifies the grantee (there is no accept round-trip; the C.2 scenarios name an existing user and write the row directly). Endpoints: GET /api/projects/:pid/members — list the project subtree's grants POST /api/projects/:pid/members — grant at project scope, or (with collection_id) at one collection DELETE /api/projects/:pid/members/:user_id — revoke (optionally ?collection_id=) The single POST keys on the optional `collection_id` so the invite UI's one control (role picker + scope picker) maps to one endpoint: * no `collection_id` → project-scope grant; gate `can_invite_at_project`. * with `collection_id` → collection-scope grant; gate `can_invite_at_collection`. There is deliberately no "grant at parent, exclude a child" parameter (C.2.5): the only knobs are role ∈ {owner, contributor} and scope ∈ {project, one collection}. Reach is bounded by the inviter's own Owner reach (C.2.3): a collection Owner who is nothing more is refused the project-scope POST. """ from __future__ import annotations from typing import Any from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel from . import ( auth, collections as collections_mod, db, memberships as memberships_mod, notify, ) class GrantBody(BaseModel): email: str role: str collection_id: str | None = None def _project_name(project_id: str) -> str | None: row = db.conn().execute( "SELECT name FROM projects WHERE id = ?", (project_id,) ).fetchone() return row["name"] if row and row["name"] else None def make_router() -> APIRouter: router = APIRouter() @router.get("/api/projects/{project_id}/members") async def list_members(project_id: str, request: Request) -> dict[str, Any]: user = auth.require_contributor(request) auth.require_project_readable(user, project_id) # The full subtree listing is a project-Owner view; a collection-only # Owner manages membership through the collection-scoped POST/DELETE. if not auth.can_invite_at_project(user, project_id): raise HTTPException(403, "You may not manage membership in this project") return {"items": memberships_mod.list_for_project(project_id)} @router.post("/api/projects/{project_id}/members") async def grant_member( project_id: str, body: GrantBody, request: Request ) -> dict[str, Any]: user = auth.require_contributor(request) auth.require_project_readable(user, project_id) role = (body.role or "").strip().lower() if role not in memberships_mod.VALID_ROLES: raise HTTPException(422, f"invalid role {body.role!r}") cid = (body.collection_id or "").strip() or None if cid is not None: # Collection-scope grant — bounded by Owner reach over that collection. col = collections_mod.get_collection(cid) if col is None or col["project_id"] != project_id: raise HTTPException(404, "Not found") if not auth.can_invite_at_collection(user, cid): raise HTTPException(403, "You may not manage membership in this collection") scope_type, scope_id = "collection", cid else: # Project-scope grant — bounded by Owner reach over the project. if not auth.can_invite_at_project(user, project_id): raise HTTPException(403, "You may not manage membership in this project") scope_type, scope_id = "project", project_id grantee = memberships_mod.user_by_email(body.email) if grantee is None: raise HTTPException( 404, "No account with that email — the invitee must sign in to the " "deployment before they can be granted a role", ) memberships_mod.grant( scope_type=scope_type, scope_id=scope_id, user_id=grantee["id"], role=role, granted_by=user.user_id, ) # §15 (C.2): name the project and role to the grantee. col_name = None if scope_type == "collection": col = collections_mod.get_collection(scope_id) col_name = (col.get("name") if col else None) or scope_id notify.notify_scope_role_granted( recipient_user_id=grantee["id"], granter_user_id=user.user_id, scope_type=scope_type, scope_id=scope_id, role=role, project_id=project_id, project_name=_project_name(project_id), collection_name=col_name, ) return { "scope_type": scope_type, "scope_id": scope_id, "user_id": grantee["id"], "role": role, "pending": grantee["permission_state"] != "granted", } @router.delete("/api/projects/{project_id}/members/{user_id}") async def revoke_member( project_id: str, user_id: int, request: Request ) -> dict[str, Any]: user = auth.require_contributor(request) auth.require_project_readable(user, project_id) cid = (request.query_params.get("collection_id") or "").strip() or None if cid is not None: col = collections_mod.get_collection(cid) if col is None or col["project_id"] != project_id: raise HTTPException(404, "Not found") if not auth.can_invite_at_collection(user, cid): raise HTTPException(403, "You may not manage membership in this collection") removed = memberships_mod.revoke( scope_type="collection", scope_id=cid, user_id=user_id ) else: if not auth.can_invite_at_project(user, project_id): raise HTTPException(403, "You may not manage membership in this project") removed = memberships_mod.revoke( scope_type="project", scope_id=project_id, user_id=user_id ) return {"removed": removed} return router