db_ops

Metadata database read/write helpers for the editor UI

Small FSDB helpers used by the metadata editor: load the flattened MIAPPE metadata of every scan, and write updated metadata back to metadata.json with a .bak backup and schema validation.

A database root must be a formal PlantDB (FSDB, i.e. a directory containing a romidb marker file). A loose directory that is not a proper ROMI DB is rejected.

all_scan_metadata

all_scan_metadata(db_path)

Return flattened scan metadata for every scan in the database.

Parameters:

Name	Type	Description	Default
db_path	Path	Path to the FSDB to load.	required

Returns:

Type	Description
dict of dict	A mapping of each scan id to its flattened MIAPPE metadata (see :func:plantdb.client.metadata_app.field_spec.flatten).

Source code in plantdb/client/metadata_app/db_ops.py

def all_scan_metadata(db_path: Path) -> dict[str, dict[str, Any]]:
    """Return flattened scan metadata for every scan in the database.

    Parameters
    ----------
    db_path : Path
        Path to the FSDB to load.

    Returns
    -------
    dict of dict
        A mapping of each scan id to its flattened MIAPPE metadata
        (see :func:`plantdb.client.metadata_app.field_spec.flatten`).
    """
    flattened: dict[str, dict[str, Any]] = {}
    for scan in _connect(db_path).get_scans():
        flattened[scan.id] = flatten(scan.get_metadata())
    return flattened

apply_bulk

apply_bulk(db_path, scan_ids, path, value, backup=True)

Set the field at path to value on every scan in scan_ids.

Parameters:

Name	Type	Description	Default
db_path	Path	Path to the FSDB containing the scans.	required
scan_ids	list of str	Scan ids to update.	required
path	str	Dot-separated field path to set.	required
value	Any	Value to set at the field.	required
backup	bool	Back up each scan's metadata before overwriting.	True

Returns:

Type	Description
list of str	The scan ids that were actually modified (i.e. whose value changed). Writes are validated and backed up.

Source code in plantdb/client/metadata_app/db_ops.py

def apply_bulk(db_path: Path, scan_ids: list[str], path: str, value: Any,
               backup: bool = True) -> list[str]:
    """Set the field at ``path`` to ``value`` on every scan in ``scan_ids``.

    Parameters
    ----------
    db_path : Path
        Path to the FSDB containing the scans.
    scan_ids : list of str
        Scan ids to update.
    path : str
        Dot-separated field path to set.
    value : Any
        Value to set at the field.
    backup : bool, default True
        Back up each scan's metadata before overwriting.

    Returns
    -------
    list of str
        The scan ids that were actually modified (i.e. whose value changed).
        Writes are validated and backed up.
    """
    modified: list[str] = []
    db = _connect(db_path)
    for scan_id in scan_ids:
        scan = db.get_scan(scan_id)
        metadata = scan.get_metadata()
        if get_field(metadata, path) == value:
            continue
        set_field(metadata, path, value)
        write_scan_metadata(scan, metadata, backup=backup)
        modified.append(scan_id)
    return modified

close_db

close_db(db_path)

Disconnect and drop the cached FSDB for db_path, if any.

Used when the app switches to a different database so only one live connection stays in memory.

Parameters:

Name	Type	Description	Default
db_path	Path	Path of the database whose cached connection to close.	required

Source code in plantdb/client/metadata_app/db_ops.py

def close_db(db_path: Path) -> None:
    """Disconnect and drop the cached FSDB for ``db_path``, if any.

    Used when the app switches to a different database so only one live
    connection stays in memory.

    Parameters
    ----------
    db_path : Path
        Path of the database whose cached connection to close.
    """
    db_path = Path(db_path).resolve()
    with _DB_LOCK:
        db = _DB_CACHE.pop(db_path, None)
    if db is not None:
        db.disconnect()

get_field

get_field(metadata, path)

Return the value of the field at dot-path in a nested dict.

Parameters:

Name	Type	Description	Default
metadata	dict of str to Any	The nested metadata dict to read.	required
path	str	Dot-separated path to the field, e.g. study.title.	required

Returns:

Type	Description
Any	The field value, or None if the path does not exist.

Source code in plantdb/client/metadata_app/db_ops.py

