common_nwbfile.py

Nwbfile

Bases: SpyglassMixin, Manual

Source code in src/spyglass/common/common_nwbfile.py

@schema
class Nwbfile(SpyglassMixin, dj.Manual):
    definition = """
    # Table for holding the NWB files.
    nwb_file_name: varchar(64)   # name of the NWB file
    ---
    nwb_file_abs_path: filepath@raw
    INDEX (nwb_file_abs_path)
    """
    # NOTE the INDEX above is implicit from filepath@... above but needs to be
    # explicit so that alter() can work

    # NOTE: See #630, #664. Excessive key length.

    @classmethod
    def insert_from_relative_file_name(cls, nwb_file_name: str) -> None:
        """Insert a new session from an existing NWB file.

        Parameters
        ----------
        nwb_file_name : str
            The relative path to the NWB file.
        """
        nwb_file_abs_path = Nwbfile.get_abs_path(nwb_file_name, new_file=True)

        if not Path(nwb_file_abs_path).exists():
            raise FileNotFoundError(f"File not found: {nwb_file_abs_path}")

        cls.insert1(
            dict(
                nwb_file_name=nwb_file_name, nwb_file_abs_path=nwb_file_abs_path
            ),
            skip_duplicates=True,
        )

    def fetch_nwb(self):
        return [
            get_nwb_file(self.get_abs_path(file))
            for file in self.fetch("nwb_file_name")
        ]

    @classmethod
    def get_abs_path(
        cls, nwb_file_name: str, new_file: bool = False, **kwargs
    ) -> str:
        """Return absolute path for a stored raw NWB file given file name.

        The SPYGLASS_BASE_DIR must be set, either as an environment or part of
        dj.config['custom']. See spyglass.settings.load_config

        Parameters
        ----------
        nwb_file_name : str
            The name of an NWB file that has been inserted into the Nwbfile()
            table. May be file substring. May include % wildcard(s).
        new_file : bool, optional
            Adding a new file to Nwbfile table. Defaults to False.

        Returns
        -------
        nwb_file_abspath : str
            The absolute path for the given file name.
        """
        file_path = raw_dir + "/" + nwb_file_name
        if new_file:
            return file_path

        query = cls & {"nwb_file_name": nwb_file_name}
        if len(query) != 1:
            raise ValueError(
                f"Could not find 1 entry for {nwb_file_name}:\n{query}"
            )

        return file_path

    @staticmethod
    def add_to_lock(nwb_file_name: str) -> None:
        """Add the specified NWB file to the list of locked items.

        The NWB_LOCK_FILE environment variable must be set to the path of the
        lock file, listing locked NWB files.

        Parameters
        ----------
        nwb_file_name : str
            The name of an NWB file in the Nwbfile table.
        """
        if not (Nwbfile() & {"nwb_file_name": nwb_file_name}):
            raise FileNotFoundError(
                f"File not found in Nwbfile table. Cannot lock {nwb_file_name}"
            )

        with open(os.getenv("NWB_LOCK_FILE"), "a+") as lock_file:
            lock_file.write(f"{nwb_file_name}\n")

    @staticmethod
    def cleanup(delete_files: bool = False) -> None:
        """Remove the filepath entries for NWB files that are not in use.

        This does not delete the files themselves unless delete_files=True is
        specified. Run this after deleting the Nwbfile() entries themselves.
        """
        schema.external["raw"].delete(delete_external_files=delete_files)

insert_from_relative_file_name(nwb_file_name) classmethod

Insert a new session from an existing NWB file.

Parameters:

Name	Type	Description	Default
nwb_file_name	str	The relative path to the NWB file.	required

Source code in src/spyglass/common/common_nwbfile.py

@classmethod
def insert_from_relative_file_name(cls, nwb_file_name: str) -> None:
    """Insert a new session from an existing NWB file.

    Parameters
    ----------
    nwb_file_name : str
        The relative path to the NWB file.
    """
    nwb_file_abs_path = Nwbfile.get_abs_path(nwb_file_name, new_file=True)

    if not Path(nwb_file_abs_path).exists():
        raise FileNotFoundError(f"File not found: {nwb_file_abs_path}")

    cls.insert1(
        dict(
            nwb_file_name=nwb_file_name, nwb_file_abs_path=nwb_file_abs_path
        ),
        skip_duplicates=True,
    )

get_abs_path(nwb_file_name, new_file=False, **kwargs) classmethod

Return absolute path for a stored raw NWB file given file name.

The SPYGLASS_BASE_DIR must be set, either as an environment or part of dj.config['custom']. See spyglass.settings.load_config

Parameters:

Name	Type	Description	Default
nwb_file_name	str	The name of an NWB file that has been inserted into the Nwbfile() table. May be file substring. May include % wildcard(s).	required
new_file	bool	Adding a new file to Nwbfile table. Defaults to False.	False

Returns:

Name	Type	Description
nwb_file_abspath	str	The absolute path for the given file name.

Source code in src/spyglass/common/common_nwbfile.py

@classmethod
def get_abs_path(
    cls, nwb_file_name: str, new_file: bool = False, **kwargs
) -> str:
    """Return absolute path for a stored raw NWB file given file name.

    The SPYGLASS_BASE_DIR must be set, either as an environment or part of
    dj.config['custom']. See spyglass.settings.load_config

    Parameters
    ----------
    nwb_file_name : str
        The name of an NWB file that has been inserted into the Nwbfile()
        table. May be file substring. May include % wildcard(s).
    new_file : bool, optional
        Adding a new file to Nwbfile table. Defaults to False.

    Returns
    -------
    nwb_file_abspath : str
        The absolute path for the given file name.
    """
    file_path = raw_dir + "/" + nwb_file_name
    if new_file:
        return file_path

    query = cls & {"nwb_file_name": nwb_file_name}
    if len(query) != 1:
        raise ValueError(
            f"Could not find 1 entry for {nwb_file_name}:\n{query}"
        )

    return file_path

add_to_lock(nwb_file_name) staticmethod

Add the specified NWB file to the list of locked items.

The NWB_LOCK_FILE environment variable must be set to the path of the lock file, listing locked NWB files.

Parameters:

Name	Type	Description	Default
nwb_file_name	str	The name of an NWB file in the Nwbfile table.	required

Source code in src/spyglass/common/common_nwbfile.py

@staticmethod
def add_to_lock(nwb_file_name: str) -> None:
    """Add the specified NWB file to the list of locked items.

    The NWB_LOCK_FILE environment variable must be set to the path of the
    lock file, listing locked NWB files.

    Parameters
    ----------
    nwb_file_name : str
        The name of an NWB file in the Nwbfile table.
    """
    if not (Nwbfile() & {"nwb_file_name": nwb_file_name}):
        raise FileNotFoundError(
            f"File not found in Nwbfile table. Cannot lock {nwb_file_name}"
        )

    with open(os.getenv("NWB_LOCK_FILE"), "a+") as lock_file:
        lock_file.write(f"{nwb_file_name}\n")

cleanup(delete_files=False) staticmethod

Remove the filepath entries for NWB files that are not in use.

