# [DEF:DatasetReviewRoutes:Module] # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @LAYER: API # @RATIONALE: Extracted from 2484-line monolith to satisfy INV_7 (400-line module limit). # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. from __future__ import annotations from datetime import datetime from typing import Any, List, Optional, Union, cast from fastapi import APIRouter, Depends, HTTPException, Query, Response, status from src.core.logger import belief_scope, logger from src.dependencies import get_config_manager, get_current_user, has_permission from src.models.auth import User from src.models.dataset_review import ( ApprovalState, ArtifactFormat, FieldProvenance, MappingMethod, PreviewStatus, ReadinessState, RecommendedAction, SessionStatus, ) from src.schemas.dataset_review import ( ClarificationAnswerDto, ClarificationStateResponse as _ClarificationStateResponseSchema, CompiledPreviewDto, ExecutionMappingDto, SemanticFieldEntryDto, SessionSummary, ValidationFindingDto, ) from src.services.dataset_review.clarification_engine import ( ClarificationAnswerCommand, ClarificationStateResult, ) from src.services.dataset_review.orchestrator import ( LaunchDatasetCommand, PreparePreviewCommand, StartSessionCommand, ) from src.services.dataset_review.repositories.session_repository import ( DatasetReviewSessionVersionConflictError, ) from src.api.routes.dataset_review_pkg._dependencies import ( BatchApproveMappingRequest, BatchApproveSemanticRequest, ClarificationAnswerRequest, ClarificationAnswerResultResponse, ClarificationStateResponse, ExportArtifactResponse, FeedbackRequest, FeedbackResponse, FieldSemanticUpdateRequest, LaunchDatasetResponse, MappingCollectionResponse, PreviewEnqueueResultResponse, SessionCollectionResponse, StartSessionRequest, UpdateExecutionMappingRequest, UpdateSessionRequest, _build_documentation_export, _build_sql_lab_redirect_url, _build_validation_export, _commit_owned_session_mutation, _get_clarification_engine, _get_latest_clarification_session_or_404, _get_owned_field_or_404, _get_owned_mapping_or_404, _get_owned_session_or_404, _get_orchestrator, _get_repository, _prepare_owned_session_mutation, _record_session_event, _require_auto_review_flag, _require_clarification_flag, _require_execution_flag, _require_session_version_header, _serialize_clarification_state, _serialize_empty_clarification_state, _serialize_execution_mapping, _serialize_preview, _serialize_run_context, _serialize_semantic_field, _serialize_session_detail, _serialize_session_summary, _update_semantic_field_state, _build_session_version_conflict_http_exception, ) router = APIRouter(prefix="/api/dataset-orchestration", tags=["Dataset Orchestration"]) # [DEF:list_sessions:Function] # @COMPLEXITY: 3 # @PURPOSE: List resumable dataset review sessions for the current user. @router.get( "/sessions", response_model=SessionCollectionResponse, dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "READ")), ], ) async def list_sessions( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.list_sessions"): logger.reason( "Listing dataset review sessions", extra={"user_id": current_user.id, "page": page, "page_size": page_size}, ) sessions = repository.list_sessions_for_user(current_user.id) start = (page - 1) * page_size end = start + page_size items = [_serialize_session_summary(s) for s in sessions[start:end]] logger.reflect( "Session page assembled", extra={"user_id": current_user.id, "returned": len(items), "total": len(sessions)}, ) return SessionCollectionResponse( items=items, total=len(sessions), page=page, page_size=page_size, has_next=end < len(sessions), ) # [/DEF:list_sessions:Function] # [DEF:start_session:Function] # @COMPLEXITY: 4 # @PURPOSE: Start a new dataset review session from a Superset link or dataset selection. @router.post( "/sessions", response_model=SessionSummary, status_code=status.HTTP_201_CREATED, dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def start_session( request: StartSessionRequest, orchestrator=Depends(_get_orchestrator), current_user: User = Depends(get_current_user), ): with belief_scope("start_session"): logger.reason( "Starting dataset review session", extra={"user_id": current_user.id, "environment_id": request.environment_id}, ) try: result = orchestrator.start_session( StartSessionCommand( user=current_user, environment_id=request.environment_id, source_kind=request.source_kind, source_input=request.source_input, ) ) except ValueError as exc: logger.explore("Session start rejected", extra={"user_id": current_user.id, "error": str(exc)}) detail = str(exc) sc = status.HTTP_404_NOT_FOUND if detail == "Environment not found" else status.HTTP_400_BAD_REQUEST raise HTTPException(status_code=sc, detail=detail) from exc logger.reflect("Session started", extra={"session_id": result.session.session_id}) return _serialize_session_summary(result.session) # [/DEF:start_session:Function] # [DEF:get_session_detail:Function] # @COMPLEXITY: 3 # @PURPOSE: Return the full accessible dataset review session aggregate. @router.get( "/sessions/{session_id}", response_model=SessionSummary, dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "READ")), ], ) async def get_session_detail( session_id: str, repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.get_session_detail"): session = _get_owned_session_or_404(repository, session_id, current_user) return _serialize_session_detail(session) # [/DEF:get_session_detail:Function] # [DEF:update_session:Function] # @COMPLEXITY: 3 # @PURPOSE: Update resumable lifecycle status for an owned session. @router.patch( "/sessions/{session_id}", response_model=SessionSummary, dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def update_session( session_id: str, request: UpdateSessionRequest, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("update_session"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) session_record = cast(Any, session) session_record.status = request.status if request.status == SessionStatus.PAUSED: session_record.recommended_action = RecommendedAction.RESUME_SESSION elif request.status in {SessionStatus.ARCHIVED, SessionStatus.CANCELLED, SessionStatus.COMPLETED}: session_record.active_task_id = None _commit_owned_session_mutation(repository, session) _record_session_event( repository, session, current_user, event_type="session_status_updated", event_summary="Dataset review session lifecycle updated", event_details={"status": session_record.status.value, "version": session_record.version}, ) return _serialize_session_summary(session) # [/DEF:update_session:Function] # [DEF:delete_session:Function] # @COMPLEXITY: 3 # @PURPOSE: Archive or hard-delete a session owned by the current user. @router.delete( "/sessions/{session_id}", status_code=status.HTTP_204_NO_CONTENT, dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def delete_session( session_id: str, hard_delete: bool = Query(False), session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("delete_session"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) if hard_delete: _record_session_event(repository, session, current_user, event_type="session_deleted", event_summary="Session hard-deleted", event_details={"hard_delete": True}) repository.db.delete(session) repository.db.commit() return Response(status_code=status.HTTP_204_NO_CONTENT) session_record = cast(Any, session) session_record.status = SessionStatus.ARCHIVED session_record.active_task_id = None _commit_owned_session_mutation(repository, session) _record_session_event(repository, session, current_user, event_type="session_archived", event_summary="Session archived", event_details={"hard_delete": False, "version": session_record.version}) return Response(status_code=status.HTTP_204_NO_CONTENT) # [/DEF:delete_session:Function] # [DEF:export_documentation:Function] # @COMPLEXITY: 3 # @PURPOSE: Export documentation output for the current session. @router.get( "/sessions/{session_id}/exports/documentation", response_model=ExportArtifactResponse, dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "READ")), ], ) async def export_documentation( session_id: str, format: ArtifactFormat = Query(ArtifactFormat.JSON), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("export_documentation"): if format not in {ArtifactFormat.JSON, ArtifactFormat.MARKDOWN}: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Only json and markdown exports are supported") session = _get_owned_session_or_404(repository, session_id, current_user) payload = _build_documentation_export(session, format) return ExportArtifactResponse( artifact_id=f"documentation-{session.session_id}-{format.value}", session_id=session.session_id, artifact_type="documentation", format=format.value, storage_ref=payload["storage_ref"], created_by_user_id=current_user.id, content=payload["content"], ) # [/DEF:export_documentation:Function] # [DEF:export_validation:Function] # @COMPLEXITY: 3 # @PURPOSE: Export validation findings for the current session. @router.get( "/sessions/{session_id}/exports/validation", response_model=ExportArtifactResponse, dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "READ")), ], ) async def export_validation( session_id: str, format: ArtifactFormat = Query(ArtifactFormat.JSON), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("export_validation"): if format not in {ArtifactFormat.JSON, ArtifactFormat.MARKDOWN}: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Only json and markdown exports are supported") session = _get_owned_session_or_404(repository, session_id, current_user) payload = _build_validation_export(session, format) return ExportArtifactResponse( artifact_id=f"validation-{session.session_id}-{format.value}", session_id=session.session_id, artifact_type="validation_report", format=format.value, storage_ref=payload["storage_ref"], created_by_user_id=current_user.id, content=payload["content"], ) # [/DEF:export_validation:Function] # [DEF:get_clarification_state:Function] # @COMPLEXITY: 3 # @PURPOSE: Return the current clarification session summary and active question payload. @router.get( "/sessions/{session_id}/clarification", response_model=ClarificationStateResponse, dependencies=[ Depends(_require_auto_review_flag), Depends(_require_clarification_flag), Depends(has_permission("dataset:session", "READ")), ], ) async def get_clarification_state( session_id: str, repository=Depends(_get_repository), clarification_engine=Depends(_get_clarification_engine), current_user: User = Depends(get_current_user), ): with belief_scope("get_clarification_state"): session = _get_owned_session_or_404(repository, session_id, current_user) if not session.clarification_sessions: return _serialize_empty_clarification_state() cs = _get_latest_clarification_session_or_404(session) question = clarification_engine.build_question_payload(session) return _serialize_clarification_state( ClarificationStateResult(clarification_session=cs, current_question=question, session=session, changed_findings=[]) ) # [/DEF:get_clarification_state:Function] # [DEF:resume_clarification:Function] # @COMPLEXITY: 3 # @PURPOSE: Resume clarification mode on the highest-priority unresolved question. @router.post( "/sessions/{session_id}/clarification/resume", response_model=ClarificationStateResponse, dependencies=[ Depends(_require_auto_review_flag), Depends(_require_clarification_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def resume_clarification( session_id: str, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), clarification_engine=Depends(_get_clarification_engine), current_user: User = Depends(get_current_user), ): with belief_scope("resume_clarification"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) cs = _get_latest_clarification_session_or_404(session) question = clarification_engine.build_question_payload(session) return _serialize_clarification_state( ClarificationStateResult(clarification_session=cs, current_question=question, session=session, changed_findings=[]) ) # [/DEF:resume_clarification:Function] # [DEF:record_clarification_answer:Function] # @COMPLEXITY: 3 # @PURPOSE: Persist one clarification answer before advancing the active pointer. @router.post( "/sessions/{session_id}/clarification/answers", response_model=ClarificationAnswerResultResponse, dependencies=[ Depends(_require_auto_review_flag), Depends(_require_clarification_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def record_clarification_answer( session_id: str, request: ClarificationAnswerRequest, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), clarification_engine=Depends(_get_clarification_engine), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.record_clarification_answer"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) try: result = clarification_engine.record_answer( ClarificationAnswerCommand( session=session, question_id=request.question_id, answer_kind=request.answer_kind, answer_value=request.answer_value, user=current_user, ) ) except ValueError as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc return ClarificationAnswerResultResponse( clarification_state=_serialize_clarification_state(result), session=_serialize_session_summary(result.session), changed_findings=[ValidationFindingDto.model_validate(f, from_attributes=True) for f in result.changed_findings], ) # [/DEF:record_clarification_answer:Function] # [DEF:update_field_semantic:Function] # @COMPLEXITY: 3 # @PURPOSE: Apply one field-level semantic candidate decision or manual override. @router.patch( "/sessions/{session_id}/fields/{field_id}/semantic", response_model=SemanticFieldEntryDto, dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def update_field_semantic( session_id: str, field_id: str, request: FieldSemanticUpdateRequest, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.update_field_semantic"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) field = _get_owned_field_or_404(session, field_id) _update_semantic_field_state(field, request, changed_by="user") sr = cast(Any, session) _commit_owned_session_mutation(repository, session, refresh_targets=[field]) _record_session_event(repository, session, current_user, event_type="semantic_field_updated", event_summary="Semantic field decision persisted", event_details={"field_id": field.field_id, "version": sr.version}) return _serialize_semantic_field(field) # [/DEF:update_field_semantic:Function] # [DEF:lock_field_semantic:Function] # @COMPLEXITY: 3 # @PURPOSE: Lock one semantic field against later automatic overwrite. @router.post( "/sessions/{session_id}/fields/{field_id}/lock", response_model=SemanticFieldEntryDto, dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def lock_field_semantic( session_id: str, field_id: str, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.lock_field_semantic"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) field = _get_owned_field_or_404(session, field_id) field.is_locked = True field.last_changed_by = "user" sr = cast(Any, session) _commit_owned_session_mutation(repository, session, refresh_targets=[field]) _record_session_event(repository, session, current_user, event_type="semantic_field_locked", event_summary="Semantic field lock persisted", event_details={"field_id": field.field_id, "version": sr.version}) return _serialize_semantic_field(field) # [/DEF:lock_field_semantic:Function] # [DEF:unlock_field_semantic:Function] # @COMPLEXITY: 3 # @PURPOSE: Unlock one semantic field so later automated candidate application may replace it. @router.post( "/sessions/{session_id}/fields/{field_id}/unlock", response_model=SemanticFieldEntryDto, dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def unlock_field_semantic( session_id: str, field_id: str, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.unlock_field_semantic"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) field = _get_owned_field_or_404(session, field_id) field.is_locked = False field.last_changed_by = "user" if field.provenance == FieldProvenance.MANUAL_OVERRIDE: field.provenance = FieldProvenance.UNRESOLVED field.needs_review = True sr = cast(Any, session) _commit_owned_session_mutation(repository, session, refresh_targets=[field]) _record_session_event(repository, session, current_user, event_type="semantic_field_unlocked", event_summary="Semantic field unlock persisted", event_details={"field_id": field.field_id, "version": sr.version}) return _serialize_semantic_field(field) # [/DEF:unlock_field_semantic:Function] # [DEF:approve_batch_semantic_fields:Function] # @COMPLEXITY: 3 # @PURPOSE: Approve multiple semantic candidate decisions in one batch. @router.post( "/sessions/{session_id}/fields/semantic/approve-batch", response_model=List[SemanticFieldEntryDto], dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def approve_batch_semantic_fields( session_id: str, request: BatchApproveSemanticRequest, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.approve_batch_semantic_fields"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) updated = [] for item in request.items: field = _get_owned_field_or_404(session, item.field_id) _update_semantic_field_state(field, FieldSemanticUpdateRequest(candidate_id=item.candidate_id, lock_field=item.lock_field), changed_by="user") updated.append(field) sr = cast(Any, session) _commit_owned_session_mutation(repository, session, refresh_targets=list(updated)) _record_session_event(repository, session, current_user, event_type="semantic_fields_batch_approved", event_summary="Batch semantic approval persisted", event_details={"count": len(updated), "version": sr.version}) return [_serialize_semantic_field(f) for f in updated] # [/DEF:approve_batch_semantic_fields:Function] # [DEF:list_execution_mappings:Function] # @COMPLEXITY: 3 # @PURPOSE: Return the current mapping-review set for one accessible session. @router.get( "/sessions/{session_id}/mappings", response_model=MappingCollectionResponse, dependencies=[ Depends(_require_auto_review_flag), Depends(_require_execution_flag), Depends(has_permission("dataset:session", "READ")), ], ) async def list_execution_mappings( session_id: str, repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.list_execution_mappings"): session = _get_owned_session_or_404(repository, session_id, current_user) return MappingCollectionResponse(items=[_serialize_execution_mapping(m) for m in session.execution_mappings]) # [/DEF:list_execution_mappings:Function] # [DEF:update_execution_mapping:Function] # @COMPLEXITY: 3 # @PURPOSE: Persist one owner-authorized execution-mapping effective value override. @router.patch( "/sessions/{session_id}/mappings/{mapping_id}", response_model=ExecutionMappingDto, dependencies=[ Depends(_require_auto_review_flag), Depends(_require_execution_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def update_execution_mapping( session_id: str, mapping_id: str, request: UpdateExecutionMappingRequest, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.update_execution_mapping"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) mapping = _get_owned_mapping_or_404(session, mapping_id) if request.effective_value is None: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="effective_value is required") mapping.effective_value = request.effective_value mapping.mapping_method = MappingMethod(request.mapping_method or MappingMethod.MANUAL_OVERRIDE.value) mapping.transformation_note = request.transformation_note mapping.approval_state = ApprovalState.APPROVED mapping.approved_by_user_id = current_user.id mapping.approved_at = datetime.utcnow() session.last_activity_at = datetime.utcnow() session.recommended_action = RecommendedAction.GENERATE_SQL_PREVIEW if session.readiness_state in {ReadinessState.MAPPING_REVIEW_NEEDED, ReadinessState.COMPILED_PREVIEW_READY, ReadinessState.RUN_READY, ReadinessState.RUN_IN_PROGRESS}: session.readiness_state = ReadinessState.COMPILED_PREVIEW_READY for preview in session.previews: if preview.preview_status == PreviewStatus.READY: preview.preview_status = PreviewStatus.STALE sr = cast(Any, session) _commit_owned_session_mutation(repository, session, refresh_targets=[mapping]) _record_session_event(repository, session, current_user, event_type="execution_mapping_updated", event_summary="Mapping override persisted", event_details={"mapping_id": mapping.mapping_id, "version": sr.version}) return _serialize_execution_mapping(mapping) # [/DEF:update_execution_mapping:Function] # [DEF:approve_execution_mapping:Function] # @COMPLEXITY: 3 # @PURPOSE: Explicitly approve a warning-sensitive mapping transformation. @router.post( "/sessions/{session_id}/mappings/{mapping_id}/approve", response_model=ExecutionMappingDto, dependencies=[ Depends(_require_auto_review_flag), Depends(_require_execution_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def approve_execution_mapping( session_id: str, mapping_id: str, request: ApproveMappingRequest, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.approve_execution_mapping"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) mapping = _get_owned_mapping_or_404(session, mapping_id) mapping.approval_state = ApprovalState.APPROVED mapping.approved_by_user_id = current_user.id mapping.approved_at = datetime.utcnow() if request.approval_note: mapping.transformation_note = request.approval_note session.last_activity_at = datetime.utcnow() if session.readiness_state == ReadinessState.MAPPING_REVIEW_NEEDED: session.recommended_action = RecommendedAction.GENERATE_SQL_PREVIEW sr = cast(Any, session) _commit_owned_session_mutation(repository, session, refresh_targets=[mapping]) _record_session_event(repository, session, current_user, event_type="execution_mapping_approved", event_summary="Mapping approval persisted", event_details={"mapping_id": mapping.mapping_id, "version": sr.version}) return _serialize_execution_mapping(mapping) # [/DEF:approve_execution_mapping:Function] # [DEF:approve_batch_execution_mappings:Function] # @COMPLEXITY: 3 # @PURPOSE: Approve multiple warning-sensitive execution mappings in one batch. @router.post( "/sessions/{session_id}/mappings/approve-batch", response_model=List[ExecutionMappingDto], dependencies=[ Depends(_require_auto_review_flag), Depends(_require_execution_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def approve_batch_execution_mappings( session_id: str, request: BatchApproveMappingRequest, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.approve_batch_execution_mappings"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) updated = [] for mid in list(dict.fromkeys(request.mapping_ids)): mapping = _get_owned_mapping_or_404(session, mid) mapping.approval_state = ApprovalState.APPROVED mapping.approved_by_user_id = current_user.id mapping.approved_at = datetime.utcnow() if request.approval_note: mapping.transformation_note = request.approval_note updated.append(mapping) session.last_activity_at = datetime.utcnow() if session.readiness_state == ReadinessState.MAPPING_REVIEW_NEEDED: session.recommended_action = RecommendedAction.GENERATE_SQL_PREVIEW sr = cast(Any, session) _commit_owned_session_mutation(repository, session, refresh_targets=list(updated)) _record_session_event(repository, session, current_user, event_type="execution_mappings_batch_approved", event_summary="Batch mapping approval persisted", event_details={"count": len(updated), "version": sr.version}) return [_serialize_execution_mapping(m) for m in updated] # [/DEF:approve_batch_execution_mappings:Function] # [DEF:trigger_preview_generation:Function] # @COMPLEXITY: 3 # @PURPOSE: Trigger Superset-side preview compilation for the current owned execution context. @router.post( "/sessions/{session_id}/preview", response_model=Union[CompiledPreviewDto, PreviewEnqueueResultResponse], dependencies=[ Depends(_require_auto_review_flag), Depends(_require_execution_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def trigger_preview_generation( session_id: str, response: Response, orchestrator=Depends(_get_orchestrator), repository=Depends(_get_repository), session_version: int = Depends(_require_session_version_header), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.trigger_preview_generation"): _prepare_owned_session_mutation(repository, session_id, current_user, session_version) try: result = orchestrator.prepare_launch_preview( PreparePreviewCommand(user=current_user, session_id=session_id, expected_version=session_version) ) except DatasetReviewSessionVersionConflictError as exc: raise _build_session_version_conflict_http_exception(exc) from exc except ValueError as exc: detail = str(exc) sc = status.HTTP_404_NOT_FOUND if detail in {"Session not found", "Environment not found"} else status.HTTP_409_CONFLICT if detail.startswith("Preview blocked:") else status.HTTP_400_BAD_REQUEST raise HTTPException(status_code=sc, detail=detail) from exc if result.preview.preview_status == PreviewStatus.PENDING: response.status_code = status.HTTP_202_ACCEPTED return PreviewEnqueueResultResponse( session_id=result.session.session_id, session_version=int(getattr(result.session, "version", 0) or 0), preview_status=result.preview.preview_status.value, task_id=None, ) response.status_code = status.HTTP_200_OK return _serialize_preview(result.preview, session_version_fallback=int(getattr(result.session, "version", 0) or 0)) # [/DEF:trigger_preview_generation:Function] # [DEF:launch_dataset:Function] # @COMPLEXITY: 3 # @PURPOSE: Execute the current owned session launch handoff and return audited SQL Lab run context. @router.post( "/sessions/{session_id}/launch", response_model=LaunchDatasetResponse, status_code=status.HTTP_201_CREATED, dependencies=[ Depends(_require_auto_review_flag), Depends(_require_execution_flag), Depends(has_permission("dataset:execution:launch", "EXECUTE")), ], ) async def launch_dataset( session_id: str, orchestrator=Depends(_get_orchestrator), repository=Depends(_get_repository), session_version: int = Depends(_require_session_version_header), config_manager=Depends(get_config_manager), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.launch_dataset"): _prepare_owned_session_mutation(repository, session_id, current_user, session_version) try: result = orchestrator.launch_dataset( LaunchDatasetCommand(user=current_user, session_id=session_id, expected_version=session_version) ) except DatasetReviewSessionVersionConflictError as exc: raise _build_session_version_conflict_http_exception(exc) from exc except ValueError as exc: detail = str(exc) sc = status.HTTP_404_NOT_FOUND if detail in {"Session not found", "Environment not found"} else status.HTTP_409_CONFLICT if detail.startswith("Launch blocked:") else status.HTTP_400_BAD_REQUEST raise HTTPException(status_code=sc, detail=detail) from exc environment = config_manager.get_environment(result.session.environment_id) env_url = getattr(environment, "url", "") if environment is not None else "" return LaunchDatasetResponse( run_context=_serialize_run_context(result.run_context), redirect_url=_build_sql_lab_redirect_url(environment_url=env_url, sql_lab_session_ref=result.run_context.sql_lab_session_ref), ) # [/DEF:launch_dataset:Function] # [DEF:record_field_feedback:Function] # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for AI-assisted semantic field content. @router.post( "/sessions/{session_id}/fields/{field_id}/feedback", response_model=FeedbackResponse, dependencies=[ Depends(_require_auto_review_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def record_field_feedback( session_id: str, field_id: str, request: FeedbackRequest, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.record_field_feedback"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) field = _get_owned_field_or_404(session, field_id) field.user_feedback = request.feedback sr = cast(Any, session) _commit_owned_session_mutation(repository, session) _record_session_event(repository, session, current_user, event_type="semantic_field_feedback_recorded", event_summary="Feedback persisted", event_details={"field_id": field.field_id, "feedback": request.feedback, "version": sr.version}) return FeedbackResponse(target_id=field.field_id, feedback=request.feedback) # [/DEF:record_field_feedback:Function] # [DEF:record_clarification_feedback:Function] # @COMPLEXITY: 3 # @PURPOSE: Persist thumbs up/down feedback for clarification question/answer content. @router.post( "/sessions/{session_id}/clarification/questions/{question_id}/feedback", response_model=FeedbackResponse, dependencies=[ Depends(_require_auto_review_flag), Depends(_require_clarification_flag), Depends(has_permission("dataset:session", "MANAGE")), ], ) async def record_clarification_feedback( session_id: str, question_id: str, request: FeedbackRequest, session_version: int = Depends(_require_session_version_header), repository=Depends(_get_repository), current_user: User = Depends(get_current_user), ): with belief_scope("dataset_review.record_clarification_feedback"): session = _prepare_owned_session_mutation(repository, session_id, current_user, session_version) cs = _get_latest_clarification_session_or_404(session) question = next((q for q in cs.questions if q.question_id == question_id), None) if question is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Clarification question not found") if question.answer is None: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Clarification answer not found") question.answer.user_feedback = request.feedback sr = cast(Any, session) _commit_owned_session_mutation(repository, session) _record_session_event(repository, session, current_user, event_type="clarification_feedback_recorded", event_summary="Feedback persisted", event_details={"question_id": question.question_id, "feedback": request.feedback, "version": sr.version}) return FeedbackResponse(target_id=question.question_id, feedback=request.feedback) # [/DEF:record_clarification_feedback:Function] # [/DEF:DatasetReviewRoutes:Module]