Database

Functions for database connection management and schema access.

Unified database schema management for Ethopy analysis.

This module provides a simple, unified interface for database connections and schema access with automatic caching and configuration management.

clear_schema_cache()

Clear all cached schemas.

Useful for testing, configuration changes, or when you want to force recreation of schemas.

Example

clear_schema_cache() schemas = get_all_schemas() # Will recreate schemas

Source code in src/ethopy_analysis/db/schemas.py

def clear_schema_cache():
    """
    Clear all cached schemas.

    Useful for testing, configuration changes, or when you want to force
    recreation of schemas.

    Example:
        clear_schema_cache()
        schemas = get_all_schemas()  # Will recreate schemas
    """
    global _cached_schemas
    _cached_schemas.clear()
    logger.info("Schema cache cleared")

get_all_schemas(config=None)

Get all three schemas (experiment, behavior, stimulus) at once.

This is the main function that handles caching and schema creation.

Parameters:

Name	Type	Description	Default
config	Optional[Dict[str, Any]]	Optional database configuration. If None, uses default config.	None

Returns:

Type	Description
Dict[str, Any]	Dictionary with keys 'experiment', 'behavior', 'stimulus' containing
Dict[str, Any]	the corresponding DataJoint virtual modules

Example

from ethopy_analysis.config.settings import load_config config = load_config() schemas = get_all_schemas(config) trials = schemas['experiment'].Trial.fetch(format='frame') licking = schemas['behavior'].Licking.fetch(format='frame') conditions = schemas['stimulus'].Condition.fetch(format='frame')

Source code in src/ethopy_analysis/db/schemas.py