This does not delete the files themselves unless delete_files=True is specified. Run this after deleting the Nwbfile() entries themselves.

Source code in src/spyglass/common/common_nwbfile.py

@staticmethod
def cleanup(delete_files: bool = False) -> None:
    """Remove the filepath entries for NWB files that are not in use.

    This does not delete the files themselves unless delete_files=True is
    specified. Run this after deleting the Nwbfile() entries themselves.
    """
    schema.external["raw"].delete(delete_external_files=delete_files)

AnalysisRegistry

Bases: Manual

Central registry tracking all custom AnalysisNwbfile tables.

This table maintains a record of all team-specific AnalysisNwbfile tables to enable coordinated cleanup, export operations, and cross-table queries. Tables are auto-registered when declared via SpyglassAnalysis mixin.

Key Methods: get_class(prefix) - Get AnalysisNwbfile class for a specific team prefix get_all_classes() - Get all registered AnalysisNwbfile class objects get_tracked_files() - Get all files tracked across all custom tables clear_cache() - Clear the class cache (useful for testing)

Usage: from spyglass.common import AnalysisRegistry

# View all registered tables
AnalysisRegistry().fetch()

# Get a specific team's table
MyTeamAnalysis = AnalysisRegistry().get_class("myteam")

# Get an instance
my_team_analysis = MyTeamAnalysis()

# Use enhanced helper methods
my_team_analysis.get_prefix()  # 'myteam'

Source code in src/spyglass/common/common_nwbfile.py

