finprint.recording

module finprint

The recording session (rec): readers, named steps, save, publish.

A session owns every step recorded on its tracked frames. Saving walks back from the output frame through the steps that produced it, so frames looked at along the way (head(), describe()) do not clutter the recipe.

Classes

RecipeError

class

class RecipeError(Exception)

Fields

Name	Type	Default	Description
code		code

Methods and properties (1)

init constructor

RecipeError(self, code: str, message: str) -> None

Source: packages/finprint/src/finprint/recording.py, line 68

TransientDataError

class

class TransientDataError(RecipeError)

Methods and properties (1)

init constructor

TransientDataError(self, message: str) -> None

Source: packages/finprint/src/finprint/recording.py, line 74

RecipeWarning

class

class RecipeWarning(UserWarning)

A recipe was saved with a note the analyst should see (partial capture, transient data).

Source: packages/finprint/src/finprint/recording.py, line 79

RecipePrivacyWarning

class

class RecipePrivacyWarning(RecipeWarning)

Cell source or step code that looks like it holds an email address or a key.

Source: packages/finprint/src/finprint/recording.py, line 83

SaveResult

class

class SaveResult

Fields

Name	Type	Default	Description
recipe	TransformRecipe
recipe_path	Path
data_path	Path | None
rows_written	bool

Methods and properties (1)

init constructor

SaveResult(
    self,
    recipe: TransformRecipe,
    recipe_path: Path,
    data_path: Path | None,
    rows_written: bool,
) -> None

Source: packages/finprint/src/finprint/recording.py, line 87

Session

class

class Session

Records reads, verbs and named steps on tracked frames. Create one with finprint.session(title).

Fields

Name	Type	Default	Description
title		title
author		author
recipe_id		'rcp_' + uuid.uuid4().hex[:16]
created_at		datetime.now(UTC).replace(microsecond=0)

Methods and properties (17)

init constructor

Session(self, title: str, author: str | None = None, *, _auto: bool = False) -> None

check_all method

def check_all(self) -> int

Look for unexplained changes on every live tracked frame; returns how many were found.

read_csv method

def read_csv(self, path: Any, **kwargs: Any) -> TrackedFrame

pandas.read_csv that records the file’s hash and fingerprint and returns a tracked frame.

read_excel method

def read_excel(self, path: Any, **kwargs: Any) -> TrackedFrame

read_parquet method

def read_parquet(self, path: Any, **kwargs: Any) -> TrackedFrame

track method

def track(self, df: pd.DataFrame, name: str = 'in-memory frame') -> TrackedFrame

Start tracking a frame built some other way. It is recorded as an in-memory input (replay cannot load it).

read_catalogue method

def read_catalogue(
    self,
    dataset_id: str,
    *,
    use: str = 'transform',
    url: str | None = None,
    principal: str | None = None,
    client: httpx.Client | None = None,
    name: str | None = None,
) -> TrackedFrame

Read a catalogued dataset through apps/data (:materialize with use, default transform).

step method

def step(self, fn: Any = None, *, name: str | None = None) -> Any

Decorator: calls with a tracked frame argument are recorded as one named, replayable step.

call method

def call(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any

Call a pandas or numpy function on tracked frames and record it (rec.call(pd.concat, [a, b])).

concat method

def concat(self, objs: Any, **kwargs: Any) -> Any

merge method

def merge(self, left: Any, right: Any, **kwargs: Any) -> Any

untrack staticmethod

def untrack(obj: Any) -> Any

transient_inputs method

def transient_inputs(self, frame: Any) -> list[RecipeInput]

build_recipe method

def build_recipe(
    self,
    frame: Any,
    output: RecipeOutput,
    *,
    cells: bool = True,
) -> TransformRecipe

save method

def save(
    self,
    frame: Any,
    path: str | os.PathLike[str],
    *,
    cells: bool = True,
) -> SaveResult

Write the data and <stem>.recipe.json beside it. Rows derived from a transient_only input are never written (TRANSIENT_DATA); the recipe still is. cells=False leaves notebook cell source out.

recipe method

def recipe(
    self,
    frame: Any,
    *,
    name: str = 'output.csv',
    cells: bool = True,
) -> TransformRecipe

The recipe for frame without writing anything (output hashed from an in-memory CSV).

publish method

def publish(
    self,
    frame: Any,
    *,
    client_id: str,
    name: str,
    url: str | None = None,
    principal: str | None = None,
    client: httpx.Client | None = None,
    cells: bool = True,
    title: str | None = None,
) -> dict[str, Any]

Upload the output and its recipe to apps/data (POST /v1/datasets:upload). title is the catalogue title (apps/data uses <client_id> <name> when it is left out).

Source: packages/finprint/src/finprint/recording.py, line 159

Functions

environment

function

def environment() -> Environment

Source: packages/finprint/src/finprint/recording.py, line 99

step_source

function

def step_source(fn: Callable[..., Any]) -> str | None

The function’s source without its decorator lines (so replay can exec it), or None.

Source: packages/finprint/src/finprint/recording.py, line 144

session

function

def session(title: str, author: str | None = None, *, _auto: bool = False) -> Session

Start a recording session and make it current. In a notebook where %load_ext finprint started an empty session, that one is renamed and returned so the cells recorded so far stay with it.

Source: packages/finprint/src/finprint/recording.py, line 823

current

function

def current() -> Session

Source: packages/finprint/src/finprint/recording.py, line 837

current_or_none

function

def current_or_none() -> Session | None

Source: packages/finprint/src/finprint/recording.py, line 844

untrack

function

def untrack(obj: Any) -> Any

The plain pandas object behind a tracked one (the same object, not a copy). Edits made to it are not recorded; the session notices them as unexplained changes and marks the recipe partial.

Source: packages/finprint/src/finprint/recording.py, line 848

step

function

def step(fn: Any = None, *, name: str | None = None) -> Any

@finprint.step: like @rec.step but bound to whichever session is current when the function runs.

Source: packages/finprint/src/finprint/recording.py, line 874

Constants

Name	Type	Value	Description
F		TypeVar('F', bound=Callable[..., Any])