linea_artifact

LineaArtifact dataclass

LineaArtifact Exposes functionalities we offer around the artifact.

Source code in lineapy/api/models/linea_artifact.py

@dataclass
class LineaArtifact:
    """LineaArtifact
    Exposes functionalities we offer around the artifact.
    """

    db: RelationalLineaDB = field(repr=False)
    _execution_id: LineaID = field(repr=False)
    _node_id: LineaID = field(repr=False)
    """node id of the artifact in the graph"""
    _session_id: LineaID = field(repr=False)
    """session id of the session that created the artifact"""
    name: str
    """name of the artifact"""
    _version: int
    """version of the artifact - currently start from 0"""
    _artifact_id: Optional[int] = field(default=None, repr=False)
    date_created: Optional[datetime] = field(default=None, repr=False)
    # setting repr to false for date_created, _artifact_id for now since it duplicates version
    """Optional because `date_created` and `_artifact_id` cannot be set by the user. 
    it is supposed to be automatically set when the artifact gets saved to the 
    db. so when creating lineaArtifact the first time, it will be unset. When 
    you get the artifact or list of artifacts as an artifact store, we retrieve 
    the date from db directly"""

    def __post_init__(self) -> None:
        """
        Fill empty _artifact_id and date_created attributes.
        """
        if self._artifact_id is None or self.date_created is None:
            artifactorm = self.db.get_artifactorm_by_name(
                artifact_name=self.name, version=self.version
            )
            self._artifact_id = artifactorm.id
            self.date_created = artifactorm.date_created

    @property
    def version(self) -> int:
        track(GetVersionEvent(""))
        return self._version

    @property
    def node_id(self) -> LineaID:
        return self._node_id

    @property
    def session_id(self) -> LineaID:
        return self._session_id

    @lru_cache(maxsize=None)
    def get_value(self) -> object:
        """
        Get and return the value of the artifact.
        """
        metadata = self.get_metadata()
        linea_metadata = metadata["lineapy"]
        saved_filepath = linea_metadata.storage_path
        if saved_filepath is None:
            return None
        else:
            track(GetValueEvent(has_value=True))
            # read from mlflow
            if "mlflow" in metadata.keys():
                return read_mlflow(metadata["mlflow"])

            # read from lineapy
            return read_pickle(saved_filepath)

    @lru_cache(maxsize=None)
    def _get_storage_path(self) -> Optional[str]:
        return self.db.get_node_value_path(self._node_id, self._execution_id)

    def _get_storage_backend(self, storage_path) -> ARTIFACT_STORAGE_BACKEND:
        """
        Get storage backend based on the storage path.
        """
        if isinstance(storage_path, str) and storage_path.startswith("runs:"):
            # MLflow log_model should return the model URI with prefix `runs:`
            return ARTIFACT_STORAGE_BACKEND.mlflow
        else:
            return ARTIFACT_STORAGE_BACKEND.lineapy

    @lru_cache(maxsize=None)
    def get_metadata(self, lineapy_only: bool = False) -> ArtifactInfo:
        """
        Get artifact backend storage metadata.

        Parameters
        ----------
        lineapy_only: bool
            If `False`, will include both LineaPy related
            metadata and metadata from storage backend(if it is not LineaPy).
            If `True`, will only return LineaPy related metadata no matter
            which storage backend is using.

        Returns
        -------
            Metadata for artifact backend storage.
        """

        assert isinstance(self._artifact_id, int)
        assert isinstance(self.date_created, datetime)

        storage_path = self._get_storage_path()
        storage_backend = self._get_storage_backend(storage_path)

        lineaartifact_metadata = LineaArtifactInfo(
            artifact_id=self._artifact_id,
            name=self.name,
            version=self.version,
            execution_id=self._execution_id,
            session_id=self._session_id,
            node_id=self._node_id,
            date_created=self.date_created,
            storage_path=storage_path,
            storage_backend=storage_backend,
        )

        metadata = ArtifactInfo(lineapy=lineaartifact_metadata)

        if not lineapy_only and storage_backend == "mlflow":
            mlflowartifactorm = (
                self.db.get_mlflowartifactmetadataorm_by_artifact_id(
                    self._artifact_id
                )
            )
            assert isinstance(mlflowartifactorm.id, int)
            assert isinstance(mlflowartifactorm.artifact_id, int)
            assert isinstance(mlflowartifactorm.tracking_uri, str)
            assert isinstance(mlflowartifactorm.model_uri, str)
            assert isinstance(mlflowartifactorm.model_flavor, str)
            metadata["mlflow"] = MLflowArtifactInfo(
                id=mlflowartifactorm.id,
                artifact_id=mlflowartifactorm.artifact_id,
                tracking_uri=mlflowartifactorm.tracking_uri,
                registry_uri=mlflowartifactorm.registry_uri,
                model_uri=mlflowartifactorm.model_uri,
                model_flavor=mlflowartifactorm.model_flavor,
            )

        return metadata

    # Note that I removed the @properties because they were not working
    # well with the lru_cache
    @lru_cache(maxsize=None)
    def _get_subgraph(self, keep_lineapy_save: bool = False) -> Graph:
        """
        Return the slice subgraph for the artifact.

        Parameters
        ----------
        keep_lineapy_save: bool
            Whether to retain `lineapy.save()` in code slice.
            Defaults to `False`.
        """
        session_graph = Graph.create_session_graph(self.db, self._session_id)
        return get_slice_graph(
            session_graph, [self._node_id], keep_lineapy_save
        )

    @lru_cache(maxsize=None)
    def _get_sessiongraph_and_subgraph_nodelist(
        self, keep_lineapy_save: bool = False
    ) -> Tuple[Graph, Set[LineaID]]:
        session_graph = Graph.create_session_graph(self.db, self._session_id)
        return session_graph, get_subgraph_nodelist(
            session_graph, [self._node_id], keep_lineapy_save
        )

    @lru_cache(maxsize=None)
    def get_code(
        self,
        use_lineapy_serialization: bool = True,
        keep_lineapy_save: bool = False,
        include_non_slice_as_comment: bool = False,
    ) -> str:
        """
        Return the slices code for the artifact.

        Parameters
        ----------
        use_lineapy_serialization: bool
            If `True`, will use the lineapy serialization to get the code.
            We will hide the serialization and the value pickler irrespective of the value type.
            If `False`, will use remove all the lineapy references and instead use the underlying serializer directly.
            Currently, we use the native `pickle` serializer.
        keep_lineapy_save: bool
            Whether to retain `lineapy.save()` in code slice.
            Defaults to `False`.
        """
        # FIXME: this seems a little heavy to just get the slice?
        track(
            GetCodeEvent(
                use_lineapy_serialization=use_lineapy_serialization,
                is_session_code=False,
            )
        )
        (
            sessiongraph,
            subgraph_nodelist,
        ) = self._get_sessiongraph_and_subgraph_nodelist(keep_lineapy_save)
        code = str(
            get_source_code_from_graph(
                subgraph_nodelist,
                session_graph=sessiongraph,
                include_non_slice_as_comment=include_non_slice_as_comment,
            )
        )
        if not use_lineapy_serialization:
            code = de_lineate_code(code, self.db)
        return prettify(code)

    @lru_cache(maxsize=None)
    def get_session_code(self, use_lineapy_serialization: bool = True) -> str:
        """
        Return the raw session code for the artifact. This will include any
        comments and non-code lines.

        Parameters
        ----------
        use_lineapy_serialization: bool
            If `True`, will use the lineapy serialization to get the code.
            We will hide the serialization and the value pickler irrespective of the value type.
            If `False`, will use remove all the lineapy references and instead use the underlying serializer directly.
            Currently, we use the native `pickle` serializer.
        """
        # using this over get_source_code_from_graph because it will process the
        # graph code and not return the original code with comments etc.
        track(
            GetCodeEvent(
                use_lineapy_serialization=use_lineapy_serialization,
                is_session_code=True,
            )
        )
        code = self.db.get_source_code_for_session(self._session_id)
        if not use_lineapy_serialization:
            code = de_lineate_code(code, self.db)
        # NOTE: we are not prettifying this code because we want to preserve what
        # the user wrote originally, without processing
        return code

    def visualize(self, path: Optional[str] = None) -> None:
        """
        Displays the graph for this artifact.
        If a path is provided, will save it to that file instead.
        """
        # adding this inside function to lazy import graphviz.
        # This way we can import lineapy without having graphviz installed.
        from lineapy.visualizer import Visualizer

        session_graph = Graph.create_session_graph(self.db, self._session_id)
        visualizer = Visualizer.for_public_node(session_graph, self._node_id)
        if path:
            visualizer.render_pdf_file(path)
        else:
            display(visualizer.ipython_display_object())

    def execute(self) -> object:
        """
        Executes the artifact graph.
        """
        slice_exec = Executor(self.db, globals())
        slice_exec.execute_graph(self._get_subgraph())
        return slice_exec.get_value(self._node_id)

    @staticmethod
    def get_artifact_from_orm(
        db: RelationalLineaDB, artifactorm: ArtifactORM
    ) -> "LineaArtifact":
        """
        Return LineaArtifact from artifactorm.
        """
        assert isinstance(artifactorm.name, str)
        return LineaArtifact(
            db=db,
            _artifact_id=artifactorm.id,
            _execution_id=artifactorm.execution_id,
            _node_id=artifactorm.node_id,
            _session_id=artifactorm.node.session_id,
            _version=artifactorm.version,  # type: ignore
            name=artifactorm.name,
            date_created=artifactorm.date_created,  # type: ignore
        )

    @staticmethod
    def get_artifact_from_name_and_version(
        db: RelationalLineaDB,
        artifact_name: str,
        version: Optional[int] = None,
    ) -> "LineaArtifact":
        """
        Return LineaArtifact from artifact name and version.
        """
        return LineaArtifact.get_artifact_from_orm(
            db, db.get_artifactorm_by_name(artifact_name, version)
        )

    @staticmethod
    def get_artifact_from_def(
        db: RelationalLineaDB, artifactdef: LineaArtifactDef
    ) -> "LineaArtifact":
        """
        Return LineaArtifact from `LineaArtifactDef`.
        """
        return LineaArtifact.get_artifact_from_name_and_version(
            db, **artifactdef
        )