def get_all_schemas(config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """
    Get all three schemas (experiment, behavior, stimulus) at once.

    This is the main function that handles caching and schema creation.

    Args:
        config: Optional database configuration. If None, uses default config.

    Returns:
        Dictionary with keys 'experiment', 'behavior', 'stimulus' containing
        the corresponding DataJoint virtual modules

    Example:
        from ethopy_analysis.config.settings import load_config
        config = load_config()
        schemas = get_all_schemas(config)
        trials = schemas['experiment'].Trial.fetch(format='frame')
        licking = schemas['behavior'].Licking.fetch(format='frame')
        conditions = schemas['stimulus'].Condition.fetch(format='frame')
    """
    # Get configuration (pass explicitly instead of loading internally)
    if config is None:
        # Import here to avoid circular imports
        from ..config.settings import get_database_config

        config = get_database_config()

    # Create simple cache key from configuration
    cache_key = _create_cache_key(config)

    # Return cached schemas if they exist
    if cache_key in _cached_schemas:
        logger.debug(f"Using cached schemas for: {_get_host_from_config(config)}")
        return _cached_schemas[cache_key]

    # Create new schemas (expensive operation)
    logger.info(f"Creating DataJoint schemas for: {_get_host_from_config(config)}")

    try:
        # Set up database connection and create schemas
        schemas = _setup_database_connection(config=config)

        # Cache the schemas
        _cached_schemas[cache_key] = schemas
        logger.info(f"Successfully cached schemas for: {_get_host_from_config(config)}")

        return schemas

    except Exception as e:
        logger.error(f"Failed to create DataJoint schemas: {e}")
        raise ConnectionError(f"Failed to create DataJoint schemas: {e}")

get_schema(schema_name, config=None)

Get a specific schema by name.

Parameters:

Name	Type	Description	Default
schema_name	str	Name of the schema to retrieve ('experiment', 'behavior', or 'stimulus')	required
config	Optional[Dict[str, Any]]	Optional database configuration. If None, uses default config.	None

Returns:

Type	Description
	DataJoint virtual module for the specified schema, or None if schema not found

Example

from ethopy_analysis.config.settings import load_config config = load_config() experiment = get_schema('experiment', config) behavior = get_schema('behavior', config) stimulus = get_schema('stimulus', config)

Usage examples:

trials_df = experiment.Trial.fetch(format='frame') licking_df = behavior.Licking.fetch(format='frame') conditions_df = stimulus.Condition.fetch(format='frame')

Source code in src/ethopy_analysis/db/schemas.py

def get_schema(schema_name: str, config: Optional[Dict[str, Any]] = None):
    """
    Get a specific schema by name.

    Args:
        schema_name: Name of the schema to retrieve ('experiment', 'behavior', or 'stimulus')
        config: Optional database configuration. If None, uses default config.

    Returns:
        DataJoint virtual module for the specified schema, or None if schema not found

    Example:
        from ethopy_analysis.config.settings import load_config
        config = load_config()
        experiment = get_schema('experiment', config)
        behavior = get_schema('behavior', config)
        stimulus = get_schema('stimulus', config)

        # Usage examples:
        trials_df = experiment.Trial.fetch(format='frame')
        licking_df = behavior.Licking.fetch(format='frame')
        conditions_df = stimulus.Condition.fetch(format='frame')
    """
    schemas = get_all_schemas(config)

    if schema_name not in schemas:
        logger.warning(
            f"Schema '{schema_name}' not found. Available schemas: {list(schemas.keys())}"
        )
        return None

    return schemas[schema_name]

show_cached_schemas()

Show information about currently cached schemas.

Returns:

Type	Description
Dict[str, str]	Dictionary with cache keys and basic info about cached schemas

Example

cached_info = show_cached_schemas() print(f"Cached configurations: {list(cached_info.keys())}")

Source code in src/ethopy_analysis/db/schemas.py

def show_cached_schemas() -> Dict[str, str]:
    """
    Show information about currently cached schemas.

    Returns:
        Dictionary with cache keys and basic info about cached schemas

    Example:
        cached_info = show_cached_schemas()
        print(f"Cached configurations: {list(cached_info.keys())}")
    """
    cache_info = {}

    for cache_key, schemas in _cached_schemas.items():
        # Extract host from simple cache key format (host_user_hash)
        try:
            host = cache_key.split("_")[0]
        except IndexError:
            host = "unknown"

        cache_info[cache_key] = {
            "host": host,
            "schemas": list(schemas.keys()),
            "num_schemas": len(schemas),
        }

    logger.info(f"Currently cached: {len(cache_info)} configurations")
    return cache_info

test_connection(config=None)

Test database connection and schema access.

Parameters:

Name	Type	Description	Default
config	Optional[Dict[str, Any]]	Optional database configuration. If None, uses default config.	None

Returns:

Type	Description
bool	True if connection successful, False otherwise

Example

from ethopy_analysis.config.settings import load_config config = load_config() if test_connection(config): print("Database connection successful!") else: print("Database connection failed!")

Source code in src/ethopy_analysis/db/schemas.py

def test_connection(config: Optional[Dict[str, Any]] = None) -> bool:
    """
    Test database connection and schema access.

    Args:
        config: Optional database configuration. If None, uses default config.

    Returns:
        True if connection successful, False otherwise

    Example:
        from ethopy_analysis.config.settings import load_config
        config = load_config()
        if test_connection(config):
            print("Database connection successful!")
        else:
            print("Database connection failed!")
    """
    try:
        if config is None:
            # Import here to avoid circular imports
            from ..config.settings import get_database_config

            config = get_database_config()

        # Test connection by creating schemas (without caching)
        _setup_database_connection(config=config)

        # Try a simple query to test the connection
        dj.conn().ping()

        logger.info("Database connection test successful")
        return True

    except Exception as e:
        logger.error(f"Database connection test failed: {e}")
        return False

validate_schema_config(config)

Validate that configuration contains required schema information.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Database configuration dictionary	required

Returns:

Type	Description
bool	True if valid, raises ValueError if invalid

Source code in src/ethopy_analysis/db/schemas.py

def validate_schema_config(config: Dict[str, Any]) -> bool:
    """
    Validate that configuration contains required schema information.

    Args:
        config: Database configuration dictionary

    Returns:
        True if valid, raises ValueError if invalid
    """
    # Handle both full config and database-only config
    if "database" in config:
        db_config = config["database"]
    else:
        db_config = config

    required_fields = ["host", "user", "password", "schemas"]
    missing_fields = [field for field in required_fields if field not in db_config]

    if missing_fields:
        raise ValueError(f"Missing required configuration fields: {missing_fields}")

    required_schemas = ["experiment", "behavior", "stimulus"]
    missing_schemas = [
        schema for schema in required_schemas if schema not in db_config["schemas"]
    ]

    if missing_schemas:
        raise ValueError(f"Missing required schema configurations: {missing_schemas}")

    return True