def get_field(metadata: dict[str, Any], path: str) -> Any:
    """Return the value of the field at dot-``path`` in a nested dict.

    Parameters
    ----------
    metadata : dict of str to Any
        The nested metadata dict to read.
    path : str
        Dot-separated path to the field, e.g. ``study.title``.

    Returns
    -------
    Any
        The field value, or ``None`` if the path does not exist.
    """
    node = metadata
    for part in path.split("."):
        if not isinstance(node, dict) or part not in node:
            return None
        node = node[part]
    return node

get_scan_dir

get_scan_dir(db_path, scan_id)

Return the on-disk directory of scan_id in the FSDB at db_path.

Parameters:

Name	Type	Description	Default
db_path	Path	Path to the FSDB containing the scan.	required
scan_id	str	Identifier of the scan.	required

Returns:

Type	Description
Path	Directory of the scan.

Source code in plantdb/client/metadata_app/db_ops.py

def get_scan_dir(db_path: Path, scan_id: str) -> Path:
    """Return the on-disk directory of ``scan_id`` in the FSDB at ``db_path``.

    Parameters
    ----------
    db_path : Path
        Path to the FSDB containing the scan.
    scan_id : str
        Identifier of the scan.

    Returns
    -------
    Path
        Directory of the scan.
    """
    return _connect(db_path).get_scan(scan_id, owner_only=False).path()

migratable_scans

migratable_scans(db_path)

Return the ids of scans that still use the legacy (pre-MIAPPE) schema.

Parameters:

Name	Type	Description	Default
db_path	Path	Path to the FSDB to inspect.	required

Returns:

Type	Description
list of str	Scan ids that require migration.

Source code in plantdb/client/metadata_app/db_ops.py

def migratable_scans(db_path: Path) -> list[str]:
    """Return the ids of scans that still use the legacy (pre-MIAPPE) schema.

    Parameters
    ----------
    db_path : Path
        Path to the FSDB to inspect.

    Returns
    -------
    list of str
        Scan ids that require migration.
    """
    return [sid for sid in _scan_ids(db_path) if scan_needs_migration(get_scan_dir(db_path, sid))]

migrate_scans_progress

migrate_scans_progress(db_path, scan_ids, logger, on_progress=None)

Migrate scan_ids scan by scan, reporting progress.

on_progress(done, total) is called after each scan is processed, so a caller can surface a live progress bar.

Parameters:

Name	Type	Description	Default
db_path	Path	Path to the FSDB containing the scans.	required
scan_ids	list of str	Scan ids to migrate.	required
logger	Logger	Logger used to report each migrated scan.	required
on_progress	Callable[[int, int], None]	Callback invoked with (done, total) after each scan.	None

Returns:

Type	Description
int	The number of scans that were actually migrated.

Source code in plantdb/client/metadata_app/db_ops.py

def migrate_scans_progress(db_path: Path, scan_ids: list[str], logger: logging.Logger,
                           on_progress: Callable[[int, int], None] | None = None) -> int:
    """Migrate ``scan_ids`` scan by scan, reporting progress.

    ``on_progress(done, total)`` is called after each scan is processed, so a
    caller can surface a live progress bar.

    Parameters
    ----------
    db_path : Path
        Path to the FSDB containing the scans.
    scan_ids : list of str
        Scan ids to migrate.
    logger : logging.Logger
        Logger used to report each migrated scan.
    on_progress : Callable[[int, int], None], optional
        Callback invoked with ``(done, total)`` after each scan.

    Returns
    -------
    int
        The number of scans that were actually migrated.
    """
    total = len(scan_ids)
    done = 0
    for sid in scan_ids:
        logger.info(f"Migrating {sid} ({done+1}/{total}): {get_scan_dir(db_path, sid)}")
        if migrate_scan_metadata(get_scan_dir(db_path, sid)):
            done += 1
        if on_progress is not None:
            on_progress(done, total)
    return done

scan_needs_migration

scan_needs_migration(scan)

Return True if a scan's metadata still holds a legacy object block.

Parameters:

Name	Type	Description	Default
scan	Scan	Scan to inspect.	required

Returns:

Type	Description
bool	True if the scan requires migration to the MIAPPE schema.

Source code in plantdb/client/metadata_app/db_ops.py

