Method read

methods/letta/orm/sqlalchemy_base.py:267–301

The primary accessor for an ORM record. Args: db_session: the database session to use when retrieving the record identifier: the identifier of the record to read, can be the id string or the UUID object for backwards compatibility actor: if specified, resu

(
        cls,
        db_session: "Session",
        identifier: Optional[str] = None,
        actor: Optional["User"] = None,
        access: Optional[List[Literal["read", "write", "admin"]]] = ["read"],
        access_type: AccessType = AccessType.ORGANIZATION,
        **kwargs,
    )

    @classmethod
    @handle_db_timeout
    def read(
        cls,
        db_session: "Session",
        identifier: Optional[str] = None,
        actor: Optional["User"] = None,
        access: Optional[List[Literal["read", "write", "admin"]]] = ["read"],
        access_type: AccessType = AccessType.ORGANIZATION,
        **kwargs,
    ) -> "SqlalchemyBase":
        """The primary accessor for an ORM record.
        Args:
            db_session: the database session to use when retrieving the record
            identifier: the identifier of the record to read, can be the id string or the UUID object for backwards compatibility
            actor: if specified, results will be scoped only to records the user is able to access
            access: if actor is specified, records will be filtered to the minimum permission level for the actor
            kwargs: additional arguments to pass to the read, used for more complex objects
        Returns:
            The matching object
        Raises:
            NoResultFound: if the object is not found
        """
        # this is ok because read_multiple will check if the
        identifiers = [] if identifier is None else [identifier]
        found = cls.read_multiple(db_session, identifiers, actor, access, access_type, **kwargs)
        if len(found) == 0:
            # for backwards compatibility.
            conditions = []
            if identifier:
                conditions.append(f"id={identifier}")
            if actor:
                conditions.append(f"access level in {access} for {actor}")
            if hasattr(cls, "is_deleted"):
                conditions.append("is_deleted=False")
            raise NoResultFound(f"{cls.__name__} not found with {', '.join(conditions if conditions else ['no conditions'])}")
        return found[0]

    @classmethod
    @handle_db_timeout