Coverage for postrfp/buyer/api/endpoints/issues.py: 100%
97 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-22 21:34 +0000
1"""
2Operations for managing Issues (invitations to respond) and their lifecycle.
3"""
5from datetime import datetime
6from typing import Optional
8from sqlalchemy.orm.session import Session
10from postrfp.authorisation import perms
11from postrfp.shared import fetch, serial
12from postrfp.shared.pager import Pager
13from postrfp.shared.decorators import http
14from postrfp.shared.exceptions import AuthorizationFailure
15from postrfp.shared.issue_transition import change_issue_status
16from postrfp.buyer.api import authorise
17from postrfp.model import (
18 Organisation,
19 Issue,
20 AuditEvent,
21 Project,
22 Participant,
23)
24from postrfp.model.humans import User
27@http
28def get_vendor(session: Session, user: User, q_org_id: str) -> serial.Supplier:
29 """
30 Vendor organisation profile (Consultant users only): organisation details plus
31 Issues involving this buyer and the vendor's users.
32 """
34 if not user.organisation.is_consultant:
35 raise AuthorizationFailure("Only Consultant users can view Vendor details")
37 # Will raise No Result Found if not in Suppliers list
38 vendor = user.organisation.suppliers.filter(Organisation.id == q_org_id).one()
40 issues = fetch.issues_for_respondent(session, user, q_org_id)
42 return serial.Supplier(
43 issues=[serial.Issue.model_validate(i) for i in issues],
44 organisation=serial.Organisation.model_validate(vendor),
45 users=[serial.User.model_validate(u) for u in vendor.users],
46 )
49@http
50def get_project_issue(
51 session: Session, user: User, project_id: int, issue_id: int
52) -> serial.Issue:
53 """
54 Retrieve a single Issue within a Project. Permission: PROJECT_ACCESS.
55 """
56 project = fetch.project(session, project_id)
57 authorise.check(user, perms.PROJECT_ACCESS, project=project)
58 issue = project.get_issue(issue_id)
59 return serial.Issue.model_validate(issue)
62@http
63def get_project_issues(session: Session, user: User, project_id: int) -> serial.Issues:
64 """
65 List all Issues for a Project (no pagination). Permission: PROJECT_ACCESS.
66 """
67 project = fetch.project(session, project_id)
68 authorise.check(user, perms.PROJECT_ACCESS, project=project)
69 fo = serial.Issue.model_validate
70 return serial.Issues([fo(i) for i in project.issues])
73@http
74def put_project_issue(
75 session: Session, user: User, project_id: int, issue_id: int, issue_doc: dict
76):
77 """
78 Update mutable Issue fields (e.g. label, deadlines, win/loss exposure). Status
79 changes are NOT allowed here (use status endpoint). Emits ISSUE_UPDATED audit event
80 with per‑field change log.
81 """
82 project = fetch.project(session, project_id)
83 authorise.check(user, perms.PROJECT_ACCESS, project=project)
84 issue = project.get_issue(issue_id)
85 authorise.check(user, perms.ISSUE_UPDATE, issue=issue)
87 change_list = []
89 for field, value in issue_doc.items():
90 change_list.append((field, getattr(issue, field), value))
91 setattr(issue, field, value)
93 AuditEvent.create(
94 session,
95 "ISSUE_UPDATED",
96 project=issue.project,
97 issue_id=issue.id,
98 user_id=user.id,
99 org_id=user.organisation.id,
100 object_id=issue.id,
101 timestamp=datetime.now(),
102 private=True,
103 change_list=change_list,
104 )
107@http
108def get_issue(session: Session, user: User, issue_id: int) -> serial.Issue:
109 """
110 Retrieve an Issue by ID (requires access to its Project). Permission: PROJECT_ACCESS.
111 """
112 issue = fetch.issue(session, issue_id)
113 authorise.check(user, perms.PROJECT_ACCESS, project=issue.project)
114 return serial.Issue.model_validate(issue)
117@http
118def post_project_issue(
119 session: Session, user: User, project_id: int, new_issue_doc: serial.NewIssue
120) -> serial.Id:
121 """
122 Create a new Issue (invitation). Exactly one of respondent_id or respondent_email
123 must be supplied. Starts in Not Sent status. Emits ISSUE_CREATED audit event with
124 initial field values.
125 """
126 project = fetch.project(session, project_id)
127 authorise.check(user, perms.ISSUE_CREATE, project=project)
128 label = new_issue_doc.label
129 respondent_id = new_issue_doc.respondent_id
130 if respondent_id is None:
131 issue = Issue(
132 respondent_email=new_issue_doc.respondent_email,
133 label=label,
134 status="Not Sent",
135 )
136 else:
137 org: Organisation = (
138 session.query(Organisation).filter_by(id=respondent_id).one()
139 )
140 issue = Issue(respondent=org, label=label, status="Not Sent")
142 project.add_issue(issue)
143 session.flush()
144 evt = AuditEvent.create(
145 session, "ISSUE_CREATED", issue_id=issue.id, project=project, user=user
146 )
147 for k in serial.NewIssue.model_fields.keys():
148 if k not in ("respondent_id", "respondent_email"):
149 evt.add_change(k, "", getattr(issue, k))
150 session.add(evt)
152 return serial.Id(id=issue.id)
155@http
156def delete_project_issue(
157 session: Session, user: User, project_id: int, issue_id: int
158) -> None:
159 """
160 Delete an Issue (permitted only while it is Not Sent, Declined, or Retracted).
161 Emits ISSUE_DELETED audit event.
162 """
163 project = fetch.project(session, project_id)
164 authorise.check(user, perms.PROJECT_ACCESS, project=project)
165 issue = project.get_issue(issue_id)
166 authorise.check(user, perms.ISSUE_DELETE, issue=issue)
167 session.delete(issue)
168 evt = AuditEvent.create(
169 session, "ISSUE_DELETED", issue_id=issue_id, project=project, user=user
170 )
171 session.add(evt)
174@http
175def post_issue_status(
176 session: Session, user: User, issue_id: int, issue_status_doc: serial.IssueStatus
177) -> None:
178 """
179 Change Issue status (uses server‑side transition rules & permission checks; may
180 set timestamps and validate mandatory answers / deadlines). Emits appropriate
181 audit event via change_issue_status.
182 """
183 issue = fetch.issue(session, issue_id)
184 authorise.check(user, perms.ISSUE_SUBMIT, project=issue.project)
185 change_issue_status(issue, user, issue_status_doc.new_status)
188@http
189def get_issues(
190 session: Session,
191 user: User,
192 issue_sort: str,
193 sort_order: str,
194 q_org_id: Optional[str] = None,
195 issue_status: Optional[str] = None,
196 project_ids: Optional[set[int]] = None,
197 pager: Optional[Pager] = None,
198) -> serial.IssuesList:
199 """
200 Paginated list of Issues visible to the user across all participating Projects.
201 Supports filtering by respondent (q_org_id) and status; sortable by deadline,
202 submitted_date, or issue_date (asc/desc).
204 "orgId" query parameter is the ID of the respondent organisation to filter by
205 "projectIds" is an array of GET parameters to filter by project ID.
206 """
207 if pager is None:
208 pager = Pager(page=1, page_size=100)
210 cols = (
211 Issue.id.label("issue_id"),
212 Issue.respondent_id,
213 Issue.respondent_email,
214 Project.id.label("project_id"),
215 Issue.status,
216 Issue.deadline,
217 Issue.submitted_date,
218 Issue.issue_date,
219 Project.title.label("project_title"),
220 Issue.winloss_exposed,
221 Issue.winloss_expiry,
222 Issue.label,
223 )
225 sort_cols = {
226 "deadline": Issue.deadline,
227 "submitted_date": Issue.submitted_date,
228 "issue_date": Issue.issue_date,
229 }
231 order_column = sort_cols[issue_sort]
232 ordering = order_column.desc() if sort_order == "desc" else order_column.asc()
234 iq = (
235 session.query(*cols)
236 .join(Project)
237 .join(Participant)
238 .filter(Participant.org_id == user.org_id)
239 )
241 if q_org_id:
242 iq = iq.filter(Issue.respondent_id == q_org_id)
244 if issue_status:
245 iq = iq.filter(Issue.status == issue_status)
247 if project_ids:
248 iq = iq.filter(Project.id.in_(project_ids))
250 total_records = iq.count()
252 records = iq.order_by(ordering).slice(pager.startfrom, pager.goto).all()
253 return serial.IssuesList(
254 data=[serial.ListIssue.model_validate(i) for i in records],
255 pagination=pager.as_pagination(total_records, len(records)),
256 )