bvandeusen
22634aa0c9
- Add parse_pagination() to routes/utils.py; replace 6 duplicate limit/offset extractions in notes, tasks, chat, projects, milestones routes - Extract _enrich_shares() in sharing.py; eliminates identical 12-line loop in list_project_shares and list_note_shares - Extract _deduplicate_by_permission() in sharing.py; eliminates identical deduplication blocks in list_shared_with_me for projects and notes Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
262 lines
9.2 KiB
Python
262 lines
9.2 KiB
Python
"""Sharing service — create/manage project and note shares."""
|
|
|
|
import logging
|
|
|
|
from sqlalchemy import select
|
|
|
|
from fabledassistant.models import async_session
|
|
from fabledassistant.models.group import Group, GroupMembership
|
|
from fabledassistant.models.note import Note
|
|
from fabledassistant.models.project import Project
|
|
from fabledassistant.models.share import NoteShare, ProjectShare
|
|
from fabledassistant.models.user import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
VALID_PERMISSIONS = {"viewer", "editor", "admin"}
|
|
|
|
|
|
async def _enrich_shares(session, shares) -> list[dict]:
|
|
"""Attach username / group_name to a list of share rows (ProjectShare or NoteShare)."""
|
|
result = []
|
|
for s in shares:
|
|
d = s.to_dict()
|
|
if s.shared_with_user_id:
|
|
user = await session.get(User, s.shared_with_user_id)
|
|
d["username"] = user.username if user else None
|
|
if s.shared_with_group_id:
|
|
group = await session.get(Group, s.shared_with_group_id)
|
|
d["group_name"] = group.name if group else None
|
|
result.append(d)
|
|
return result
|
|
|
|
|
|
def _deduplicate_by_permission(shares, id_attr: str) -> dict[int, str]:
|
|
"""Return {resource_id: best_permission} keeping the highest-ranked permission per resource."""
|
|
from fabledassistant.services.access import PERMISSION_RANK
|
|
seen: dict[int, str] = {}
|
|
for share in shares:
|
|
rid = getattr(share, id_attr)
|
|
prev = seen.get(rid)
|
|
if prev is None or PERMISSION_RANK[share.permission] > PERMISSION_RANK[prev]:
|
|
seen[rid] = share.permission
|
|
return seen
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Project shares
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def share_project(
|
|
acting_user_id: int,
|
|
project_id: int,
|
|
*,
|
|
target_user_id: int | None = None,
|
|
target_group_id: int | None = None,
|
|
permission: str = "viewer",
|
|
) -> ProjectShare | None:
|
|
if permission not in VALID_PERMISSIONS:
|
|
return None
|
|
from fabledassistant.services.access import can_admin_project
|
|
if not await can_admin_project(acting_user_id, project_id):
|
|
return None
|
|
async with async_session() as session:
|
|
share = ProjectShare(
|
|
project_id=project_id,
|
|
shared_with_user_id=target_user_id,
|
|
shared_with_group_id=target_group_id,
|
|
permission=permission,
|
|
invited_by=acting_user_id,
|
|
)
|
|
session.add(share)
|
|
await session.commit()
|
|
await session.refresh(share)
|
|
return share
|
|
|
|
|
|
async def update_project_share(
|
|
acting_user_id: int, share_id: int, permission: str
|
|
) -> ProjectShare | None:
|
|
if permission not in VALID_PERMISSIONS:
|
|
return None
|
|
async with async_session() as session:
|
|
share = await session.get(ProjectShare, share_id)
|
|
if not share:
|
|
return None
|
|
from fabledassistant.services.access import can_admin_project
|
|
if not await can_admin_project(acting_user_id, share.project_id):
|
|
return None
|
|
share.permission = permission
|
|
await session.commit()
|
|
await session.refresh(share)
|
|
return share
|
|
|
|
|
|
async def remove_project_share(acting_user_id: int, share_id: int) -> bool:
|
|
async with async_session() as session:
|
|
share = await session.get(ProjectShare, share_id)
|
|
if not share:
|
|
return False
|
|
from fabledassistant.services.access import can_admin_project
|
|
if not await can_admin_project(acting_user_id, share.project_id):
|
|
return False
|
|
await session.delete(share)
|
|
await session.commit()
|
|
return True
|
|
|
|
|
|
async def list_project_shares(project_id: int) -> list[dict]:
|
|
async with async_session() as session:
|
|
shares = (await session.execute(
|
|
select(ProjectShare).where(ProjectShare.project_id == project_id)
|
|
)).scalars().all()
|
|
return await _enrich_shares(session, shares)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Note shares
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def share_note(
|
|
acting_user_id: int,
|
|
note_id: int,
|
|
*,
|
|
target_user_id: int | None = None,
|
|
target_group_id: int | None = None,
|
|
permission: str = "viewer",
|
|
) -> NoteShare | None:
|
|
if permission not in VALID_PERMISSIONS:
|
|
return None
|
|
from fabledassistant.services.access import get_note_permission
|
|
perm = await get_note_permission(acting_user_id, note_id)
|
|
if perm not in ("admin", "owner"):
|
|
return None
|
|
async with async_session() as session:
|
|
share = NoteShare(
|
|
note_id=note_id,
|
|
shared_with_user_id=target_user_id,
|
|
shared_with_group_id=target_group_id,
|
|
permission=permission,
|
|
invited_by=acting_user_id,
|
|
)
|
|
session.add(share)
|
|
await session.commit()
|
|
await session.refresh(share)
|
|
return share
|
|
|
|
|
|
async def update_note_share(
|
|
acting_user_id: int, share_id: int, permission: str
|
|
) -> NoteShare | None:
|
|
if permission not in VALID_PERMISSIONS:
|
|
return None
|
|
async with async_session() as session:
|
|
share = await session.get(NoteShare, share_id)
|
|
if not share:
|
|
return None
|
|
from fabledassistant.services.access import get_note_permission
|
|
perm = await get_note_permission(acting_user_id, share.note_id)
|
|
if perm not in ("admin", "owner"):
|
|
return None
|
|
share.permission = permission
|
|
await session.commit()
|
|
await session.refresh(share)
|
|
return share
|
|
|
|
|
|
async def remove_note_share(acting_user_id: int, share_id: int) -> bool:
|
|
async with async_session() as session:
|
|
share = await session.get(NoteShare, share_id)
|
|
if not share:
|
|
return False
|
|
from fabledassistant.services.access import get_note_permission
|
|
perm = await get_note_permission(acting_user_id, share.note_id)
|
|
if perm not in ("admin", "owner"):
|
|
return False
|
|
await session.delete(share)
|
|
await session.commit()
|
|
return True
|
|
|
|
|
|
async def list_note_shares(note_id: int) -> list[dict]:
|
|
async with async_session() as session:
|
|
shares = (await session.execute(
|
|
select(NoteShare).where(NoteShare.note_id == note_id)
|
|
)).scalars().all()
|
|
return await _enrich_shares(session, shares)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared-with-me
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def list_shared_with_me(user_id: int) -> dict:
|
|
"""All projects and notes shared with the user (directly or via group)."""
|
|
async with async_session() as session:
|
|
user_group_ids = (await session.execute(
|
|
select(GroupMembership.group_id).where(GroupMembership.user_id == user_id)
|
|
)).scalars().all()
|
|
|
|
# --- Projects ---
|
|
proj_direct = (await session.execute(
|
|
select(ProjectShare).where(ProjectShare.shared_with_user_id == user_id)
|
|
)).scalars().all()
|
|
|
|
proj_group: list[ProjectShare] = []
|
|
if user_group_ids:
|
|
proj_group = (await session.execute(
|
|
select(ProjectShare).where(
|
|
ProjectShare.shared_with_group_id.in_(user_group_ids)
|
|
)
|
|
)).scalars().all()
|
|
|
|
seen_projects = _deduplicate_by_permission(list(proj_direct) + list(proj_group), "project_id")
|
|
|
|
projects = []
|
|
for pid, perm in seen_projects.items():
|
|
proj = await session.get(Project, pid)
|
|
if proj:
|
|
owner = await session.get(User, proj.user_id)
|
|
projects.append({
|
|
"id": proj.id,
|
|
"title": proj.title,
|
|
"description": proj.description,
|
|
"status": proj.status,
|
|
"color": proj.color,
|
|
"updated_at": proj.updated_at.isoformat(),
|
|
"owner_username": owner.username if owner else None,
|
|
"permission": perm,
|
|
})
|
|
|
|
# --- Notes / Tasks ---
|
|
note_direct = (await session.execute(
|
|
select(NoteShare).where(NoteShare.shared_with_user_id == user_id)
|
|
)).scalars().all()
|
|
|
|
note_group: list[NoteShare] = []
|
|
if user_group_ids:
|
|
note_group = (await session.execute(
|
|
select(NoteShare).where(
|
|
NoteShare.shared_with_group_id.in_(user_group_ids)
|
|
)
|
|
)).scalars().all()
|
|
|
|
seen_notes = _deduplicate_by_permission(list(note_direct) + list(note_group), "note_id")
|
|
|
|
notes = []
|
|
for nid, perm in seen_notes.items():
|
|
note = await session.get(Note, nid)
|
|
if note:
|
|
owner = await session.get(User, note.user_id)
|
|
notes.append({
|
|
"id": note.id,
|
|
"title": note.title,
|
|
"is_task": note.is_task,
|
|
"project_id": note.project_id,
|
|
"updated_at": note.updated_at.isoformat(),
|
|
"owner_username": owner.username if owner else None,
|
|
"permission": perm,
|
|
})
|
|
|
|
return {"projects": projects, "notes": notes}
|