utils.filesystem

build_file_map

Build a dictionary mapping relative paths to absolute file paths for all files under the given absolute path.

Source 
def build_file_map(dir: Path) -> dict[str, str]
dir Path

Path to the directory to build the file map for

Returns

Dictionary where keys are relative paths and values are absolute paths on the local filesystem

copy_files

Copy files from absolute paths to relative paths under output_dir.

Source 
def copy_files(files: Mapping[str, str | Path], output_dir: Path) -> dict[str, Path]
files Mapping[str, str | Path]

A mapping of relative paths (keys) to absolute file paths (values)

output_dir Path

The base directory to copy files to

Returns

A dictionary mapping the original relative paths to the new absolute paths in output_dir

ensure_extension

Ensures the path has one of the given extensions, adding the first extension in the list if none is present.

Source 
def ensure_extension(path: str | Path, extensions: str | list[str]) -> str | Path
path str | Path

The path to ensure the extension of.

extensions str | list[str]

The extensions to ensure the path has.

Returns

A path with one of the given extensions.

filter_file_map

Filter files from a files mapping using optional include/exclude glob patterns.

Source 
def filter_file_map(
    files: Mapping[str, str],
    include: Sequence[str] | None = None,
    exclude: Sequence[str] | None = None,
) -> dict[str, str]
files Mapping[str, str]

Mapping of relative file paths to absolute file paths on host.

include Sequence[str] | None

