Migration API Reference¶

runic.migrate is the schema migration engine. For the full workflow see Quickstart and Tutorial; the CLI is documented in CLI Reference.

runic.migrate.Runic¶

Runic is the single class a developer needs. It combines all DB-connected operations (upgrade, downgrade, stamp, current) with offline DAG queries (history, heads, revision creation) in one coherent API.

class runic.migrate.context.Runic(adapter, script_location, *, preview=False, target_manifest=None, track_checksums=True, track_installed_by=True, truncate_slug_length=40, file_template=None)[source]¶

Bases: object

The single SDK entry point for runic migrations.

Handles both DB-connected operations (upgrade, downgrade, stamp, current) and offline DAG queries (get_history, get_heads, create_revision, …).

Example:

from pathlib import Path
from runic import Runic
from runic.migrate.adapters.falkordb import FalkorDBAdapter

adapter = FalkorDBAdapter.from_url("falkor://localhost:6379", "my_graph")
runic = Runic(adapter, script_location=Path("runic/"))
runic.migrate.upgrade("head")

Parameters:

adapter (GraphAdapter)
script_location (Path)
preview (bool)
target_manifest (SchemaManifest | None)
track_checksums (bool)
track_installed_by (bool)
truncate_slug_length (int)
file_template (str | None)

validate()[source]¶

Check that applied revisions’ local files match their stored checksums.

Returns a list of error strings (empty means everything is valid). Missing checksum entries are skipped (backward compatible with databases migrated before checksum tracking was introduced). Returns [] immediately when track_checksums=False.

Return type:: list[str]

get_history(range_=None)[source]¶

Return revision history newest-first.

range_ accepts the start:end format (either side may be omitted to mean base / head respectively).

Parameters:: range_ (str | None)
Return type:: list[RevisionInfo]

get_heads()[source]¶

Return all head revisions (not referenced as any down_revision).

Return type:: list[Revision]

get_branch_points()[source]¶

Return each branch-point revision paired with its direct child revision ids.

Return type:: list[tuple[Revision, list[str]]]

create_revision(message, head=None, rev_id=None, branch_labels=None, depends_on=None)[source]¶

Create a new migration revision script and return its path.

Parameters:

message (str)
head (str | None)
rev_id (str | None)
branch_labels (list[str] | None)
depends_on (list[str] | None)

Return type:

Path

show_revision(rev)[source]¶

Return full metadata for a single revision (by id or unique prefix).

Parameters:: rev (str)
Return type:: Revision

exception runic.migrate.context.IrreversibleMigrationError[source]¶: Bases: Exception

Programmatic usage example¶

import logging
from pathlib import Path
from runic import Runic
from runic.migrate.adapters import create_adapter

log = logging.getLogger(__name__)

adapter = create_adapter(
    "falkordb",
    url="falkor://:mypassword@localhost:6379",
    graph_name="my_graph",
)

runic = Runic(adapter, script_location=Path("runic/"))

errors = runic.migrate.validate()
if errors:
    raise RuntimeError("\n".join(errors))

runic.migrate.upgrade("head", installed_by="deploy-bot")
log.info("current: %s", runic.migrate.current())

history = runic.migrate.get_history()
for entry in history:
    log.info("%s  %s", entry.revision, entry.message)

runic.migrate.downgrade("base")

runic.migrate.init¶

runic.migrate.service.init(directory, *, force=False)[source]¶

Scaffold a new runic migration environment on disk.

Raises FileExistsError when directory exists and force is False.

Parameters:

directory (Path)
force (bool)

Return type:

None

runic.migrate.context¶

The runic.migrate.context module exposes a module-level singleton API that env.py uses so the CLI can discover the configured context after executing the file. SDK users should prefer instantiating Runic directly rather than using this API.

runic.migrate.context.configure(adapter, script_location=None, preview=False, *, target_manifest=None, track_checksums=True, track_installed_by=True, truncate_slug_length=40, file_template=None, _env_path=None)[source]¶

Parameters:

adapter (GraphAdapter)
script_location (Path | None)
preview (bool)
target_manifest (SchemaManifest | None)
track_checksums (bool)
track_installed_by (bool)
truncate_slug_length (int)
file_template (str | None)
_env_path (Path | None)

Return type:

None

runic.migrate.context.get()[source]¶

Return type:: Runic

runic.migrate.context.is_preview()[source]¶

Return type:: bool

runic.migrate.adapters¶

runic.migrate.adapters.create_adapter(backend, **kwargs)[source]¶

Instantiate a named adapter from keyword arguments.

Two connection variants are supported for "falkordb":

URL variant — credentials embedded in the connection string:

create_adapter(
    "falkordb", url="falkor://:mypassword@localhost:6379", graph_name="my_graph"
)