@schema
class AnalysisRegistry(dj.Manual):
    """Central registry tracking all custom AnalysisNwbfile tables.

    This table maintains a record of all team-specific AnalysisNwbfile tables
    to enable coordinated cleanup, export operations, and cross-table queries.
    Tables are auto-registered when declared via SpyglassAnalysis mixin.

    Key Methods:
        get_class(prefix) - Get AnalysisNwbfile class for a specific team prefix
        get_all_classes() - Get all registered AnalysisNwbfile class objects
        get_tracked_files() - Get all files tracked across all custom tables
        clear_cache() - Clear the class cache (useful for testing)

    Usage:
        from spyglass.common import AnalysisRegistry

        # View all registered tables
        AnalysisRegistry().fetch()

        # Get a specific team's table
        MyTeamAnalysis = AnalysisRegistry().get_class("myteam")

        # Get an instance
        my_team_analysis = MyTeamAnalysis()

        # Use enhanced helper methods
        my_team_analysis.get_prefix()  # 'myteam'
    """

    definition = """
    full_table_name: varchar(128)  # full table name of the analysis
    ---
    created_at = CURRENT_TIMESTAMP: timestamp  # when registered
    created_by : varchar(32)                   # who registered
    """

    # Class-level cache for dynamic table classes
    _class_cache: dict = {}

    def insert1(self, key: Union[str, dict], **kwargs) -> None:
        """Auto-add created_by if not provided.

        Parameters
        ----------
        key : str or dict
            The full_table_name as a string or a dict with the key
            'full_table_name'.
        kwargs : additional arguments to pass to insert1.
        """

        if isinstance(key, str):
            key = {"full_table_name": key}

        if query := self & key:
            logger.debug(f"Entry already exists: {key['full_table_name']}")
            return query

        if "created_by" not in key:
            key["created_by"] = dj.config["database.user"]

        super().insert1(key, **kwargs)

    # ---------------- Spawn analysis classes from table names ----------------

    @staticmethod
    def _parse_table_name(
        full_table_name: str,
    ) -> tuple[str, str, str, str]:
        """Parse full table name into components.

        Extracts database, table name, prefix, and suffix from a full name.

        Parameters
        ----------
        full_table_name : str
            Full table name in format `database`.`table_name`
            where database follows {prefix}_{suffix} convention.

        Returns
        -------
        database : str
            The database name (e.g., "testuser_nwbfile")
        table_name : str
            The table name (e.g., "analysis_nwbfile")
        prefix : str
            The database prefix before last underscore (e.g., "testuser")
        suffix : str
            The database suffix after last underscore (e.g., "nwbfile")

        Example
        -------
        >>> table_name = "`user_nwbfile`.`analysis_nwbfile`"
        >>> AnalysisRegistry._parse_table_name(table_name)
        ('user_nwbfile', 'analysis_nwbfile', 'user', 'nwbfile')
        """
        # Remove backticks and split into database and table
        database, table_name = full_table_name.replace("`", "").split(".")
        # Split database into prefix and suffix at last underscore
        prefix, suffix = database.rsplit("_", 1)

        return database, table_name, prefix, suffix

    def _is_valid_entry(
        self, full_table_name: str, raise_err: bool = True
    ) -> bool:
        """Check if the given table name corresponds to a valid SpyglassAnalysis.

        Parameters
        ----------
        full_table_name : str
            The full table name.
        raise_err : bool, optional
            If True, raise ValueError on invalid table name.

        Returns
        -------
        is_valid : bool
            True if the table is a valid SpyglassAnalysis, False otherwise.


        """
        database, table_name, prefix, suffix = self._parse_table_name(
            full_table_name
        )

        err = None

        if table_name != "analysis_nwbfile":
            err = f"Table name must be 'analysis_nwbfile': {table_name}"

        if suffix != "nwbfile":
            err = f"Database suffix must be 'nwbfile': {suffix}"

        # Validate prefix (alphanumeric and underscore only)
        if not re.match(r"^[a-z0-9_]+$", prefix, re.IGNORECASE):
            err = f"Invalid prefix format: {prefix}"

        if raise_err and err is not None:
            raise ValueError(err)

        return err is None

    @classmethod
    def _create_class(cls, full_name: str) -> type:
        """Create an enhanced custom analysis table class.

        Returns a class with:
        - All AnalysisMixin methods
        - Cached for reuse
        - Helper methods for common operations
        - Better repr and documentation

        Parameters
        ----------
        full_name : str
            Full table name (e.g., "`myteam_nwbfile`.`analysis_nwbfile`")

        Returns
        -------
        type
            Enhanced table class with helper methods
        """
        database, table_name, prefix, _ = cls._parse_table_name(full_name)
        camel_name = dj.utils.to_camel_case(table_name)

        if (
            database not in dj.list_schemas()
            or table_name not in dj.Schema(database).list_tables()
        ):
            raise dj.errors.MissingTableError(
                f"Cannot create class for missing table: {full_name}. "
                "Ensure the schema is created and you have permissions to it."
                f"with dj.list_schemas(); dj.Schema({database}).list_tables()"
            )

        class AnalysisMeta(type):
            """Metaclass for custom AnalysisNwbfile classes."""

            def __repr__(cls):
                """Enhanced class repr showing prefix."""
                return (
                    f"<class '{cls.__name__}'"
                    + f"prefix='{cls._analysis_prefix}'>"
                )

        class EnhancedAnalysisNwbfile(
            SpyglassAnalysis, dj.FreeTable, metaclass=AnalysisMeta
        ):
            f"""Custom AnalysisNwbfile table for {prefix}.

            Automatically created by AnalysisRegistry for schema {full_name}.
            Provides same functionality as common AnalysisNwbfile but with
            isolated database locks.
            """

            full_table_name = full_name
            _analysis_prefix = prefix

            def __init__(self):
                # Always pass connection and table name to FreeTable
                super().__init__(conn=dj.conn(), full_table_name=full_name)

            def __repr__(self) -> str:
                """Enhanced repr showing custom table info."""

                return (
                    f"<{camel_name} (custom '{prefix}' analysis table)>\n"
                    + super().__repr__()
                )

        # Set the class name dynamically
        EnhancedAnalysisNwbfile.__name__ = camel_name
        EnhancedAnalysisNwbfile.__qualname__ = camel_name
        EnhancedAnalysisNwbfile.__module__ = (
            f"spyglass.common.common_nwbfile[{prefix}]"
        )

        return EnhancedAnalysisNwbfile

    def _get_tbl_from_name(self, full_name: str) -> type:
        """Return cached or create enhanced table class.

        Now uses caching and creates enhanced classes with helper methods.

        Parameters
        ----------
        full_name : str
            The full table name.

        Returns
        -------
        type
            Enhanced table class with caching
        """
        # Check cache first
        if full_name in self._class_cache:
            return self._class_cache[full_name]

        # Create enhanced class
        cls = self._create_class(full_name)

        # Cache it
        self._class_cache[full_name] = cls

        return cls

    def get_class(self, key: Union[str, Dict]) -> Optional[type]:
        """Return the class object for the given full_table_name, uninitialized.

        Parameters
        ----------
        key : str or dict
            The prefix or full_table_name as a string or a dict with the key
            'full_table_name'.

        Returns
        -------
        class_obj : type or None
            The class object for the given full_table_name, or None.
        """
        if isinstance(key, str) and "analysis_nwbfile" not in key:
            key = f"`{key}_nwbfile`.`analysis_nwbfile`"
        if isinstance(key, str):
            key = {"full_table_name": key}

        # TODO: Add common case to table on registry declaration
        common_map = {
            Nwbfile().full_table_name: Nwbfile,
            AnalysisNwbfile().full_table_name: AnalysisNwbfile,
        }

        if key["full_table_name"] in common_map:
            return common_map[key["full_table_name"]]

        if not (self & key):
            logger.warning(f"Entry not found: {key['full_table_name']}")
            return None

        return self._get_tbl_from_name(key["full_table_name"])

    @property
    def all_classes(self) -> List[SpyglassAnalysis]:
        """Return all registered analysis table class objects, initialized.

        Returns
        -------
        class_objs : list of dj.FreeTable
            A list of all registered analysis table class objects.
        """
        return [
            self._get_tbl_from_name(key["full_table_name"])()
            for key in self.fetch(as_dict=True)
            if self._is_valid_entry(key["full_table_name"], raise_err=False)
        ]

    @classmethod
    def clear_cache(cls):
        """Clear the class cache.

        Useful for testing or when custom tables are modified.
        After clearing, the next call to get_class() will recreate
        the class objects.

        Examples
        --------
        >>> AnalysisRegistry.clear_cache()
        >>> # Next get_class() call will create fresh class instances
        """
        cls._class_cache.clear()

    def get_externals(
        self, store: str = "analysis"
    ) -> List[dj.external.ExternalTable]:
        """Return external table objects for all registered analysis schemas.

        External tables are used to manage externally stored files (e.g., on S3).
        Each custom AnalysisNwbfile schema has a corresponding external table
        named `{prefix}_nwbfile`.`~external_analysis`.

        Used for updating externals after file surgeries or migrations.

        Parameters
        ----------
        store : str, optional
            The external store name to use. Default is "analysis".
            This should match a configured store in dj.config['stores'].

        Returns
        -------
        externals : list of dj.external.ExternalTable
            A list of ExternalTable objects for all registered schemas.

        Example
        -------
        >>> from spyglass.common import AnalysisRegistry
        >>> registry = AnalysisRegistry()
        >>> externals = registry.get_externals()
        >>> for ext in externals:
        ...     print(ext.database)
        """
        ExtTable = dj.external.ExternalTable
        ext_kwargs = dict(connection=dj.conn(), store=store)

        # Get unique database prefixes from registered tables
        databases = set(
            [
                self._parse_table_name(tbl_name)[0]
                for tbl_name in self.fetch("full_table_name")
                if self._is_valid_entry(tbl_name, raise_err=False)
            ]
        )

        return [  # Create ExternalTable for each database
            ExtTable(**ext_kwargs, database=database)
            for database in sorted(databases)
        ]

    # ------------------ Blocking inserts during maintenance ------------------

    def _get_block_info(self, table: str) -> str:
        """Parse table name to get database and trigger name."""
        _ = self._is_valid_entry(table, raise_err=True)

        # Extract database and prefix using helper method
        database, _, prefix, _ = self._parse_table_name(table)

        return database, f"{prefix}_block_inserts"

    def _block_exists(self, table: str) -> bool:
        """Check if a block trigger exists for the given table.

        Parameters
        ----------
        table : str
            The full table name of the analysis table to check.

        Returns
        -------
        exists : bool
            True if the block trigger exists, False otherwise.
        """
        database, trigger = self._get_block_info(table)
        kwargs = dict(database=database, trigger=trigger, table=table)

        result = dj.conn().query(SQL_TRIGGER_QUERY.format(**kwargs))
        return result.fetchone()[0] > 0

    def _block_single_table(
        self, table: str, dry_run: bool = False
    ) -> Optional[str]:
        """Block new inserts into a single analysis table.

        Parameters
        ----------
        table : str
            The full table name of the analysis table to block.
        dry_run : bool, optional
            If True, log blocking without making changes. Defaults to False.

        Returns
        -------
        error_msg : str or None
            A message when the table could not be blocked -- trigger creation
            failed, or a blocker already exists -- otherwise None.
        """
        try:
            database, trigger = self._get_block_info(table)
            kwargs = dict(database=database, trigger=trigger, table=table)

            if self._block_exists(table):
                return (
                    f"Failed to block {table}: blocking trigger already "
                    "exists; another cleanup may be active or the trigger "
                    "may be stale"
                )

            if dry_run:
                logger.info(f"Dry run: would block inserts into {table}")
                return None

            # Create trigger
            dj.conn().query(SQL_BLOCK_TEMPLATE.format(**kwargs))

        except Exception as e:
            return f"Failed to block {table}: {e}"

        return None

    def block_new_inserts(self, dry_run: bool = False) -> None:
        """Block new inserts into all registered analysis tables.

        Creates BEFORE INSERT triggers on all registered custom analysis tables
        to prevent data modifications during maintenance operations. Refuses
        to adopt an existing trigger because it may belong to another cleanup.

        Parameters
        ----------
        dry_run : bool, optional
            If True, log blocking without making changes. Defaults to False.

        Raises
        ------
        RuntimeError
            If blocker inspection or trigger creation fails, or if any blocker
            already exists. A partial creation failure may leave some tables
            blocked because triggers do not carry per-run ownership.
        """
        # Freeze one deterministic acquisition order. Concurrent acquisitions
        # with the same stable registry snapshot then contend on the same first
        # trigger rather than each creating a different prefix of the set.
        tables = sorted(set(self.fetch("full_table_name")))

        # Inspect every table before any DDL. A pre-existing trigger is either
        # owned by an active cleanup or stale after a failed one; adopting it
        # would let this call later remove a trigger it did not create.
        existing = []
        for table in tables:
            try:
                if self._block_exists(table):
                    existing.append(table)
            except Exception as err:
                raise RuntimeError(
                    f"Failed to inspect insert blocker for {table}; refusing "
                    f"cleanup before creating any triggers: {err}"
                ) from err

        if existing:
            blocked = "\n".join(f"  - {table}" for table in existing)
            raise RuntimeError(
                "Refusing cleanup because insert-blocking triggers already "
                f"exist for:\n{blocked}\n{STALE_BLOCKER_GUIDANCE}"
            )

        for table in tables:
            error = self._block_single_table(table, dry_run=dry_run)
            if error is not None:
                # Raise instead of continuing: this may be a concurrent trigger
                # claim or another DDL failure. Continuing would add more
                # triggers to a partial, ambiguous acquisition.
                raise RuntimeError(
                    f"Failed to block 1 table(s):\n{error}\nSome analysis "
                    f"tables may remain blocked. {STALE_BLOCKER_GUIDANCE}"
                )

    def unblock_new_inserts(self) -> None:
        """Unblock new inserts into all registered analysis tables.

        Removes BEFORE INSERT triggers from all registered custom analysis
        tables, re-enabling normal insert operations.

        Raises
        ------
        RuntimeError
            If any trigger removal fails.
        """
        errors = []

        for table in self.fetch("full_table_name"):
            try:
                database, trigger = self._get_block_info(table)
                if not self._block_exists(table):
                    continue
                dj.conn().query(f"DROP TRIGGER {database}.{trigger};")
            except Exception as e:
                errors.append(f"Failed to unblock {table}: {e}")

        if errors:
            raise RuntimeError(
                f"Failed to unblock {len(errors)} table(s):\n"
                + "\n".join(errors)
            )