Patterns to keep (e.g., [“**/*.py”]). If omitted/empty, all files are included.

exclude Sequence[str] | None

Patterns to drop. Applied after include filtering.

Returns

Dictionary mapping relative database file paths to their absolute paths containing only the files that match the include/exclude patterns.

find_unused_filename

Return path itself if it doesn’t exist, otherwise append an incrementing numeric suffix (_1, _2, …) before the extension until an unused name is found.

Source 
def find_unused_filename(path: Path) -> Path
path Path

load_data_file

Load data from a JSON or YAML file. The file type is automatically determined based on the file extension. The top-level data in the file must be a dictionary.

Source 
def load_data_file(file: Path) -> dict[str, Any]
file Path

Path to the data file

Returns

Dictionary containing the data

local_repo_dir

Resolve a repository directory to a local filesystem path, materializing from various sources.

  • If source is “local”: return the path directly.
  • If source is “metadata”: resolve location as a key in state.metadata.
    • If value is a path-like (str | Path): return that path.
    • If value is a Mapping[str, str | Path]: copy files into a temporary directory and return it.
  • If source is “sandbox”: copy the directory from the sandbox into a temporary directory and return it.
  • If source is “snapshot”: resolve location as <store_key>[<index>] or [<index>] where store_key is a key in the store containing an array of snapshot entries and index is the index of a snapshot entry. Use the snapshot_path from this entry to copy the directory from the sandbox into a temporary directory and return it. If store_key is omitted (i.e., format is [<index>]), it defaults to repo_snapshots. If index is first or last, the indexes 0 and -1 are used respectively.

The function supports passing a Sequence of locations to search; the first that resolves is used. A location string prefixed with one of “local:”, “metadata:”, “sandbox:”, or “snapshot:” overrides the source parameter for that entry only.

Source 
@asynccontextmanager
async def local_repo_dir(
    state: TaskState,
    location: RepoLocation | Sequence[RepoLocation],
    source: RepoLocationSource = "local",
    exclude: Sequence[str] = DEFAULT_EXCLUDE,
) -> AsyncIterator[Path]
state TaskState

TaskState containing a metadata dictionary

location RepoLocation | Sequence[RepoLocation]

Locations to search when resolving the repository directory

source RepoLocationSource

Where to look for the repository directory, if not overridden by a location string prefix (default: “local”)

exclude Sequence[str]

Returns

AsyncIterator[Path] yielding the resolved path

Raises

  • ValueError: If no location resolved from the given locations

Examples

# Resolves to the value of `state.metadata["repo_dir"]`
async with local_repo_dir(state, "metadata:repo_dir") as repo_path:
    print(repo_path)
# Resolves to the value of `state.metadata["repo_dir"]` if it exists, otherwise copies the
# `/workspace` directory from the sandbox into a temporary directory and returns it.
async with local_repo_dir(state, ["metadata:repo_dir", "sandbox:/workspace"]) as repo_path:
    print(repo_path)
# Copies `/workspace` from the sandbox to a temporary directory and returns it.
async with local_repo_dir(state, "/workspace", "sandbox") as repo_path:
    print(repo_path)
# Copies the first snapshot under the `repo_snapshots` key in the store into a temporary
# directory and returns the directory path.
async with local_repo_dir(state, "snapshot:[first]") as repo_path:
    print(repo_path)
# Copies the second snapshot under the `my_snapshots` key in the store into a temporary
# directory and returns the directory path.
async with local_repo_dir(state, "snapshot:my_snapshots[1]") as repo_path:
    print(repo_path)

normalize_path

Normalizes a path by resolving . and .. components.

This function handles forward-slash paths (e.g., for package resources or URLs) and removes redundant path components. Leading .. components are preserved since they cannot be resolved without knowing the parent path.

Source 
def normalize_path(path: str) -> str
path str

A path string that may contain . or .. components.

Returns

The normalized path with . and .. components resolved where possible.

Examples

normalize_path("a/b/../c")
# 'a/c'

normalize_path("a/./b/c")
# 'a/b/c'

normalize_path("../a/b")
# '../a/b'

normalize_path("a/../../b")
# '../b'

normalize_path("./a/b")
# 'a/b'

parse_repo_location

Parse a repository location into a source and pointer to the repository. The pointer may be a Path, a string path, a key in the metadata, or a snapshot index.

Source 
def parse_repo_location(
    location: RepoLocation,
    default_source: RepoLocationSource,
) -> tuple[RepoLocationSource, str | Path]
location RepoLocation

The repository location to parse

default_source RepoLocationSource

The source to use if the location is a Path or no source is specified in the location

Returns

A tuple of the source and a pointer to the repository

Raises

  • ValueError: If the repository location source is invalid

resolve_repo_location

Resolve a repository location or list of locations to a path or mapping.

  • If source is “local” or “sandbox”: return the path directly.
  • If source is “metadata”: resolve location as a key in state.metadata.
    • If value is a path-like (str | Path): return that path.
    • If value is a Mapping[str, str | Path]: return the mapping if allow_mappings is True, otherwise raise a ValueError.
  • If source is “snapshot”: resolve location as <store_key>[<index>] or [<index>] where store_key is a key in the store containing an array of snapshot entries and index is the index of a snapshot entry. Return the snapshot_path from this entry. If store_key is omitted (i.e., format is [<index>]), it defaults to repo_snapshots. If index is first or last, the indexes 0 and -1 are used respectively.

The function supports passing a Sequence of locations to search; the first that resolves is used. A location string prefixed with one of “local:”, “metadata:”, “sandbox:”, or “snapshot:” overrides the source parameter for that entry only.

Source 
async def resolve_repo_location(
    state: TaskState,
    location: RepoLocation | Sequence[RepoLocation],
    source: RepoLocationSource = "local",
    allow_mappings: bool = False,
) -> Path | Mapping[str, str | Path] | None
state TaskState

TaskState containing a metadata dictionary

location RepoLocation | Sequence[RepoLocation]

Locations to search when resolving the repository path or mapping

source RepoLocationSource

Where to look for the repository path or mapping, if not overridden by a location string prefix (default: “local”)

allow_mappings bool

Allow mappings to be returned (default: False)

Returns

Path | Mapping[str, str | Path] | None: The resolved path or mapping, or None if no location could be resolved

Raises

  • ValueError: If the location resolved to a mapping but allow_mappings is False.

to_path

Convert a string or Path to a Path efficiently.

Source 
def to_path(value: str | Path) -> Path
value str | Path