INNER CODE UNIT · Python

query_id_of

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

def query_id_of(message, user):
    match = re.fullmatch(r"query:(\d+):.+", message.channel or "")
    if match is None or not (user.is_bot and user.username == SKYBOT_USERNAME):
        return None
    return int(match[1])


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."""

View source record →