INNER CODE UNIT · Python

requester_of

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

def requester_of(session, message, user):
    query_id = query_id_of(message, user)
    if query_id is None:
        return message.user_id
    return session.scalar(
        sa.select(AssistantQuery.owner_id).where(AssistantQuery.id == query_id)
    )


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
        )

View source record →