Method _list_objects

sdk/python/feast/infra/registry/sql.py:1581–1628

Args: proto_only: If True, return raw protobuf objects without calling from_proto(). Used by proto() to build the RegistryProto cache efficiently — avoids the from_proto()/to_proto() round-trip and works uniformly for all object ty

(
        self,
        table: Table,
        project: str,
        proto_class: Any,
        python_class: Any,
        proto_field_name: str,
        tags: Optional[dict[str, str]] = None,
        proto_only: bool = False,
        skip_udf: bool = False,
    )

            return None

    def _list_objects(
        self,
        table: Table,
        project: str,
        proto_class: Any,
        python_class: Any,
        proto_field_name: str,
        tags: Optional[dict[str, str]] = None,
        proto_only: bool = False,
        skip_udf: bool = False,
    ):
        """
        Args:
            proto_only: If True, return raw protobuf objects without calling
                from_proto(). Used by proto() to build the RegistryProto cache
                efficiently — avoids the from_proto()/to_proto() round-trip and
                works uniformly for all object types (entities, data sources, etc.).
            skip_udf: If True, call from_proto() but skip deserializing UDFs
                (dill.loads). Returns Python objects suitable for filtering and
                display without requiring the UDF's source module to be installed.
                Only relevant for feature view types that contain UDFs.
        """
        import inspect

        supports_skip_udf = (
            skip_udf
            and "skip_udf" in inspect.signature(python_class.from_proto).parameters
        )

        with self.read_engine.begin() as conn:
            stmt = select(table).where(table.c.project_id == project)
            rows = conn.execute(stmt).all()
            if rows:
                objects = []
                for row in rows:
                    proto = proto_class.FromString(row._mapping[proto_field_name])
                    if proto_only:
                        objects.append(proto)
                    else:
                        obj = (
                            python_class.from_proto(proto, skip_udf=True)
                            if supports_skip_udf
                            else python_class.from_proto(proto)
                        )
                        if utils.has_all_tags(obj.tags, tags):
                            objects.append(obj)
                return objects
        return []

    def _set_last_updated_metadata(
        self, last_updated: datetime, project: str, conn=None