insert1(key, **kwargs)

Auto-add created_by if not provided.

Parameters:

Name	Type	Description	Default
key	str or dict	The full_table_name as a string or a dict with the key 'full_table_name'.	required
kwargs	additional arguments to pass to insert1.		{}

Source code in src/spyglass/common/common_nwbfile.py

def insert1(self, key: Union[str, dict], **kwargs) -> None:
    """Auto-add created_by if not provided.

    Parameters
    ----------
    key : str or dict
        The full_table_name as a string or a dict with the key
        'full_table_name'.
    kwargs : additional arguments to pass to insert1.
    """

    if isinstance(key, str):
        key = {"full_table_name": key}

    if query := self & key:
        logger.debug(f"Entry already exists: {key['full_table_name']}")
        return query

    if "created_by" not in key:
        key["created_by"] = dj.config["database.user"]

    super().insert1(key, **kwargs)

get_class(key)

Return the class object for the given full_table_name, uninitialized.

Parameters:

Name	Type	Description	Default
key	str or dict	The prefix or full_table_name as a string or a dict with the key 'full_table_name'.	required

Returns:

Name	Type	Description
class_obj	type or None	The class object for the given full_table_name, or None.

Source code in src/spyglass/common/common_nwbfile.py

def get_class(self, key: Union[str, Dict]) -> Optional[type]:
    """Return the class object for the given full_table_name, uninitialized.

    Parameters
    ----------
    key : str or dict
        The prefix or full_table_name as a string or a dict with the key
        'full_table_name'.

    Returns
    -------
    class_obj : type or None
        The class object for the given full_table_name, or None.
    """
    if isinstance(key, str) and "analysis_nwbfile" not in key:
        key = f"`{key}_nwbfile`.`analysis_nwbfile`"
    if isinstance(key, str):
        key = {"full_table_name": key}

    # TODO: Add common case to table on registry declaration
    common_map = {
        Nwbfile().full_table_name: Nwbfile,
        AnalysisNwbfile().full_table_name: AnalysisNwbfile,
    }

    if key["full_table_name"] in common_map:
        return common_map[key["full_table_name"]]

    if not (self & key):
        logger.warning(f"Entry not found: {key['full_table_name']}")
        return None

    return self._get_tbl_from_name(key["full_table_name"])

all_classes property

Return all registered analysis table class objects, initialized.

Returns:

Name	Type	Description
class_objs	list of dj.FreeTable	A list of all registered analysis table class objects.

clear_cache() classmethod

Clear the class cache.

Useful for testing or when custom tables are modified. After clearing, the next call to get_class() will recreate the class objects.

Examples:

>>> AnalysisRegistry.clear_cache()
>>> # Next get_class() call will create fresh class instances

Source code in src/spyglass/common/common_nwbfile.py

@classmethod
def clear_cache(cls):
    """Clear the class cache.

    Useful for testing or when custom tables are modified.
    After clearing, the next call to get_class() will recreate
    the class objects.

    Examples
    --------
    >>> AnalysisRegistry.clear_cache()
    >>> # Next get_class() call will create fresh class instances
    """
    cls._class_cache.clear()

get_externals(store='analysis')

Return external table objects for all registered analysis schemas.

External tables are used to manage externally stored files (e.g., on S3). Each custom AnalysisNwbfile schema has a corresponding external table named {prefix}_nwbfile.~external_analysis.

Used for updating externals after file surgeries or migrations.

Parameters:

Name	Type	Description	Default
store	str	The external store name to use. Default is "analysis". This should match a configured store in dj.config['stores'].	'analysis'

Returns:

Name	Type	Description
externals	list of dj.external.ExternalTable	A list of ExternalTable objects for all registered schemas.

Example

from spyglass.common import AnalysisRegistry registry = AnalysisRegistry() externals = registry.get_externals() for ext in externals: ... print(ext.database)

Source code in src/spyglass/common/common_nwbfile.py

def get_externals(
    self, store: str = "analysis"
) -> List[dj.external.ExternalTable]:
    """Return external table objects for all registered analysis schemas.

    External tables are used to manage externally stored files (e.g., on S3).
    Each custom AnalysisNwbfile schema has a corresponding external table
    named `{prefix}_nwbfile`.`~external_analysis`.

    Used for updating externals after file surgeries or migrations.

    Parameters
    ----------
    store : str, optional
        The external store name to use. Default is "analysis".
        This should match a configured store in dj.config['stores'].

    Returns
    -------
    externals : list of dj.external.ExternalTable
        A list of ExternalTable objects for all registered schemas.

    Example
    -------
    >>> from spyglass.common import AnalysisRegistry
    >>> registry = AnalysisRegistry()
    >>> externals = registry.get_externals()
    >>> for ext in externals:
    ...     print(ext.database)
    """
    ExtTable = dj.external.ExternalTable
    ext_kwargs = dict(connection=dj.conn(), store=store)

    # Get unique database prefixes from registered tables
    databases = set(
        [
            self._parse_table_name(tbl_name)[0]
            for tbl_name in self.fetch("full_table_name")
            if self._is_valid_entry(tbl_name, raise_err=False)
        ]
    )

    return [  # Create ExternalTable for each database
        ExtTable(**ext_kwargs, database=database)
        for database in sorted(databases)
    ]

