def case_closed_status_flow()

in src/dispatch/case/flows.py [0:0]

• 22 lines of code
• 15 McCabe index (conditional complexity)

def case_closed_status_flow(case: Case, db_session=None):
    """Runs the case closed transition flow."""
    # we set the closed_at time
    case.closed_at = datetime.utcnow()
    db_session.add(case)
    db_session.commit()

    # Archive the conversation if there is a dedicated channel
    if case.dedicated_channel:
        conversation_flows.archive_conversation(subject=case, db_session=db_session)

    # Check if the case visibility is open
    if case.visibility != Visibility.open:
        return

    # Get the active storage plugin for the case's project
    storage_plugin = plugin_service.get_active_instance(
        db_session=db_session, project_id=case.project.id, plugin_type="storage"
    )

    if not storage_plugin:
        return

    # we update the ticket
    ticket_flows.update_case_ticket(case=case, db_session=db_session)

    # Open document access if configured
    if storage_plugin.configuration.open_on_close:
        for document in case.documents:
            document_flows.open_document_access(document=document, db_session=db_session)

    # Mark documents as read-only if configured
    if storage_plugin.configuration.read_only:
        for document in case.documents:
            document_flows.mark_document_as_readonly(document=document, db_session=db_session)

    if case.dedicated_channel:
        # we send a direct message to all participants asking them
        # to rate and provide feedback about the case
        send_case_rating_feedback_message(case, db_session)