Utility Functions

def ensure_dir_exists(path: str | Path, mode: int = 0o777) -> None:
    """Ensure that a directory exists.
    
    If it doesn't exist, try to create it, protecting against race conditions
    if another process is doing the same. The default permissions are 
    determined by the current umask.
    
    Args:
        path: Directory path to create
        mode: Directory permissions (default: 0o777, modified by umask)
        
    Raises:
        OSError: If directory cannot be created or path exists but is not a directory
    """

from jupyter_core.utils import ensure_dir_exists
from pathlib import Path

# Ensure a config directory exists
config_dir = Path.home() / ".jupyter" / "custom"
ensure_dir_exists(config_dir, mode=0o700)  # User-only permissions

# Safe to use even if directory already exists
ensure_dir_exists(config_dir)  # No error if already exists

def deprecation(message: str, internal: str | list[str] = "jupyter_core/") -> None:
    """Generate a deprecation warning targeting the first frame that is not 'internal'.
    
    Automatically calculates the correct stack level to ensure the warning
    points to user code rather than internal library code.
    
    Args:
        message: Deprecation warning message
        internal: String or list of strings identifying internal code paths
                 (if these appear in frame filenames, frames are considered internal)
    """

from jupyter_core.utils import deprecation

def old_function():
    deprecation(
        "old_function is deprecated, use new_function instead",
        internal=["jupyter_core/", "mypackage/internal/"]
    )
    # Function implementation...

# When called by user code, warning points to user's call site
old_function()  # Warning shows this line, not internal deprecation() call

def ensure_event_loop(prefer_selector_loop: bool = False) -> asyncio.AbstractEventLoop:
    """Ensure an asyncio event loop exists and return it.
    
    Gets the current event loop, or creates a new one if none exists.
    Handles platform-specific loop selection.
    
    Args:
        prefer_selector_loop: On Windows, prefer SelectorEventLoop over default
                             (useful for Tornado compatibility)
        
    Returns:
        asyncio.AbstractEventLoop: Current or newly created event loop
    """

def run_sync(coro: Callable[..., Awaitable[T]]) -> Callable[..., T]:
    """Wraps coroutine in a function that blocks until it has executed.
    
    Allows calling async functions from sync code. Handles both cases
    where an event loop is already running (uses background thread) and
    where no loop exists (runs loop directly).
    
    Args:
        coro: Coroutine function to wrap
        
    Returns:
        Callable: Synchronous wrapper function
    """

async def ensure_async(obj: Awaitable[T] | T) -> T:
    """Convert a non-awaitable object to a coroutine if needed.
    
    Handles functions that may return either sync values or awaitables,
    ensuring consistent async handling.
    
    Args:
        obj: Object that may or may not be awaitable
        
    Returns:
        T: The result, whether it was originally sync or async
    """

from jupyter_core.utils import run_sync, ensure_event_loop
import asyncio

# Wrap an async function to be callable from sync code
@run_sync
async def async_operation(data):
    await asyncio.sleep(1)
    return f"Processed: {data}"

# Can now call async function synchronously
result = async_operation("test data")
print(result)  # "Processed: test data"

# Ensure event loop exists
loop = ensure_event_loop()
print(f"Loop type: {type(loop)}")

class _TaskRunner:
    """A task runner that runs an asyncio event loop on a background thread.
    
    Used internally by run_sync() to handle cases where an event loop
    is already running in the current thread.
    """
    
    def __init__(self) -> None:
        """Initialize the task runner."""
    
    def run(self, coro) -> Any:
        """Synchronously run a coroutine on a background thread.
        
        Args:
            coro: Coroutine to execute
            
        Returns:
            Any: Result of coroutine execution
        """

def _get_frame(level: int) -> FrameType | None:
    """Get the frame at the given stack level.
    
    Uses sys._getframe if available (faster) or falls back to inspect.stack.
    
    Args:
        level: Stack level to retrieve (0 = current frame)
        
    Returns:
        FrameType | None: Frame object or None if level is invalid
    """

def _external_stacklevel(internal: list[str]) -> int:
    """Find the stacklevel of the first frame that doesn't contain internal strings.
    
    Used by deprecation() to calculate the correct stacklevel for warnings.
    
    Args:
        internal: List of strings that identify internal code paths
        
    Returns:
        int: Stack level of first external frame
    """

from typing import Any, Awaitable, Callable, TypeVar
from types import FrameType
from contextvars import ContextVar
import asyncio

# Type variable for generic async functions
T = TypeVar("T")

# Context variable for tracking event loops
_loop: ContextVar[asyncio.AbstractEventLoop | None]

# Internal registry for task runners  
_runner_map: dict[str, _TaskRunner]