Coverage for postrfp / buyer / api / endpoints / projects.py: 99%
279 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-03 01:35 +0000
1"""
2Manage projects, project permissions & project notes
3"""
5from datetime import datetime
8from sqlalchemy import select
9from sqlalchemy.orm import Session
10from sqlalchemy.exc import IntegrityError
12from postrfp.authorisation import perms
13from postrfp.shared import fetch, update
14from postrfp.shared.decorators import http
15from postrfp.shared.pager import Pager
16from postrfp.shared.issue_transition import change_issue_status
17from postrfp.shared import serial
18from postrfp.model import (
19 Project,
20 AuditEvent,
21 Participant,
22 ProjectPermission,
23 SectionPermission,
24 User,
25 Section,
26 Issue,
27 ProjectWatchList,
28 ProjectField,
29 ProjectApproval,
30)
31from postrfp.model.audit import evt_types
32from postrfp.model.misc import Category
33from postrfp.model.exc import BusinessRuleViolation, DuplicateDataProvided
34from postrfp.authorisation.roles import ROLES
35from postrfp.buyer.api import authorise, domain_permissions
38@http
39def get_projects(
40 session: Session,
41 user: User,
42 project_statuses: set[str] | None = None,
43 project_sort: str = "date_created",
44 sort_order: str = "desc",
45 q_participant_id: str | None = None,
46 q_title: str | None = None,
47 q_category_id: str | None = None,
48 q_workflow_id: str | None = None,
49 pager: Pager | None = None,
50) -> serial.ProjectList:
51 """
52 An array of projects that the user's organisation is a participant
53 of and that the user, if restricted, has permission to access.
55 If no values are provided for project_statuses then projects with any status are returned.
57 Using the "participant_id" query param will filter the list of projects to include only
58 those for which have a Participant organisation whose ID is that of participant_id. This
59 parameter is only available to users in Consultant organisations and the participant_id must
60 be the ID of an organisation that is a Client of the Consultant.
61 """
62 if pager is None:
63 pager = Pager(page=1, page_size=100)
65 if q_participant_id:
66 participant_org = fetch.organisation(session, q_participant_id)
67 authorise.check(user, perms.MANAGE_ORGANISATION, target_org=participant_org)
69 pq = fetch.projects_with_watched(session, user, participant_id=q_participant_id)
71 if q_category_id:
72 pq = pq.filter(Project.categories.any(Category.id == q_category_id))
74 sort = getattr(Project, project_sort)
75 sort_directed = sort.asc() if sort_order == "asc" else sort.desc()
76 pq = pq.order_by(sort_directed)
78 if project_statuses:
79 from postrfp.model.fsm import Status
81 statuses_q = select(Status.id).where(Status.name.in_(project_statuses))
82 pq = pq.filter(Project.current_status_id.in_(statuses_q))
84 if q_title:
85 pq = pq.filter(Project.title.contains(q_title))
87 if q_workflow_id:
88 pq = pq.filter(Project.workflow_id == q_workflow_id)
90 total_records = pq.count()
91 records = pq.slice(pager.startfrom, pager.goto).all()
93 return serial.ProjectList(
94 data=list(serial.ListProject.model_validate(p) for p in records),
95 pagination=pager.as_pagination(total_records, len(records)),
96 )
99@http
100def get_project(session: Session, project_id: int, user: User) -> serial.FullProject:
101 """Get a project checking that the user is in a participant
102 organisation and has explicit user permissions for that project
103 """
104 project = fetch.project(session, project_id, with_description=True)
105 authorise.check(user, perms.PROJECT_ACCESS, project=project)
106 schema: serial.FullProject = serial.FullProject.model_validate(project)
107 schema.set_permissions(user)
108 return schema
111@http
112def post_project(
113 session: Session, user: User, new_project_doc: dict
114) -> serial.NewProjectIds:
115 """
116 Create a new Project
118 ID values provided in the category_ids field of the JSON request body
119 are used to link the new project to categories with those ID values.
121 The 'section_id' field in the Response is the ID of the Root Section for the
122 newly created project.
123 """
124 user.check_permission(perms.PROJECT_CREATE)
125 ptitle = new_project_doc.pop("title")
126 project_fields = new_project_doc.pop("project_fields", [])
127 category_ids = new_project_doc.pop("category_ids", [])
129 sec_title = new_project_doc.pop("questionnaire_title")
130 root_section = Section(title=sec_title, number="")
132 project = Project(ptitle, **new_project_doc)
133 project.owner_org = user.organisation
134 project.author = user
135 session.add(project)
136 participant = Participant(organisation=user.organisation, role="Administrator")
137 project.participants.add(participant)
138 session.flush()
139 assert project.workflow is not None
140 project.current_status = project.workflow.get_initial_status()
141 evt = AuditEvent.create(
142 session,
143 evt_types.PROJECT_CREATED,
144 object_id=project.id,
145 user=user,
146 project=project,
147 )
149 root_section.project_id = project.id
150 project.root_section = root_section
151 root_section.renumber()
153 # project_fields
154 for idx, pf in enumerate(project_fields):
155 new_pf = ProjectField(position=idx, **pf)
156 evt.add_change(f"extra fields/{new_pf.key}", None, new_pf.value)
157 project.project_fields.append(new_pf)
159 if category_ids:
160 cat_map = {cat.id: cat for cat in user.organisation.categories}
161 for cat_id in category_ids:
162 if cat_id not in cat_map:
163 org_id = user.organisation.id
164 raise ValueError(f"Category #{cat_id} is not found for org {org_id}")
165 cat = cat_map[cat_id]
166 project.categories.append(cat)
168 for k, v in new_project_doc.items():
169 if not v:
170 continue
171 evt.add_change(k.title(), "", v) # title() converts string to Title Case
172 session.add(evt)
173 session.flush()
175 return serial.NewProjectIds(id=project.id, section_id=root_section.id)
178@http
179def put_project(
180 session: Session, user: User, project_id: int, update_project_doc: dict
181):
182 """
183 Update the project with the values contained in the UpdateableProject document.
185 If 'project_fields' is provided the existing values are overwritten.
186 """
188 project = fetch.project(session, project_id)
189 authorise.check(user, perms.PROJECT_EDIT, project=project)
191 evt = AuditEvent.create(
192 session, "PROJECT_UPDATED", object_id=project.id, user=user, project=project
193 )
195 # project_fields
196 project_fields = update_project_doc.pop("project_fields", None)
197 if project_fields:
198 project.project_fields.clear()
199 session.flush() # Required for delete-orphan cascade to work
200 for idx, pf in enumerate(project_fields):
201 new_pf = ProjectField(position=idx, **pf)
202 evt.add_change(f"extra fields/{new_pf.key}", None, new_pf.value)
203 project.project_fields.append(new_pf)
205 for attr_name, new_val in update_project_doc.items():
206 current_val = getattr(project, attr_name)
207 if current_val != new_val:
208 evt.add_change(attr_name.title(), current_val, new_val)
209 setattr(project, attr_name, new_val)
210 session.add(evt)
213@http
214def delete_project(session: Session, user: User, project_id: int):
215 """
216 Delete the given project together with all sections, subsections and questions
217 """
218 user.check_permission(perms.PROJECT_DELETE)
219 project = fetch.project(session, project_id)
221 update.delete_qinstances_update_def_refcounts(session, project_id)
223 session.query(Section).filter(Section.id == project.section_id).delete()
224 session.query(Issue).filter(Issue.project_id == project.id).delete()
225 evt = AuditEvent.create(
226 session, "PROJECT_DELETED", object_id=project_id, user=user, project=project
227 )
228 session.add(evt)
229 session.flush()
230 session.delete(project)
233@http
234def get_project_participants(
235 session: Session, project_id: int, user: User
236) -> serial.ParticipantList:
237 """
238 List organisations that are registered as Participants in the given project together
239 with the Role assigned in the given project.
240 """
241 project = fetch.project(session, project_id)
242 authorise.check(user, perms.PROJECT_ACCESS, project=project)
243 return serial.ParticipantList(
244 list(serial.Participant.model_validate(p) for p in project.participants)
245 )
248@http
249def put_project_participants(
250 session: Session, user: User, project_id: int, participants_list: list
251) -> serial.ParticipantList:
252 """
253 Update the Organisations participating in the given project together with their Roles
255 The current user's own Organisation cannot be removed as participant.
256 """
257 project = fetch.project(session, project_id)
258 authorise.check(user, perms.PROJECT_MANAGE_ROLES, project=project)
260 new_participants_dict = {p["org_id"]: p["role"] for p in participants_list}
262 for p in project.participants:
263 if p.org_id not in new_participants_dict:
264 if p.org_id != user.org_id: # don't unparticipate the current user
265 session.delete(p)
266 else:
267 p.role = new_participants_dict[p.org_id]
268 del new_participants_dict[p.org_id]
270 # new_participants_dict now contains only orgs not already a Participant in this project
271 for org_id, role in new_participants_dict.items():
272 org = fetch.organisation(session, org_id)
273 if not org.is_buyside:
274 raise ValueError(
275 (
276 f"Only Buyer or Consultant organisations can be Participants."
277 f" {org_id} is a {org.type}"
278 )
279 )
280 project.participants.add(Participant(organisation=org, role=role))
282 return serial.ParticipantList(
283 list(serial.Participant.model_validate(p) for p in project.participants)
284 )
287@http
288def put_project_permissions(
289 session: Session, user: User, project_id: int, perm_doc: serial.ProjectPermission
290):
291 """
292 Assign section permissions for a restricted user to access the project
293 """
294 project = fetch.project(session, project_id)
295 authorise.check(user, perms.MANAGE_USERS, target_org=user.organisation)
296 authorise.check(
297 user, perms.PROJECT_MANAGE_ROLES, project=project, target_org=user.organisation
298 )
299 return domain_permissions.save_project_permissions(user, project, perm_doc)
302@http
303def get_project_users(
304 session: Session, user: User, project_id: int, restricted_users: bool = False
305) -> serial.UserList:
306 """
307 List all participant users with access to the project.
308 """
309 project = fetch.project(session, project_id)
310 authorise.check(user, perms.PROJECT_ACCESS, project=project)
311 authorise.check(user, perms.MANAGE_USERS, target_org=user.organisation)
312 user_query = fetch.project_users(user, project_id, restricted_users)
313 return serial.UserList([serial.User.model_validate(u) for u in user_query])
316@http
317def get_project_user_permissions(
318 session: Session, user: User, project_id: int, target_user: User | None = None
319) -> serial.ProjectUser:
320 """
321 Returns a User dictionary augmented with sets of permissions (string names) given the
322 the user's effective permissions in the given project. This set is derived from the
323 intersection of the user's permissions plus the user's participant organisation
324 permission (role) in the project.
326 If target_user is given then the permissions apply to that user, otherwise
327 the permissions apply to the current user (user arg)
328 """
329 project = fetch.project(session, project_id)
330 authorise.check(user, perms.PROJECT_ACCESS, project=project)
332 if target_user is None or target_user == user.id:
333 perms_user = user
334 else:
335 perms_user = fetch.user(session, target_user.id)
336 authorise.check(
337 user,
338 perms.MANAGE_USERS,
339 target_org=perms_user.organisation,
340 target_user=perms_user,
341 )
343 participant = project.get_participant(perms_user.organisation)
345 participant_role = participant.role or "No Role Assigned"
347 project_permissions = ROLES[participant_role]
349 udict = serial.ProjectUser.model_validate(perms_user)
350 udict.effectivePermissions = sorted(
351 project_permissions & perms_user.all_permissions
352 )
353 udict.userPermissions = sorted(perms_user.all_permissions)
354 udict.participantPermissions = sorted(project_permissions)
355 udict.participantRole = participant.role
357 return udict
360@http
361def get_project_permissions(
362 session: Session, user: User, project_id: int, target_user: User
363):
364 """
365 returns {
366 target_user: user whose permissions are in focus
367 permissions: list of section ids the user can access
368 }
369 """
370 project = fetch.project(session, project_id)
371 authorise.check(user, perms.PROJECT_ACCESS, project=project)
372 authorise.check(
373 user,
374 perms.MANAGE_USERS,
375 target_user=target_user,
376 target_org=target_user.organisation,
377 )
379 proj_query = (
380 session.query(SectionPermission)
381 .join(ProjectPermission)
382 .join(Participant)
383 .filter(
384 Participant.org_id == target_user.organisation.id,
385 SectionPermission.user == target_user,
386 Participant.project == project,
387 )
388 )
390 return {
391 "permissions": [sp.section_id for sp in proj_query],
392 "user": target_user.as_dict(),
393 }
396@http
397def get_project_watchlist(
398 session: Session, user: User, project_id: int
399) -> list[serial.Watcher]:
400 """
401 List the users watching this project
402 """
403 project = fetch.project(session, project_id)
404 authorise.check(user, perms.PROJECT_ACCESS, project=project)
405 return [
406 serial.Watcher.model_validate(w)
407 for w in fetch.project_watchers(session, project)
408 ]
411@http
412def post_project_watch(
413 session: Session, user: User, project_id: int, watch_doc: serial.TargetUser
414) -> list[serial.Watcher]:
415 """
416 Assign a new user as a watcher of the given project.
417 Returns a list of user ids watching the current project
418 """
419 project = fetch.project(session, project_id)
420 authorise.check(user, perms.PROJECT_ACCESS, project=project)
422 target_user_id = watch_doc.targetUser
424 if target_user_id is None:
425 watcher_user = user
426 else:
427 watcher_user = fetch.user(session, target_user_id)
428 authorise.check(
429 user,
430 perms.MANAGE_USERS,
431 target_user=watcher_user,
432 target_org=watcher_user.organisation,
433 )
435 if not project.add_watcher(watcher_user):
436 raise BusinessRuleViolation(
437 "User %s is already watching Project %s" % (watcher_user.id, project.title)
438 )
439 session.flush()
440 return get_project_watchlist(session, user, project.id)
443@http
444def delete_project_watch(
445 session: Session, user: User, project_id: int, target_user: User
446) -> list[serial.Watcher]:
447 """
448 Remove the target user from the watchlist for this project
449 """
450 project = fetch.project(session, project_id)
451 authorise.check(user, perms.PROJECT_ACCESS, project=project)
452 if target_user is not user:
453 authorise.check(
454 user,
455 perms.MANAGE_USERS,
456 target_user=target_user,
457 target_org=target_user.organisation,
458 )
459 project.watch_list.filter(ProjectWatchList.user == target_user).delete()
460 return get_project_watchlist(session, user, project.id)
463@http
464def put_project_category(
465 session: Session, user: User, project_id: int, category_id: int
466):
467 """
468 Assign the category with the given ID to the project at this URL
469 @permission: editProject
470 """
471 project = fetch.project(session, project_id)
472 authorise.check(user, perms.PROJECT_EDIT, project=project)
473 category = fetch.category_for_user(session, user, category_id)
474 if category in project.categories:
475 raise ValueError(
476 f"Category ID {category_id} is already assigned to project {project_id}"
477 )
478 project.categories.append(category)
481@http
482def delete_project_category(
483 session: Session, user: User, project_id: int, category_id: int
484):
485 """Remove the category with the given ID from the project at this URL"""
486 project = fetch.project(session, project_id)
487 authorise.check(user, perms.PROJECT_EDIT, project=project)
488 category = fetch.category_for_user(session, user, category_id)
489 if category not in project.categories:
490 raise ValueError(
491 f"Category ID {category_id} is not assigned to project {project_id}"
492 )
493 project.categories.remove(category)
496@http
497def put_project_publish(
498 session: Session, user: User, project_id: int, publish_doc: dict
499) -> list[serial.PublishResult]:
500 """
501 Publish a project.
503 Changes the status of the Project from 'Draft' to 'Live'.
505 ### Update Issue Statuses
507 When project is published it is often useful to simultaneously "release" the issues that
508 belong to the project. "Release" means to change the status of an issue from 'Not Sent' to
509 either 'Opportunity' or 'Accepted', depending on the value of the parent project's
510 'Required Acceptance' field.
512 If the Project's require_acceptance field is true, the the status of each Issue
513 will change from 'Not Sent' to 'Opportunity'.
515 If the Project's required_acceptance field is false the each Issue will move from
516 status 'Not Sent' to 'Accepted'.
518 Only issues whose ID values are included in the JSON body 'release_issue_ids' field will
519 be updated.
521 If there are no values given in the JSON body field 'release_issue_id'
522 then the status of all issues belong to the project remain unchanged.
524 ID values in 'release_issue_ids' that do not belong to the current project are ignored.
525 """
526 project = fetch.project(session, project_id)
527 authorise.check(user, perms.PROJECT_PUBLISH, project=project)
529 dt_now = datetime.now()
531 # Table inherits from FromClause, which is the real type of Project.__table__
532 project.status_name = "Live"
534 project.date_published = dt_now
535 pevt = AuditEvent.create(
536 session, "PROJECT_PUBLISHED", object_id=project_id, user=user, project=project
537 )
539 pevt.add_change("status", "Draft", "Live")
540 pevt.add_change("date_published", None, str(dt_now))
542 issue_id_set = set(publish_doc.get("release_issue_ids", []))
544 new_issue_status = "Opportunity"
546 res = []
547 for issue in project.issues:
548 if issue.status == "Not Sent" and issue.id in issue_id_set:
549 change_issue_status(issue, user, new_issue_status)
550 res_id = issue.respondent_id or issue.respondent_email
551 assert res_id is not None, f"Issue {issue.id} has no respondent ID or email"
552 res.append(serial.PublishResult(issue_id=issue.id, respondent_id=res_id))
553 return res
556@http
557def put_project_close(session: Session, user: User, project_id: int):
558 """
559 Close the given project.
561 Changes the status of the Project from 'Live' to 'Closed'.
562 Any pending issues at status 'Opportunity' will be retracted.
563 """
564 project = fetch.project(session, project_id)
565 authorise.check(user, perms.PROJECT_CLOSE, project=project)
567 for opportunity in project.opportunity_issues:
568 change_issue_status(opportunity, user, "Retracted")
570 project.status_name = "Closed"
572 pevt = AuditEvent.create(
573 session, "PROJECT_CLOSED", object_id=project_id, user=user, project=project
574 )
576 pevt.add_change("status", "Live", "Closed")
577 session.add(pevt)
580@http
581def get_project_approvals(
582 session: Session, user: User, project_id: int
583) -> list[serial.ProjectApproval]:
584 """
585 List all approvals for the given project.
587 An approval is recorded when a user approves a project for its current status.
588 A user may only approve a given project once for each distinct project status.
589 @permission: accessProject
590 """
591 project = fetch.project(session, project_id)
592 authorise.check(user, perms.PROJECT_ACCESS, project=project)
594 return [serial.ProjectApproval.model_validate(a) for a in project.approvals]
597@http
598def post_project_approval(
599 session: Session, user: User, project_id: int
600) -> list[serial.ProjectApproval]:
601 """
602 Log an approval for the given project by the current user.
604 A user may only approve a given project once for each distinct project status.
605 If the user has already approved the project for its current status then
606 an error is raised.
608 The approval is recorded in the ProjectApproval table together with
609 the project status at the time of approval.
611 @permission: approveProject
612 """
613 project = fetch.project(session, project_id)
614 authorise.check(user, perms.PROJECT_APPROVE, project=project)
616 try:
617 approval = project.log_approval(user)
618 session.flush()
619 except IntegrityError as e:
620 session.rollback()
621 raise DuplicateDataProvided(
622 f"User {user.id} has already approved Project ID {project.id} "
623 f"at status {project.status_name}"
624 ) from e
626 AuditEvent.create(
627 session,
628 evt_types.PROJECT_APPROVAL_LOGGED,
629 object_id=approval.id,
630 user=user,
631 project=project,
632 )
633 return [serial.ProjectApproval.model_validate(a) for a in project.approvals]
636@http
637def delete_project_approval(
638 session: Session, user: User, project_id: int, approval_id: int
639) -> list[serial.ProjectApproval]:
640 """
641 Remove the given approval from the project.
643 Approvals can be deleted by:
644 - the user who created the approval
645 - a user with both editProject and approveProject permissions
647 @permission: approveProject
648 """
649 project = fetch.project(session, project_id)
650 approval = session.get_one(ProjectApproval, approval_id)
651 if approval.project_id != project.id:
652 raise ValueError(
653 f"Approval ID {approval_id} does not belong to Project ID {project_id}"
654 )
655 if approval.user_id != user.id:
656 approval_user = session.get_one(User, approval.user_id)
657 authorise.check(user, perms.MANAGE_USERS, target_org=approval_user.organisation)
659 authorise.check(user, perms.PROJECT_APPROVE, project=project)
661 project.approvals.remove(approval)
662 session.flush()
664 AuditEvent.create(
665 session,
666 evt_types.PROJECT_APPROVAL_DELETED,
667 object_id=approval_id,
668 user=user,
669 project=project,
670 )
671 return [serial.ProjectApproval.model_validate(a) for a in project.approvals]