date_created: Optional[datetime] = field(default=None, repr=False) class-attribute

Optional because date_created and _artifact_id cannot be set by the user. it is supposed to be automatically set when the artifact gets saved to the db. so when creating lineaArtifact the first time, it will be unset. When you get the artifact or list of artifacts as an artifact store, we retrieve the date from db directly

name: str class-attribute

name of the artifact

__post_init__()

Fill empty _artifact_id and date_created attributes.

Source code in lineapy/api/models/linea_artifact.py

def __post_init__(self) -> None:
    """
    Fill empty _artifact_id and date_created attributes.
    """
    if self._artifact_id is None or self.date_created is None:
        artifactorm = self.db.get_artifactorm_by_name(
            artifact_name=self.name, version=self.version
        )
        self._artifact_id = artifactorm.id
        self.date_created = artifactorm.date_created

execute()

Executes the artifact graph.

Source code in lineapy/api/models/linea_artifact.py

def execute(self) -> object:
    """
    Executes the artifact graph.
    """
    slice_exec = Executor(self.db, globals())
    slice_exec.execute_graph(self._get_subgraph())
    return slice_exec.get_value(self._node_id)

get_artifact_from_def(db, artifactdef) staticmethod

Return LineaArtifact from LineaArtifactDef.

Source code in lineapy/api/models/linea_artifact.py