block_new_inserts(dry_run=False)

Block new inserts into all registered analysis tables.

Creates BEFORE INSERT triggers on all registered custom analysis tables to prevent data modifications during maintenance operations. Refuses to adopt an existing trigger because it may belong to another cleanup.

Parameters:

Name	Type	Description	Default
dry_run	bool	If True, log blocking without making changes. Defaults to False.	False

Raises:

Type	Description
RuntimeError	If blocker inspection or trigger creation fails, or if any blocker already exists. A partial creation failure may leave some tables blocked because triggers do not carry per-run ownership.

Source code in src/spyglass/common/common_nwbfile.py

def block_new_inserts(self, dry_run: bool = False) -> None:
    """Block new inserts into all registered analysis tables.

    Creates BEFORE INSERT triggers on all registered custom analysis tables
    to prevent data modifications during maintenance operations. Refuses
    to adopt an existing trigger because it may belong to another cleanup.

    Parameters
    ----------
    dry_run : bool, optional
        If True, log blocking without making changes. Defaults to False.

    Raises
    ------
    RuntimeError
        If blocker inspection or trigger creation fails, or if any blocker
        already exists. A partial creation failure may leave some tables
        blocked because triggers do not carry per-run ownership.
    """
    # Freeze one deterministic acquisition order. Concurrent acquisitions
    # with the same stable registry snapshot then contend on the same first
    # trigger rather than each creating a different prefix of the set.
    tables = sorted(set(self.fetch("full_table_name")))

    # Inspect every table before any DDL. A pre-existing trigger is either
    # owned by an active cleanup or stale after a failed one; adopting it
    # would let this call later remove a trigger it did not create.
    existing = []
    for table in tables:
        try:
            if self._block_exists(table):
                existing.append(table)
        except Exception as err:
            raise RuntimeError(
                f"Failed to inspect insert blocker for {table}; refusing "
                f"cleanup before creating any triggers: {err}"
            ) from err

    if existing:
        blocked = "\n".join(f"  - {table}" for table in existing)
        raise RuntimeError(
            "Refusing cleanup because insert-blocking triggers already "
            f"exist for:\n{blocked}\n{STALE_BLOCKER_GUIDANCE}"
        )

    for table in tables:
        error = self._block_single_table(table, dry_run=dry_run)
        if error is not None:
            # Raise instead of continuing: this may be a concurrent trigger
            # claim or another DDL failure. Continuing would add more
            # triggers to a partial, ambiguous acquisition.
            raise RuntimeError(
                f"Failed to block 1 table(s):\n{error}\nSome analysis "
                f"tables may remain blocked. {STALE_BLOCKER_GUIDANCE}"
            )

unblock_new_inserts()

Unblock new inserts into all registered analysis tables.

Removes BEFORE INSERT triggers from all registered custom analysis tables, re-enabling normal insert operations.

Raises:

Type	Description
RuntimeError	If any trigger removal fails.

Source code in src/spyglass/common/common_nwbfile.py

def unblock_new_inserts(self) -> None:
    """Unblock new inserts into all registered analysis tables.

    Removes BEFORE INSERT triggers from all registered custom analysis
    tables, re-enabling normal insert operations.

    Raises
    ------
    RuntimeError
        If any trigger removal fails.
    """
    errors = []

    for table in self.fetch("full_table_name"):
        try:
            database, trigger = self._get_block_info(table)
            if not self._block_exists(table):
                continue
            dj.conn().query(f"DROP TRIGGER {database}.{trigger};")
        except Exception as e:
            errors.append(f"Failed to unblock {table}: {e}")

    if errors:
        raise RuntimeError(
            f"Failed to unblock {len(errors)} table(s):\n"
            + "\n".join(errors)
        )

AnalysisNwbfile

Bases: SpyglassAnalysis, Manual

Source code in src/spyglass/common/common_nwbfile.py

