Coverage for postrfp/ref/handlers/subjects.py: 100%
73 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-22 21:34 +0000
1"""
2HTTP endpoints for Subject CRUD operations
3"""
5from typing import Optional, List
6from sqlalchemy.orm import Session
7from sqlalchemy import or_
9from postrfp.shared.pager import Pager
10from postrfp.shared.decorators import http
11from postrfp.model import User
12from postrfp.model.ref import Subject, SubjectType, Content
13from postrfp.shared.serial.refmodels import (
14 SubjectDocument,
15 ListResponse,
16 DeletionResponse,
17 SubjectSummary,
18)
21@http
22def get_subjects(
23 session: Session,
24 user: User,
25 q_name: Optional[str] = None,
26 q_type: Optional[str] = None,
27 q_parent_id: Optional[int] = None,
28 pager: Optional[Pager] = None,
29) -> ListResponse:
30 """
31 List subjects with optional filtering
33 Subjects represent entities that content items can be about, such as countries,
34 organizations, markets, industry sectors, or products. They can have hierarchical
35 relationships and carry their own metadata and permission systems.
37 **Query Parameters:**
38 - `q_name`: Filter subjects by name (case-insensitive partial match)
39 - `q_type`: Filter by subject type (country, organization, market, sector, product, other)
40 - `q_parent_id`: Filter to show only direct children of the specified parent subject
41 - Standard pagination parameters supported
43 **Returns:** Paginated list of subjects matching the filter criteria
45 **Examples:**
46 - Get all banking organizations: `?q_type=organization&q_name=bank`
47 - Get all cities in a country: `?q_parent_id=123&q_type=location`
48 """
49 if pager is None:
50 pager = Pager(page=1, page_size=20)
52 query = session.query(Subject)
54 # Apply filters
55 if q_name:
56 query = query.filter(Subject.name.ilike(f"%{q_name}%"))
58 if q_type:
59 try:
60 subject_type = SubjectType(q_type.lower())
61 query = query.filter(Subject.subject_type == subject_type)
62 except ValueError:
63 pass # Invalid type, ignore filter
65 if q_parent_id is not None:
66 query = query.filter(Subject.parent_id == q_parent_id)
68 # Get total count for pagination
69 total_count = query.count()
71 # Apply pagination - use startfrom instead of offset
72 subjects = query.offset(pager.startfrom).limit(pager.page_size).all()
74 # Convert to response models
75 items = [SubjectSummary.model_validate(subject) for subject in subjects]
76 pagination = pager.as_pagination(total_count, len(items))
78 return ListResponse(items=items, pagination=pagination)
81@http
82def get_subject(session: Session, user: User, subject_id: int) -> SubjectDocument:
83 """
84 Get a specific subject by ID
86 Retrieves complete information about a subject including its name, type,
87 description, hierarchical position, managing organization, and metadata.
89 **Path Parameters:**
90 - `subject_id`: Unique identifier of the subject
92 **Returns:** Complete subject information
94 **Raises:**
95 - `NoResultFound`: If the subject ID does not exist
97 **Subject Types:**
98 - `country`: Geographic countries or regions
99 - `organization`: Companies, banks, institutions
100 - `market`: Stock exchanges, trading venues
101 - `sector`: Industry classifications
102 - `product`: Financial products or services
103 - `other`: Custom subject types
104 """
105 subject = session.get_one(Subject, subject_id)
106 return SubjectDocument.model_validate(subject)
109@http
110def post_subject(
111 session: Session, user: User, subject_data: SubjectDocument
112) -> SubjectDocument:
113 """
114 Create a new subject
116 Creates a new subject that can be used to categorize content items. Subjects
117 can have hierarchical relationships with other subjects and their own permission
118 systems for controlling access to related content.
120 **Request Body:** Subject data including:
121 - `name`: Display name for the subject (required)
122 - `code`: Optional short identifier or symbol
123 - `description`: Optional detailed description
124 - `subject_type`: Type classification (required)
125 - `parent_id`: Optional parent subject for hierarchical relationships
126 - `managing_org_id`: Optional organization that manages this subject
127 - `subject_metadata`: Optional key-value metadata (country codes, market identifiers, etc.)
129 **Returns:** Created subject with assigned ID and timestamps
131 **Permission Impact:**
132 When content references this subject, the subject's permissions may affect
133 content access depending on the content's permission inheritance mode.
135 **Hierarchical Relationships:**
136 Use `parent_id` to create hierarchies like:
137 - City → State → Country
138 - Subsidiary → Parent Company
139 - Sub-sector → Industry Sector
140 """
141 # Convert string to enum
142 try:
143 subject_type_enum = SubjectType(subject_data.subject_type.lower())
144 except ValueError:
145 raise ValueError(f"Invalid subject_type: {subject_data.subject_type}")
147 # Create new subject
148 subject = Subject(
149 name=subject_data.name,
150 code=subject_data.code,
151 description=subject_data.description,
152 subject_type=subject_type_enum,
153 parent_id=subject_data.parent_id,
154 managing_org_id=subject_data.managing_org_id,
155 subject_metadata=subject_data.subject_metadata or {},
156 )
158 session.add(subject)
159 session.flush() # Get the ID
161 return SubjectDocument.model_validate(subject)
164@http
165def put_subject(
166 session: Session, user: User, subject_id: int, subject_data: SubjectDocument
167) -> SubjectDocument:
168 """
169 Update an existing subject
171 Modifies all fields of an existing subject. Changes to subject properties
172 may affect access to content that references this subject, depending on
173 those content items' permission inheritance settings.
175 **Path Parameters:**
176 - `subject_id`: ID of the subject to update
178 **Request Body:** Complete subject data (all fields will be updated)
180 **Returns:** Updated subject information
182 **Raises:**
183 - `NoResultFound`: If the subject ID does not exist
184 - `ValueError`: If invalid subject_type provided
186 **Important Notes:**
187 - Changing hierarchical relationships (`parent_id`) affects content categorization
188 - Updates to managing organization may impact permission inheritance
189 - Metadata changes can affect external system integrations
190 """
191 subject = session.get_one(Subject, subject_id)
193 # Convert string to enum
194 try:
195 subject_type_enum = SubjectType(subject_data.subject_type.lower())
196 except ValueError:
197 raise ValueError(f"Invalid subject_type: {subject_data.subject_type}")
199 # Update fields
200 subject.name = subject_data.name
201 subject.code = subject_data.code
202 subject.description = subject_data.description
203 subject.subject_type = subject_type_enum
204 subject.parent_id = subject_data.parent_id
205 subject.managing_org_id = subject_data.managing_org_id
206 subject.subject_metadata = subject_data.subject_metadata or {}
208 session.flush()
210 return SubjectDocument.model_validate(subject)
213@http
214def delete_subject(session: Session, user: User, subject_id: int) -> DeletionResponse:
215 """
216 Delete a subject
218 Permanently removes a subject from the system. This operation is protected
219 by referential integrity checks to prevent data inconsistency.
221 **Path Parameters:**
222 - `subject_id`: ID of the subject to delete
224 **Returns:** Success confirmation with deletion details
226 **Raises:**
227 - `NoResultFound`: If the subject ID does not exist
228 - `ValueError`: If the subject has dependencies that prevent deletion:
229 - Subject has child subjects in the hierarchy
230 - Subject is referenced as primary subject by content items
231 - Subject is referenced in content subject associations
233 **Protection Mechanisms:**
234 The system prevents deletion of subjects that would break data integrity:
236 1. **Hierarchical Protection:** Cannot delete subjects with children
237 2. **Content Protection:** Cannot delete subjects referenced by content
238 3. **Permission Protection:** Associated permissions are cascade-deleted
240 **Recommendation:**
241 Before deleting, consider if the subject should be:
242 - Marked as inactive/deprecated instead of deleted
243 - Replaced by transferring references to another subject
244 - Have its child subjects reassigned to a different parent
245 """
246 subject = session.get_one(Subject, subject_id)
248 # Check if subject has children
249 children_count = (
250 session.query(Subject).filter(Subject.parent_id == subject_id).count()
251 )
252 if children_count > 0:
253 raise ValueError(f"Cannot delete subject with {children_count} child subjects")
255 # Check if subject is used by content
256 content_count = (
257 session.query(Content)
258 .filter(
259 or_(
260 Content.primary_subject_id == subject_id,
261 Content.subjects.any(id=subject_id),
262 )
263 )
264 .count()
265 )
267 if content_count > 0:
268 raise ValueError(
269 f"Cannot delete subject referenced by {content_count} content items"
270 )
272 session.delete(subject)
274 return DeletionResponse(success=True, message=f"Subject {subject_id} deleted")
277@http
278def get_subject_children(
279 session: Session, user: User, subject_id: int
280) -> List[SubjectDocument]:
281 """
282 Get all direct children of a subject
284 Retrieves subjects that have the specified subject as their direct parent
285 in the hierarchical relationship. This is useful for building tree views
286 or navigation interfaces.
288 **Path Parameters:**
289 - `subject_id`: ID of the parent subject
291 **Returns:** List of direct child subjects (not recursive)
293 **Use Cases:**
294 - Building hierarchical navigation menus
295 - Showing organizational structure
296 - Geographic drill-down (country → states → cities)
297 - Industry classification browsing
299 **Note:**
300 This endpoint only returns direct children. For the complete hierarchy,
301 you may need to make recursive calls or use the main subjects list endpoint
302 with appropriate filtering.
304 **Example Hierarchies:**
305 - Geographic: USA → California → San Francisco
306 - Corporate: Holding Company → Subsidiaries → Divisions
307 - Market: Region → Exchange → Trading Segments
308 """
309 children = session.query(Subject).filter(Subject.parent_id == subject_id).all()
310 return [SubjectDocument.model_validate(child) for child in children]