Params variant — explicit host/port/auth kwargs:

create_adapter(
    "falkordb",
    host="localhost",
    port=6379,
    username="myuser",
    password="mypassword",
    graph_name="my_graph",
)

Parameters:

backend (str)
kwargs (Any)

Return type:

GraphAdapter

runic.migrate.operations¶

class runic.migrate.operations.GraphOperations(adapter, preview=False)[source]¶

Bases: DataOperations

Migration-script API: data manipulation + DDL, both with preview mode.

Extends DataOperations with DDL operations (indexes, constraints, snapshots) that delegate to the underlying adapter.

Migration scripts receive this object as their ops argument:

def upgrade(ops: GraphOperations) -> None:
    ops.create_range_index("Person", "email")
    ops.rename_property("Person", "fname", "first_name")

Parameters:

adapter (GraphAdapter)
preview (bool)

runic.migrate.manifest¶

Schema manifest classes used with autogenerate. See Autogenerate for usage examples.

class runic.migrate.manifest.SchemaManifest(range_indexes: 'list[RangeIndex]' = <factory>, fulltext_indexes: 'list[FulltextIndex]' = <factory>, vector_indexes: 'list[VectorIndex]' = <factory>, constraints: 'list[UniqueConstraint | MandatoryConstraint]' = <factory>)[source]

Bases: object

Parameters:

range_indexes (list[RangeIndex])
fulltext_indexes (list[FulltextIndex])
vector_indexes (list[VectorIndex])
constraints (list[UniqueConstraint | MandatoryConstraint])

class runic.migrate.manifest.RangeIndex(label: 'str', prop: 'str', rel: 'bool' = False)[source]

Bases: object

Parameters:

label (str)
prop (str)
rel (bool)

class runic.migrate.manifest.FulltextIndex(label: 'str', props: 'list[str] | tuple[str, ...]', language: 'str | None' = None, stopwords: 'list[str] | tuple[str, ...] | None' = None) → 'None'[source]

Bases: object

Parameters:

label (str)
props (tuple[str, ...])
language (str | None)
stopwords (tuple[str, ...] | None)

class runic.migrate.manifest.VectorIndex(label: 'str', prop: 'str', dimension: 'int', similarity: 'str', m: 'int' = 16, ef_construction: 'int' = 200, ef_runtime: 'int' = 10)[source]

Bases: object

Parameters:

label (str)
prop (str)
dimension (int)
similarity (str)
m (int)
ef_construction (int)
ef_runtime (int)

class runic.migrate.manifest.UniqueConstraint(entity: 'str', label: 'str', props: 'list[str] | tuple[str, ...]') → 'None'[source]

Bases: object

Parameters:

entity (str)
label (str)
props (tuple[str, ...])

class runic.migrate.manifest.MandatoryConstraint(entity: 'str', label: 'str', props: 'list[str] | tuple[str, ...]') → 'None'[source]

Bases: object

Parameters:

entity (str)
label (str)
props (tuple[str, ...])

runic.migrate.script¶

Internal revision DAG types returned by methods on Runic; rarely constructed directly.

class runic.migrate.script.Revision(revision: 'str', down_revision: 'str | tuple[str, ...] | None', branch_labels: 'list[str]', depends_on: 'list[str]', irreversible: 'bool', snapshot: 'bool', message: 'str', create_date: 'datetime', path: 'Path', module: 'Any' = None)[source]¶

Bases: object

Parameters:

revision (str)
down_revision (str | tuple[str, ...] | None)
branch_labels (list[str])
depends_on (list[str])
irreversible (bool)
snapshot (bool)
message (str)
create_date (datetime)
path (Path)
module (Any)

class runic.migrate.script.RevisionInfo(revision: 'str', down_revision: 'str | tuple[str, ...] | None', message: 'str', create_date: 'datetime', is_head: 'bool', is_branch_point: 'bool')[source]¶

Bases: object

Parameters:

revision (str)
down_revision (str | tuple[str, ...] | None)
message (str)
create_date (datetime)
is_head (bool)
is_branch_point (bool)

exception runic.migrate.script.RevisionNotFound[source]¶: Bases: Exception

exception runic.migrate.script.AmbiguousRevision[source]¶: Bases: Exception

runic.migrate.exceptions — Migration Exceptions¶

exception runic.migrate.exceptions.MultipleHeadsError[source]¶: Bases: Exception

exception runic.migrate.exceptions.MultipleBasesError[source]¶: Bases: Exception

exception runic.migrate.exceptions.ConstraintFailedError[source]¶: Bases: Exception

exception runic.migrate.exceptions.ConstraintTimeoutError[source]¶: Bases: Exception

runic.migrate.testing¶

Pytest fixtures for integration tests. Requires falkordblite.