@schema
class AnalysisNwbfile(SpyglassAnalysis, dj.Manual):
    definition = """
    # Table for NWB files that contain results of analysis.
    analysis_file_name: varchar(64)                # name of the file
    ---
    -> Nwbfile                                     # name of the parent NWB file. Used for naming and metadata copy
    analysis_file_abs_path: filepath@analysis      # the full path to the file
    analysis_file_description = "": varchar(2000)  # an optional description of this analysis
    analysis_parameters = NULL: blob               # additional relevant parameters. Currently used only for analyses
                                                   # that span multiple NWB files
    INDEX (analysis_file_abs_path)
    """
    # NOTE the INDEX above is implicit from filepath@...
    # above but needs to be explicit so that alter() can work

    # See #630, #664. Excessive key length.

    def _build_untracked_file_plan(
        self,
        custom_tables: List[SpyglassAnalysis],
        *,
        policy: CleanupPolicy,
        now_ns: Optional[int] = None,
    ) -> CleanupPlan:
        """Snapshot tracked paths once and scan the analysis directory."""
        tracked = set()
        for table in [self, *custom_tables]:
            tracked.update(
                Path(path) for _, path in table._ext_tbl.fetch_external_paths()
            )

        return CleanupPlan(
            self._analysis_dir,
            tracked,
            logger=logger,
            policy=policy,
            now_ns=now_ns,
        )

    def _cleanup_custom_table(
        self,
        analysis_tbl: SpyglassAnalysis,
        common_orphans: dj.expression.QueryExpression,
        dry_run: bool,
        table_num: int,
        num_tables: int,
    ) -> dj.expression.QueryExpression:
        """Clean up a single custom analysis table.

        Parameters
        ----------
        analysis_tbl : SpyglassAnalysis
            The custom analysis table to clean up.
        common_orphans : dj.expression.QueryExpression
            The common orphans to update with valid entries.
        dry_run : bool
            If True, only report what would be deleted.
        table_num : int
            Current table number for logging.
        num_tables : int
            Total number of tables for logging.

        Returns
        -------
        dj.expression.QueryExpression
            Updated common orphans with valid entries removed.
        """
        prefix = analysis_tbl.database.split("_")[0]

        # Delete orphans from this analysis table
        orphans = analysis_tbl.delete_orphans(dry_run=dry_run, safemode=False)
        n_orphans = len(orphans) if orphans is not None else 0

        # Clean up this table's external entries
        unused = analysis_tbl.cleanup_external(
            dry_run=dry_run, delete_external_files=True
        )
        self._info_msg(
            f"  [{table_num}/{num_tables}] {prefix}: {n_orphans} orphans, "
            + f"{len(unused)} unused externals"
        )

        # Remove valid entries from common orphans
        if bool(analysis_tbl):
            common_orphans -= analysis_tbl.proj()

        return common_orphans

    def cleanup(
        self,
        dry_run: bool = False,
        *,
        max_delete_fraction: float = 0.9,
        max_delete_to_tracked_ratio: float = 10.0,
        min_file_age_hours: float = 24.0,
    ) -> None:
        """Clean up common and all custom AnalysisNwbfile tables.

        Removes orphaned analysis files across both common and custom tables.
        A file is considered orphaned if it has no downstream foreign key
        references. This method coordinates cleanup across all registered
        custom AnalysisNwbfile tables to prevent premature deletion.

        Process:
            1. Discover custom tables and snapshot filesystem/tracking state.
            2. Validate the complete untracked-file deletion plan.
            3. Delete eligible filesystem candidates from that snapshot.
            4. For each custom analysis table:
               a. Delete orphaned entries (no downstream references)
               b. Clean up unused external file entries
               c. Remove valid entries from common orphan list
            5. Delete remaining common orphans.
            6. Clean up common external entries without deleting their files.

        Example:
            from spyglass.common import AnalysisNwbfile

            # Run cleanup across all tables
            AnalysisNwbfile().cleanup(dry_run=False)

        Note:
            This is a destructive operation. Ensure you have backups before
            running cleanup on production databases. File deletions cannot
            be undone.

        See Also:
            docs/src/ForDevelopers/Management.md for detailed cleanup guide.

        Parameters
        ----------
        dry_run : bool
            If True, perform a non-destructive dry run: log and report all
            cleanup actions without deleting database entries or files.
            If False, apply the cleanup changes, including deleting orphaned
            entries and associated files.
        max_delete_fraction : float
            Maximum fraction of eligible analysis NWB files that may be
            deleted by filesystem cleanup. The eligible set is the planned
            deletions plus scanned files recognized as tracked; age-deferred
            files are excluded. Set high by default (0.9) so it catches a
            catastrophically misconfigured analysis directory rather than
            routine large cleanups. Defaults to 0.9.
        max_delete_to_tracked_ratio : float
            Maximum ratio of filesystem cleanup deletions to tracked analysis
            files found in the scan. At the default ``max_delete_fraction``
            this limit cannot bind: writing D for deletions and T for tracked
            files, ``D / (D + T) <= 0.9`` forces ``D / T <= 9``, so any plan
            that clears the fraction limit is already within the ratio limit.
            It becomes the operative guard only when
            ``max_delete_fraction`` is raised above 10/11 (~0.909).
            This limit applies only to
            filesystem deletion of untracked or empty analysis NWB files, not
            to orphan row deletion. Defaults to 10.0.
        min_file_age_hours : float
            Untracked files newer than this are deferred to the next cleanup
            rather than deleted, protecting work that exists on disk but is
            not yet registered -- notably a file written to another volume
            and symlinked in before its row is inserted. Defaults to 24.0.
            The target modification time is the age basis. Pass 0 only for
            intentional immediate cleanup.

        Raises
        ------
        ValueError
            If a numeric safety limit is non-finite or outside its bounds.
        RuntimeError
            If insert blocking or unblocking fails, a destructive plan is
            refused, or registry/database cleanup fails.
        """
        # Validate every limit into an immutable policy BEFORE any
        # insert-blocking trigger is acquired. An unvalidated NaN or inf would
        # make every comparison False and silently disable the guard; worse,
        # validating only later (inside the plan) would leave triggers
        # installed on a bad argument. The plan reuses this validated policy
        # at the deletion boundary without repeating the raw numeric checks.
        policy = CleanupPolicy(
            max_delete_fraction=max_delete_fraction,
            max_delete_to_tracked_ratio=max_delete_to_tracked_ratio,
            min_file_age_hours=min_file_age_hours,
        )

        heading = "============== Analysis Cleanup "
        suffix = "(Dry Run) ==============" if dry_run else "=============="
        self._info_msg(heading + suffix)

        registry = AnalysisRegistry()
        # Stays OUTSIDE the try. Moving it inside would let a partial
        # acquisition fall into `finally: unblock_new_inserts()`, which
        # drops EVERY trigger including ones owned by a concurrent run.
        # Full ownership tracking needs a cleanup lease (follow-up).
        registry.block_new_inserts(dry_run=dry_run)

        # An explicit flag, not sys.exc_info(): that returns the exception
        # being handled ANYWHERE up the calling stack, so a caller shaped
        # `except Exception: cleanup()` would make it non-None even when the
        # body succeeded -- silently downgrading an unblock failure that
        # leaves inserts blocked database-wide.
        body_failed = False
        try:
            # Inside the try: a throw from get_orphans() previously landed
            # between block and try, leaving insert triggers installed
            # database-wide with no unblock.
            custom_tables = list(registry.all_classes)
            num_tables = len(custom_tables) + 1  # +1 for common table
            common_orphans = self.get_orphans().proj()

            untracked_file_plan = self._build_untracked_file_plan(
                custom_tables, policy=policy
            )

            # Delete files before database cleanup. Files newly orphaned by
            # this run's row deletion are caught on the next invocation. The
            # plan already carries the validated policy (age gate and deletion
            # limits), so no limits are re-passed here.
            untracked_file_plan.execute(dry_run=dry_run)

            # Process each custom analysis table.
            # Subtract valid entries from common_orphans
            for i, analysis_tbl in enumerate(custom_tables, start=1):
                common_orphans = self._cleanup_custom_table(
                    analysis_tbl, common_orphans, dry_run, i, num_tables
                )

            # Delete remaining common orphans
            n_orphans = len(common_orphans)

            if bool(common_orphans) and not dry_run:
                common_orphans.delete_quick()

            # Clean up common external table entries
            unused = self.cleanup_external(
                dry_run=dry_run, delete_external_files=False
            )

            self._info_msg(
                f"  [{num_tables}/{num_tables}] common: {n_orphans} "
                f"orphans, {len(unused)} unused externals"
            )

        except BaseException:
            body_failed = True
            raise

        finally:
            if not dry_run:
                try:
                    registry.unblock_new_inserts()
                except Exception as unblock_err:
                    # A failed unblock halts ALL inserts across the database
                    # until manually cleared, so this must be loud regardless
                    # of whether another exception is already propagating.
                    logger.critical(
                        "Failed to unblock inserts after cleanup: "
                        f"{unblock_err}. Analysis inserts remain BLOCKED "
                        "database-wide until restored; run "
                        "AnalysisRegistry().unblock_new_inserts() manually."
                    )
                    # Re-raise only when the body itself succeeded;
                    # otherwise we would mask the original cleanup error
                    # (the critical log above is the signal).
                    if not body_failed:
                        raise

    def check_all_files(
        self, resolve_tables: bool = False, verbose: bool = False
    ) -> dict:
        """Check files across all analysis tables for issues.

        Iterates through common and all custom AnalysisNwbfile tables,
        checking file existence and readability. Populates AnalysisFileIssues
        with any problems found. This monitoring operation does not delete
        files, but it does write issue rows and can be run independently of
        cleanup at different frequencies.

        Parameters
        ----------
        resolve_tables : bool, optional
            After all issues are collected, populate the table field for each
            issue by querying downstream child tables. More efficient than
            per-table resolution since children are fetched once per analysis
            table across all newly inserted issues. Default False.

        Returns
        -------
        results : dict
            Dictionary mapping table names to issue counts

        Example
        -------
        >>> from spyglass.common import AnalysisNwbfile
        >>> results = AnalysisNwbfile().check_all_files(resolve_tables=True)
        >>> print(f"Total issues: {sum(results.values())}")

        See Also
        --------
        AnalysisFileIssues : Table that stores detected issues
        AnalysisFileIssues.resolve_table_refs : Populate table field on demand
        """
        from spyglass.common.common_file_tracking import AnalysisFileIssues

        self._info_msg("Checking analysis files across all tables")
        registry = AnalysisRegistry()

        # Include common table + all custom tables
        analysis_tables = [self] + list(registry.all_classes)
        num_tables = len(analysis_tables)

        results = {}
        file_checker = AnalysisFileIssues()

        # B: Fetch recompute-deleted files once for all tables
        deleted_files = file_checker._get_recompute_deleted()

        for i, analysis_tbl in enumerate(analysis_tables, start=1):
            tbl_name = analysis_tbl.full_table_name
            self._info_msg(f"  [{i}/{num_tables}] Checking {tbl_name} files")

            issue_count = file_checker.check_files(
                analysis_tbl, deleted_files=deleted_files, verbose=verbose
            )
            results[tbl_name] = issue_count

            if issue_count > 0:
                logger.warning(f"    Found {issue_count} file issues")

        total_issues = sum(results.values())
        self._info_msg(f"File check complete: {total_issues} issues found")

        if resolve_tables and total_issues > 0:
            self._info_msg("Resolving downstream table references for issues")
            file_checker.resolve_table_refs()

        return results

