Method list

methods/MIRIX/mirix/orm/sqlalchemy_base.py:195–359

List records with cursor-based pagination, ordering by created_at. Cursor is an ID, but pagination is based on the cursor object's created_at value. Args: db_session: SQLAlchemy session cursor: ID of the last item seen (for pagination) st

(
        cls,
        *,
        db_session: AsyncSession,
        cursor: Optional[str] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        limit: Optional[int] = 50,
        query_text: Optional[str] = None,
        query_embedding: Optional[List[float]] = None,
        ascending: bool = True,
        tags: Optional[List[str]] = None,
        match_all_tags: bool = False,
        actor: Optional["Client"] = None,
        access: Optional[List[Literal["read", "write", "admin"]]] = ["read"],
        access_type: AccessType = AccessType.ORGANIZATION,
        join_model: Optional[Base] = None,
        join_conditions: Optional[Union[Tuple, List]] = None,
        **kwargs,
    )

    @handle_db_timeout
    @retry_db_operation(max_retries=3, base_delay=0.1, max_delay=2.0)
    async def list(
        cls,
        *,
        db_session: AsyncSession,
        cursor: Optional[str] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        limit: Optional[int] = 50,
        query_text: Optional[str] = None,
        query_embedding: Optional[List[float]] = None,
        ascending: bool = True,
        tags: Optional[List[str]] = None,
        match_all_tags: bool = False,
        actor: Optional["Client"] = None,
        access: Optional[List[Literal["read", "write", "admin"]]] = ["read"],
        access_type: AccessType = AccessType.ORGANIZATION,
        join_model: Optional[Base] = None,
        join_conditions: Optional[Union[Tuple, List]] = None,
        **kwargs,
    ) -> List["SqlalchemyBase"]:
        """
        List records with cursor-based pagination, ordering by created_at.
        Cursor is an ID, but pagination is based on the cursor object's created_at value.

        Args:
            db_session: SQLAlchemy session
            cursor: ID of the last item seen (for pagination)
            start_date: Filter items after this date
            end_date: Filter items before this date
            limit: Maximum number of items to return
            query_text: Text to search for
            query_embedding: Vector to search for similar embeddings
            ascending: Sort direction
            tags: List of tags to filter by
            match_all_tags: If True, return items matching all tags. If False, match any tag.
            **kwargs: Additional filters to apply
        """
        if start_date and end_date and start_date > end_date:
            raise ValueError("start_date must be earlier than or equal to end_date")

        logger.debug("Listing %s with kwarg filters %s", cls.__name__, kwargs)
        session = db_session
        # If cursor provided, get the reference object
        cursor_obj = None
        if cursor:
            cursor_obj = await session.get(cls, cursor)
            if not cursor_obj:
                raise NoResultFound(f"No {cls.__name__} found with id {cursor}")

        query = select(cls)

        if join_model and join_conditions:
            query = query.join(join_model, and_(*join_conditions))

        # Apply access predicate if actor is provided
        if actor:
            query = cls.apply_access_predicate(query, actor, access, access_type)