INNER CODE UNIT · Python

triggering_analysis

skyportal/skyportal · services/assistant/app.py:391

def triggering_analysis(session, message, user):
    """The latest completed analysis of the service a query run on a source matches."""
    query_id = query_id_of(message, user)
    if query_id is None or message.context_type != "source":
        return None
    service_match = session.scalar(
        sa.select(AssistantQuery.analysis_service_match).where(
            AssistantQuery.id == query_id
        )
    )
    if not service_match:
        return None
    return session.scalar(
        sa.select(ObjAnalysis.id)
        .join(AnalysisService, AnalysisService.id == ObjAnalysis.analysis_service_id)
        .where(
            ObjAnalysis.obj_id == message.context_id,
            ObjAnalysis.status == "completed",

View source record →