@staticmethod
def get_artifact_from_def(
    db: RelationalLineaDB, artifactdef: LineaArtifactDef
) -> "LineaArtifact":
    """
    Return LineaArtifact from `LineaArtifactDef`.
    """
    return LineaArtifact.get_artifact_from_name_and_version(
        db, **artifactdef
    )

get_artifact_from_name_and_version(db, artifact_name, version=None) staticmethod

Return LineaArtifact from artifact name and version.

Source code in lineapy/api/models/linea_artifact.py

@staticmethod
def get_artifact_from_name_and_version(
    db: RelationalLineaDB,
    artifact_name: str,
    version: Optional[int] = None,
) -> "LineaArtifact":
    """
    Return LineaArtifact from artifact name and version.
    """
    return LineaArtifact.get_artifact_from_orm(
        db, db.get_artifactorm_by_name(artifact_name, version)
    )

get_artifact_from_orm(db, artifactorm) staticmethod

Return LineaArtifact from artifactorm.

Source code in lineapy/api/models/linea_artifact.py

@staticmethod
def get_artifact_from_orm(
    db: RelationalLineaDB, artifactorm: ArtifactORM
) -> "LineaArtifact":
    """
    Return LineaArtifact from artifactorm.
    """
    assert isinstance(artifactorm.name, str)
    return LineaArtifact(
        db=db,
        _artifact_id=artifactorm.id,
        _execution_id=artifactorm.execution_id,
        _node_id=artifactorm.node_id,
        _session_id=artifactorm.node.session_id,
        _version=artifactorm.version,  # type: ignore
        name=artifactorm.name,
        date_created=artifactorm.date_created,  # type: ignore
    )