cleanup(dry_run=False, *, max_delete_fraction=0.9, max_delete_to_tracked_ratio=10.0, min_file_age_hours=24.0)

Clean up common and all custom AnalysisNwbfile tables.

Removes orphaned analysis files across both common and custom tables. A file is considered orphaned if it has no downstream foreign key references. This method coordinates cleanup across all registered custom AnalysisNwbfile tables to prevent premature deletion.

Process: 1. Discover custom tables and snapshot filesystem/tracking state. 2. Validate the complete untracked-file deletion plan. 3. Delete eligible filesystem candidates from that snapshot. 4. For each custom analysis table: a. Delete orphaned entries (no downstream references) b. Clean up unused external file entries c. Remove valid entries from common orphan list 5. Delete remaining common orphans. 6. Clean up common external entries without deleting their files.

Example: from spyglass.common import AnalysisNwbfile

# Run cleanup across all tables
AnalysisNwbfile().cleanup(dry_run=False)

Note: This is a destructive operation. Ensure you have backups before running cleanup on production databases. File deletions cannot be undone.

See Also: docs/src/ForDevelopers/Management.md for detailed cleanup guide.

Parameters:

Name	Type	Description	Default
dry_run	bool	If True, perform a non-destructive dry run: log and report all cleanup actions without deleting database entries or files. If False, apply the cleanup changes, including deleting orphaned entries and associated files.	False
max_delete_fraction	float	Maximum fraction of eligible analysis NWB files that may be deleted by filesystem cleanup. The eligible set is the planned deletions plus scanned files recognized as tracked; age-deferred files are excluded. Set high by default (0.9) so it catches a catastrophically misconfigured analysis directory rather than routine large cleanups. Defaults to 0.9.	0.9
max_delete_to_tracked_ratio	float	Maximum ratio of filesystem cleanup deletions to tracked analysis files found in the scan. At the default max_delete_fraction this limit cannot bind: writing D for deletions and T for tracked files, D / (D + T) <= 0.9 forces D / T <= 9, so any plan that clears the fraction limit is already within the ratio limit. It becomes the operative guard only when max_delete_fraction is raised above 10/11 (~0.909). This limit applies only to filesystem deletion of untracked or empty analysis NWB files, not to orphan row deletion. Defaults to 10.0.	10.0
min_file_age_hours	float	Untracked files newer than this are deferred to the next cleanup rather than deleted, protecting work that exists on disk but is not yet registered -- notably a file written to another volume and symlinked in before its row is inserted. Defaults to 24.0. The target modification time is the age basis. Pass 0 only for intentional immediate cleanup.	24.0

Raises:

Type	Description
ValueError	If a numeric safety limit is non-finite or outside its bounds.
RuntimeError	If insert blocking or unblocking fails, a destructive plan is refused, or registry/database cleanup fails.

Source code in src/spyglass/common/common_nwbfile.py

