cass: origin-visit-status-get-latest

ActivePublic

Actions

def origin_visit_status_get_latest(
        self, origin: str, visit: int,
        allowed_statuses: Optional[List[str]] = None,
        require_snapshot: bool = False
) -> Optional[Dict[str, Any]]:
    """Given an origin visit id, return its latest origin_visit_status.

    """
    query = (
        "SELECT * FROM origin_visit_status "
        "WHERE origin = %s AND visit = %s "
        "ORDER BY date DESC"
    )
    rows = list(self._execute_with_retries(query, [origin, visit]))

    # filtering is done python side as we cannot do it server side
    if allowed_statuses:
        rows = [row for row in rows if row.status in allowed_statuses]
    if require_snapshot:
        rows = [row for row in rows if row.snapshot is not None]
    if not rows:
        return None
    # visit_status = max(rows, key=lambda v: (v.date, v.visit), default=None)
    return self._format_origin_visit_status_row(rows[0])