get_code(use_lineapy_serialization=True, keep_lineapy_save=False, include_non_slice_as_comment=False)

Return the slices code for the artifact.

Parameters:

Name	Type	Description	Default
use_lineapy_serialization	bool	If True, will use the lineapy serialization to get the code. We will hide the serialization and the value pickler irrespective of the value type. If False, will use remove all the lineapy references and instead use the underlying serializer directly. Currently, we use the native pickle serializer.	True
keep_lineapy_save	bool	Whether to retain lineapy.save() in code slice. Defaults to False.	False

Source code in lineapy/api/models/linea_artifact.py

@lru_cache(maxsize=None)
def get_code(
    self,
    use_lineapy_serialization: bool = True,
    keep_lineapy_save: bool = False,
    include_non_slice_as_comment: bool = False,
) -> str:
    """
    Return the slices code for the artifact.

    Parameters
    ----------
    use_lineapy_serialization: bool
        If `True`, will use the lineapy serialization to get the code.
        We will hide the serialization and the value pickler irrespective of the value type.
        If `False`, will use remove all the lineapy references and instead use the underlying serializer directly.
        Currently, we use the native `pickle` serializer.
    keep_lineapy_save: bool
        Whether to retain `lineapy.save()` in code slice.
        Defaults to `False`.
    """
    # FIXME: this seems a little heavy to just get the slice?
    track(
        GetCodeEvent(
            use_lineapy_serialization=use_lineapy_serialization,
            is_session_code=False,
        )
    )
    (
        sessiongraph,
        subgraph_nodelist,
    ) = self._get_sessiongraph_and_subgraph_nodelist(keep_lineapy_save)
    code = str(
        get_source_code_from_graph(
            subgraph_nodelist,
            session_graph=sessiongraph,
            include_non_slice_as_comment=include_non_slice_as_comment,
        )
    )
    if not use_lineapy_serialization:
        code = de_lineate_code(code, self.db)
    return prettify(code)

get_metadata(lineapy_only=False)

Get artifact backend storage metadata.

Parameters:

Name	Type	Description	Default
lineapy_only	bool	If False, will include both LineaPy related metadata and metadata from storage backend(if it is not LineaPy). If True, will only return LineaPy related metadata no matter which storage backend is using.	False

Returns:

Type	Description
Metadata for artifact backend storage.

Source code in lineapy/api/models/linea_artifact.py

@lru_cache(maxsize=None)
def get_metadata(self, lineapy_only: bool = False) -> ArtifactInfo:
    """
    Get artifact backend storage metadata.

    Parameters
    ----------
    lineapy_only: bool
        If `False`, will include both LineaPy related
        metadata and metadata from storage backend(if it is not LineaPy).
        If `True`, will only return LineaPy related metadata no matter
        which storage backend is using.

    Returns
    -------
        Metadata for artifact backend storage.
    """

    assert isinstance(self._artifact_id, int)
    assert isinstance(self.date_created, datetime)

    storage_path = self._get_storage_path()
    storage_backend = self._get_storage_backend(storage_path)

    lineaartifact_metadata = LineaArtifactInfo(
        artifact_id=self._artifact_id,
        name=self.name,
        version=self.version,
        execution_id=self._execution_id,
        session_id=self._session_id,
        node_id=self._node_id,
        date_created=self.date_created,
        storage_path=storage_path,
        storage_backend=storage_backend,
    )

    metadata = ArtifactInfo(lineapy=lineaartifact_metadata)

    if not lineapy_only and storage_backend == "mlflow":
        mlflowartifactorm = (
            self.db.get_mlflowartifactmetadataorm_by_artifact_id(
                self._artifact_id
            )
        )
        assert isinstance(mlflowartifactorm.id, int)
        assert isinstance(mlflowartifactorm.artifact_id, int)
        assert isinstance(mlflowartifactorm.tracking_uri, str)
        assert isinstance(mlflowartifactorm.model_uri, str)
        assert isinstance(mlflowartifactorm.model_flavor, str)
        metadata["mlflow"] = MLflowArtifactInfo(
            id=mlflowartifactorm.id,
            artifact_id=mlflowartifactorm.artifact_id,
            tracking_uri=mlflowartifactorm.tracking_uri,
            registry_uri=mlflowartifactorm.registry_uri,
            model_uri=mlflowartifactorm.model_uri,
            model_flavor=mlflowartifactorm.model_flavor,
        )

    return metadata