def cleanup(
    self,
    dry_run: bool = False,
    *,
    max_delete_fraction: float = 0.9,
    max_delete_to_tracked_ratio: float = 10.0,
    min_file_age_hours: float = 24.0,
) -> None:
    """Clean up common and all custom AnalysisNwbfile tables.

    Removes orphaned analysis files across both common and custom tables.
    A file is considered orphaned if it has no downstream foreign key
    references. This method coordinates cleanup across all registered
    custom AnalysisNwbfile tables to prevent premature deletion.

    Process:
        1. Discover custom tables and snapshot filesystem/tracking state.
        2. Validate the complete untracked-file deletion plan.
        3. Delete eligible filesystem candidates from that snapshot.
        4. For each custom analysis table:
           a. Delete orphaned entries (no downstream references)
           b. Clean up unused external file entries
           c. Remove valid entries from common orphan list
        5. Delete remaining common orphans.
        6. Clean up common external entries without deleting their files.

    Example:
        from spyglass.common import AnalysisNwbfile

        # Run cleanup across all tables
        AnalysisNwbfile().cleanup(dry_run=False)

    Note:
        This is a destructive operation. Ensure you have backups before
        running cleanup on production databases. File deletions cannot
        be undone.

    See Also:
        docs/src/ForDevelopers/Management.md for detailed cleanup guide.

    Parameters
    ----------
    dry_run : bool
        If True, perform a non-destructive dry run: log and report all
        cleanup actions without deleting database entries or files.
        If False, apply the cleanup changes, including deleting orphaned
        entries and associated files.
    max_delete_fraction : float
        Maximum fraction of eligible analysis NWB files that may be
        deleted by filesystem cleanup. The eligible set is the planned
        deletions plus scanned files recognized as tracked; age-deferred
        files are excluded. Set high by default (0.9) so it catches a
        catastrophically misconfigured analysis directory rather than
        routine large cleanups. Defaults to 0.9.
    max_delete_to_tracked_ratio : float
        Maximum ratio of filesystem cleanup deletions to tracked analysis
        files found in the scan. At the default ``max_delete_fraction``
        this limit cannot bind: writing D for deletions and T for tracked
        files, ``D / (D + T) <= 0.9`` forces ``D / T <= 9``, so any plan
        that clears the fraction limit is already within the ratio limit.
        It becomes the operative guard only when
        ``max_delete_fraction`` is raised above 10/11 (~0.909).
        This limit applies only to
        filesystem deletion of untracked or empty analysis NWB files, not
        to orphan row deletion. Defaults to 10.0.
    min_file_age_hours : float
        Untracked files newer than this are deferred to the next cleanup
        rather than deleted, protecting work that exists on disk but is
        not yet registered -- notably a file written to another volume
        and symlinked in before its row is inserted. Defaults to 24.0.
        The target modification time is the age basis. Pass 0 only for
        intentional immediate cleanup.

    Raises
    ------
    ValueError
        If a numeric safety limit is non-finite or outside its bounds.
    RuntimeError
        If insert blocking or unblocking fails, a destructive plan is
        refused, or registry/database cleanup fails.
    """
    # Validate every limit into an immutable policy BEFORE any
    # insert-blocking trigger is acquired. An unvalidated NaN or inf would
    # make every comparison False and silently disable the guard; worse,
    # validating only later (inside the plan) would leave triggers
    # installed on a bad argument. The plan reuses this validated policy
    # at the deletion boundary without repeating the raw numeric checks.
    policy = CleanupPolicy(
        max_delete_fraction=max_delete_fraction,
        max_delete_to_tracked_ratio=max_delete_to_tracked_ratio,
        min_file_age_hours=min_file_age_hours,
    )

    heading = "============== Analysis Cleanup "
    suffix = "(Dry Run) ==============" if dry_run else "=============="
    self._info_msg(heading + suffix)

    registry = AnalysisRegistry()
    # Stays OUTSIDE the try. Moving it inside would let a partial
    # acquisition fall into `finally: unblock_new_inserts()`, which
    # drops EVERY trigger including ones owned by a concurrent run.
    # Full ownership tracking needs a cleanup lease (follow-up).
    registry.block_new_inserts(dry_run=dry_run)

    # An explicit flag, not sys.exc_info(): that returns the exception
    # being handled ANYWHERE up the calling stack, so a caller shaped
    # `except Exception: cleanup()` would make it non-None even when the
    # body succeeded -- silently downgrading an unblock failure that
    # leaves inserts blocked database-wide.
    body_failed = False
    try:
        # Inside the try: a throw from get_orphans() previously landed
        # between block and try, leaving insert triggers installed
        # database-wide with no unblock.
        custom_tables = list(registry.all_classes)
        num_tables = len(custom_tables) + 1  # +1 for common table
        common_orphans = self.get_orphans().proj()

        untracked_file_plan = self._build_untracked_file_plan(
            custom_tables, policy=policy
        )

        # Delete files before database cleanup. Files newly orphaned by
        # this run's row deletion are caught on the next invocation. The
        # plan already carries the validated policy (age gate and deletion
        # limits), so no limits are re-passed here.
        untracked_file_plan.execute(dry_run=dry_run)

        # Process each custom analysis table.
        # Subtract valid entries from common_orphans
        for i, analysis_tbl in enumerate(custom_tables, start=1):
            common_orphans = self._cleanup_custom_table(
                analysis_tbl, common_orphans, dry_run, i, num_tables
            )

        # Delete remaining common orphans
        n_orphans = len(common_orphans)

        if bool(common_orphans) and not dry_run:
            common_orphans.delete_quick()

        # Clean up common external table entries
        unused = self.cleanup_external(
            dry_run=dry_run, delete_external_files=False
        )

        self._info_msg(
            f"  [{num_tables}/{num_tables}] common: {n_orphans} "
            f"orphans, {len(unused)} unused externals"
        )

    except BaseException:
        body_failed = True
        raise

    finally:
        if not dry_run:
            try:
                registry.unblock_new_inserts()
            except Exception as unblock_err:
                # A failed unblock halts ALL inserts across the database
                # until manually cleared, so this must be loud regardless
                # of whether another exception is already propagating.
                logger.critical(
                    "Failed to unblock inserts after cleanup: "
                    f"{unblock_err}. Analysis inserts remain BLOCKED "
                    "database-wide until restored; run "
                    "AnalysisRegistry().unblock_new_inserts() manually."
                )
                # Re-raise only when the body itself succeeded;
                # otherwise we would mask the original cleanup error
                # (the critical log above is the signal).
                if not body_failed:
                    raise

check_all_files(resolve_tables=False, verbose=False)

Check files across all analysis tables for issues.

Iterates through common and all custom AnalysisNwbfile tables, checking file existence and readability. Populates AnalysisFileIssues with any problems found. This monitoring operation does not delete files, but it does write issue rows and can be run independently of cleanup at different frequencies.

Parameters:

Name	Type	Description	Default
resolve_tables	bool	After all issues are collected, populate the table field for each issue by querying downstream child tables. More efficient than per-table resolution since children are fetched once per analysis table across all newly inserted issues. Default False.	False

Returns:

Name	Type	Description
results	dict	Dictionary mapping table names to issue counts

Example

from spyglass.common import AnalysisNwbfile results = AnalysisNwbfile().check_all_files(resolve_tables=True) print(f"Total issues: {sum(results.values())}")

See Also

AnalysisFileIssues : Table that stores detected issues AnalysisFileIssues.resolve_table_refs : Populate table field on demand

Source code in src/spyglass/common/common_nwbfile.py

def check_all_files(
    self, resolve_tables: bool = False, verbose: bool = False
) -> dict:
    """Check files across all analysis tables for issues.

    Iterates through common and all custom AnalysisNwbfile tables,
    checking file existence and readability. Populates AnalysisFileIssues
    with any problems found. This monitoring operation does not delete
    files, but it does write issue rows and can be run independently of
    cleanup at different frequencies.

    Parameters
    ----------
    resolve_tables : bool, optional
        After all issues are collected, populate the table field for each
        issue by querying downstream child tables. More efficient than
        per-table resolution since children are fetched once per analysis
        table across all newly inserted issues. Default False.

    Returns
    -------
    results : dict
        Dictionary mapping table names to issue counts

    Example
    -------
    >>> from spyglass.common import AnalysisNwbfile
    >>> results = AnalysisNwbfile().check_all_files(resolve_tables=True)
    >>> print(f"Total issues: {sum(results.values())}")

    See Also
    --------
    AnalysisFileIssues : Table that stores detected issues
    AnalysisFileIssues.resolve_table_refs : Populate table field on demand
    """
    from spyglass.common.common_file_tracking import AnalysisFileIssues

    self._info_msg("Checking analysis files across all tables")
    registry = AnalysisRegistry()

    # Include common table + all custom tables
    analysis_tables = [self] + list(registry.all_classes)
    num_tables = len(analysis_tables)

    results = {}
    file_checker = AnalysisFileIssues()

    # B: Fetch recompute-deleted files once for all tables
    deleted_files = file_checker._get_recompute_deleted()

    for i, analysis_tbl in enumerate(analysis_tables, start=1):
        tbl_name = analysis_tbl.full_table_name
        self._info_msg(f"  [{i}/{num_tables}] Checking {tbl_name} files")

        issue_count = file_checker.check_files(
            analysis_tbl, deleted_files=deleted_files, verbose=verbose
        )
        results[tbl_name] = issue_count

        if issue_count > 0:
            logger.warning(f"    Found {issue_count} file issues")

    total_issues = sum(results.values())
    self._info_msg(f"File check complete: {total_issues} issues found")

    if resolve_tables and total_issues > 0:
        self._info_msg("Resolving downstream table references for issues")
        file_checker.resolve_table_refs()

    return results