lineapy.db package


lineapy.db.db module

class lineapy.db.db.RelationalLineaDB(url: str)[source]
  • Note that LineaDB coordinates with asset manager and relational db.

    • The asset manager deals with binaries (e.g., cached values) The relational db deals with more structured data, such as the Nodes and edges.

  • Also, at some point we might have a “cache” such that the readers don’t have to go to the database if it’s already loaded, but that’s low priority.

artifact_in_db(node_id: LineaID, execution_id: LineaID, name: str, version: int) bool[source]

Returns true if the artifact is already in the DB.


Close the database connection.

commit() None[source]

End the transaction and commit the changes.

delete_artifact_by_name(artifact_name: str, version: Union[int, str])[source]

Deletes the most recent artifact with a certain name. If a version is not specified, it will delete the most recent version sorted by date_created

classmethod from_config(options: lineapy.utils.config.lineapy_config) lineapy.db.db.RelationalLineaDB[source]

Creates a new database.

If no url is provided, it will use the result of lineapy_config.safe_get("database_url")

classmethod from_environment(url: str) lineapy.db.db.RelationalLineaDB[source]

Creates a new database.

If no url is provided, it will use the result of lineapy_config.safe_get("database_url")

get_all_artifacts() List[lineapy.db.relational.ArtifactORM][source]

Used by the artifact store to get all the artifacts

get_artifactorm_by_name(artifact_name: str, version: Optional[int] = None) lineapy.db.relational.ArtifactORM[source]

Gets the most recent artifact with a certain name. If a version is not specified, it will return the most recent version sorted by date_created

get_artifacts_for_session(session_id: LineaID) List[lineapy.db.relational.ArtifactORM][source]

Gets a code slice for an artifact by name, assuming there is only one artifact with that name,

get_latest_artifact_version(artifact_name: str) int[source]

Get the latest version number of an artifact. If the artifact does not exist, it will return -1

get_libraries_for_session(session_id: LineaID) List[lineapy.db.relational.ImportNodeORM][source]

Gets all dependencies for a session, assuming all the libs in a particular session will be required to set up a new env.

get_node_by_id(linea_id: LineaID) Union[,,,,,,,][source]

Returns the node by looking up the database by ID SQLAlchemy is able to translate between the two types on demand

get_node_value_path(node_id: LineaID, execution_id: LineaID) Optional[str][source]

Get the path to the value of the artifact.


other – Additional argument to let you query another artifact’s value path. This is set to be optional and if its not set, we will use the current artifact

get_nodes_for_session(session_id: LineaID) List[Union[,,,,,,,]][source]

Get all the nodes associated with the session, which does NOT include things like SessionContext

get_variable_by_id(linea_id: LineaID) List[str][source]

Returns the variable names(as a list) for a node with a certain ID

get_variables_for_session(session_id: LineaID) List[Tuple[LineaID, str]][source]

Returns the variable names for a session, as (LineaID, variable_name)

node_value_in_db(node_id: LineaID, execution_id: LineaID) bool[source]

Returns true if the node value is already in the DB.

number_of_artifacts_per_node(node_id: LineaID, execution_id: LineaID) int[source]

Returns number of artifacts that refer to the same execution node.

write_source_code(source_code: None[source]

Writes a source code object to the database.

It first has to convert it to a SourceCodeORM object, which has the fields inlined instead of a union

lineapy.db.relational module

This file contains the ORM versions of the graph node in

Pydantic allows us to extract out a Dataclass like object from the ORM, but not let us directly write to the ORM.



non exhaustive list

SessionContext - ImportNode (One to Many) - HardwareSpec (Many to One)

Node - SessionContext (Many to One)

CallNode - Node (Many to Many)

class lineapy.db.relational.ArtifactDependencyORM(**kwargs)[source]
class lineapy.db.relational.ArtifactORM(**kwargs)[source]

An artifact is a named pointer to a node.

class lineapy.db.relational.BaseNodeORM(**kwargs)[source]

node.source_code has a path value if node.session.environment_type == “script” otherwise the environment type is “jupyter” and it has a jupyter execution count and session id, which is equal to the node.session


  • Because other nodes are inheriting from BaseNodeORM, finding a node based on its id is easy (something like the following):

    session.query(BaseNodeORM).filter( == linea_id)
  • Each node inheriting from BaseNodeORM must have non null values for all of lineno, col_offset, end_lineno, end_col_offset and source_code_id or nulls for all of them.

class lineapy.db.relational.CallNodeORM(**kwargs)[source]
class lineapy.db.relational.ElseNodeORM(**kwargs)[source]
class lineapy.db.relational.ExecutionORM(**kwargs)[source]

An execution represents one Python interpreter invocation of some number of nodes

class lineapy.db.relational.GlobalNodeORM(**kwargs)[source]
class lineapy.db.relational.GlobalReferenceORM(**kwargs)[source]
class lineapy.db.relational.IfNodeORM(**kwargs)[source]
class lineapy.db.relational.ImplicitDependencyORM(**kwargs)[source]
class lineapy.db.relational.ImportNodeORM(**kwargs)[source]
class lineapy.db.relational.InputParameterORM(**kwargs)[source]
class lineapy.db.relational.KeywordArgORM(**kwargs)[source]
class lineapy.db.relational.LiteralNodeORM(**kwargs)[source]
class lineapy.db.relational.LookupNodeORM(**kwargs)[source]
class lineapy.db.relational.MutateNodeORM(**kwargs)[source]
class lineapy.db.relational.NodeValueORM(**kwargs)[source]

A node value represents the value of a node during some execution.

It is uniquely identified by the node_id and execution_id.

The following invariant holds: value.node.session == value.execution.session

class lineapy.db.relational.PipelineORM(**kwargs)[source]
class lineapy.db.relational.PositionalArgORM(**kwargs)[source]
class lineapy.db.relational.SessionContextORM(**kwargs)[source]
class lineapy.db.relational.SourceCodeORM(**kwargs)[source]
class lineapy.db.relational.VariableNodeORM(**kwargs)[source]

lineapy.db.utils module

lineapy.db.utils.create_lineadb_engine(url: str) sqlalchemy.engine.base.Engine[source]

Create a SQLAlchemy engine for LineaDB. Take care of the SQLite database file name and configuration.

lineapy.db.utils.parse_artifact_version(version) Union[int, str][source]

Attempts to parse user-passed artifact version into a valid artifact version. A valid artifact version is either - postive int - string “all” or “latest”

Raises ValueError on failure to parse known types. Raises NotImplementedError on unknown types.

Module contents