get_session_code(use_lineapy_serialization=True)

Return the raw session code for the artifact. This will include any comments and non-code lines.

Parameters:

Name	Type	Description	Default
use_lineapy_serialization	bool	If True, will use the lineapy serialization to get the code. We will hide the serialization and the value pickler irrespective of the value type. If False, will use remove all the lineapy references and instead use the underlying serializer directly. Currently, we use the native pickle serializer.	True

Source code in lineapy/api/models/linea_artifact.py

@lru_cache(maxsize=None)
def get_session_code(self, use_lineapy_serialization: bool = True) -> str:
    """
    Return the raw session code for the artifact. This will include any
    comments and non-code lines.

    Parameters
    ----------
    use_lineapy_serialization: bool
        If `True`, will use the lineapy serialization to get the code.
        We will hide the serialization and the value pickler irrespective of the value type.
        If `False`, will use remove all the lineapy references and instead use the underlying serializer directly.
        Currently, we use the native `pickle` serializer.
    """
    # using this over get_source_code_from_graph because it will process the
    # graph code and not return the original code with comments etc.
    track(
        GetCodeEvent(
            use_lineapy_serialization=use_lineapy_serialization,
            is_session_code=True,
        )
    )
    code = self.db.get_source_code_for_session(self._session_id)
    if not use_lineapy_serialization:
        code = de_lineate_code(code, self.db)
    # NOTE: we are not prettifying this code because we want to preserve what
    # the user wrote originally, without processing
    return code

get_value()

Get and return the value of the artifact.

Source code in lineapy/api/models/linea_artifact.py

@lru_cache(maxsize=None)
def get_value(self) -> object:
    """
    Get and return the value of the artifact.
    """
    metadata = self.get_metadata()
    linea_metadata = metadata["lineapy"]
    saved_filepath = linea_metadata.storage_path
    if saved_filepath is None:
        return None
    else:
        track(GetValueEvent(has_value=True))
        # read from mlflow
        if "mlflow" in metadata.keys():
            return read_mlflow(metadata["mlflow"])

        # read from lineapy
        return read_pickle(saved_filepath)

visualize(path=None)

Displays the graph for this artifact. If a path is provided, will save it to that file instead.

Source code in lineapy/api/models/linea_artifact.py

def visualize(self, path: Optional[str] = None) -> None:
    """
    Displays the graph for this artifact.
    If a path is provided, will save it to that file instead.
    """
    # adding this inside function to lazy import graphviz.
    # This way we can import lineapy without having graphviz installed.
    from lineapy.visualizer import Visualizer

    session_graph = Graph.create_session_graph(self.db, self._session_id)
    visualizer = Visualizer.for_public_node(session_graph, self._node_id)
    if path:
        visualizer.render_pdf_file(path)
    else:
        display(visualizer.ipython_display_object())

get_lineaartifactdef(art_entry)

Convert artifact entry (string) or (string, integer) to LineaArtifactDef.

Source code in lineapy/api/models/linea_artifact.py

def get_lineaartifactdef(
    art_entry: Union[str, Tuple[str, Optional[int]]]
) -> LineaArtifactDef:
    """
    Convert artifact entry (string) or (string, integer) to `LineaArtifactDef`.
    """
    args: LineaArtifactDef
    if isinstance(art_entry, str):
        args = {"artifact_name": art_entry}
    elif isinstance(art_entry, tuple):
        args = {"artifact_name": art_entry[0], "version": art_entry[1]}
    else:
        raise ValueError(
            "An artifact should be passed in as a string or (string, integer) tuple."
        )
    return args

---

Was this helpful?

Help us improve docs with your feedback!

Send