def scan_needs_migration(scan: "Scan") -> bool:
    """Return ``True`` if a scan's metadata still holds a legacy ``object`` block.

    Parameters
    ----------
    scan : plantdb.commons.fsdb.core.Scan
        Scan to inspect.

    Returns
    -------
    bool
        ``True`` if the scan requires migration to the MIAPPE schema.
    """
    return migrate_metadata(_load_scan_metadata(scan))[1]

set_field

set_field(metadata, path, value)

Return metadata with the field at dot-path set to value.

Intermediate sections are created as needed.

Parameters:

Name	Type	Description	Default
metadata	dict of str to Any	The nested metadata dict to modify (in place).	required
path	str	Dot-separated path to the field, e.g. study.title.	required
value	Any	Value to set at the field.	required

Returns:

Type	Description
dict of str to Any	The same metadata dict, modified in place.

Source code in plantdb/client/metadata_app/db_ops.py

def set_field(metadata: dict[str, Any], path: str, value: Any) -> dict[str, Any]:
    """Return ``metadata`` with the field at dot-``path`` set to ``value``.

    Intermediate sections are created as needed.

    Parameters
    ----------
    metadata : dict of str to Any
        The nested metadata dict to modify (in place).
    path : str
        Dot-separated path to the field, e.g. ``study.title``.
    value : Any
        Value to set at the field.

    Returns
    -------
    dict of str to Any
        The same ``metadata`` dict, modified in place.
    """
    parts = path.split(".")
    node = metadata
    for part in parts[:-1]:
        if not isinstance(node.get(part), dict):
            node[part] = {}
        node = node[part]
    node[parts[-1]] = value
    return metadata

update_biological

update_biological(metadata, tree)

Return metadata with the biological sections replaced by tree.

Non-biological keys (owner, created, ...) are preserved.

Parameters:

Name	Type	Description	Default
metadata	dict of str to Any	The full scan metadata to update.	required
tree	dict of str to Any	New MIAPPE biological tree to install.	required

Returns:

Type	Description
dict of str to Any	A copy of metadata with the biological sections replaced.

Source code in plantdb/client/metadata_app/db_ops.py

def update_biological(metadata: dict[str, Any], tree: dict[str, Any]) -> dict[str, Any]:
    """Return ``metadata`` with the biological sections replaced by ``tree``.

    Non-biological keys (owner, created, ...) are preserved.

    Parameters
    ----------
    metadata : dict of str to Any
        The full scan metadata to update.
    tree : dict of str to Any
        New MIAPPE biological tree to install.

    Returns
    -------
    dict of str to Any
        A copy of ``metadata`` with the biological sections replaced.
    """
    updated = dict(metadata)
    for section in _BIOLOGICAL_SECTIONS:
        updated.pop(section, None)
    updated.update(tree)
    return updated

write_scan_metadata

write_scan_metadata(scan, metadata, backup=True)

Validate and write metadata to scan_dir/metadata/metadata.json.

A .bak copy of the previous file is written first when backup is True.

Parameters:

Name	Type	Description	Default
scan	Scan	Scan instance to use for metadata modification.	required
metadata	dict of str to Any	The metadata to persist.	required
backup	bool	Write a .bak copy of the previous file before overwriting.	True

Raises:

Type	Description
ValueError	If the MIAPPE biological block of metadata is invalid.

Source code in plantdb/client/metadata_app/db_ops.py

def write_scan_metadata(scan: "Scan", metadata: dict[str, Any], backup: bool = True) -> None:
    """Validate and write ``metadata`` to ``scan_dir/metadata/metadata.json``.

    A ``.bak`` copy of the previous file is written first when ``backup`` is ``True``.

    Parameters
    ----------
    scan : plantdb.commons.fsdb.core.Scan
        Scan instance to use for metadata modification.
    metadata : dict of str to Any
        The metadata to persist.
    backup : bool, default True
        Write a ``.bak`` copy of the previous file before overwriting.

    Raises
    ------
    ValueError
        If the MIAPPE biological block of ``metadata`` is invalid.
    """
    validate_biological_metadata(metadata)
    md_path = _scan_metadata_path(scan)
    if backup and md_path.is_file():
        shutil.copy2(md_path, md_path.with_suffix(md_path.suffix + ".bak"))
    scan.set_metadata(metadata)