File

File is a special DataModel, which is automatically generated when a DataChain is created from files, such as in dc.read_storage:

import datachain as dc

chain = dc.read_storage("gs://datachain-demo/dogs-and-cats")
chain.print_schema()

Output:

file: File@v1
    source: str
    path: str
    size: int
    version: str
    etag: str
    is_latest: bool
    last_modified: datetime
    location: Union[dict, list[dict], NoneType]

File classes include various metadata fields describing the underlying file, along with methods to read and manipulate file contents.

File

File(**kwargs)

Bases: DataModel

DataModel for reading binary files.

Attributes:

• source (str) –

  The source of the file (e.g., 's3://bucket-name/').

• path (str) –

  The path to the file (e.g., 'path/to/file.txt').

• size (int) –

  The size of the file in bytes. Defaults to 0.

• version (str) –

  The version of the file. Defaults to an empty string.

• etag (str) –

  The ETag of the file. Defaults to an empty string.

• is_latest (bool) –

  Whether the file is the latest version. Defaults to True.

• last_modified (datetime) –

  The last modified timestamp of the file. Defaults to Unix epoch (1970-01-01T00:00:00).

• location (dict | list[dict]) –

  The location of the file. Defaults to None.

Source code in datachain/lib/file.py

def __init__(self, **kwargs):
    super().__init__(**kwargs)
    self._catalog = None
    self._caching_enabled: bool = False
    self._download_cb: Callback = DEFAULT_CALLBACK

as_audio_file

as_audio_file() -> AudioFile

Convert the file to a AudioFile object.

Source code in datachain/lib/file.py

def as_audio_file(self) -> "AudioFile":
    """Convert the file to a `AudioFile` object."""
    if isinstance(self, AudioFile):
        return self
    file = AudioFile(**self.model_dump())
    file._set_stream(self._catalog, caching_enabled=self._caching_enabled)
    return file

as_image_file

as_image_file() -> ImageFile

Convert the file to a ImageFile object.

Source code in datachain/lib/file.py

def as_image_file(self) -> "ImageFile":
    """Convert the file to a `ImageFile` object."""
    if isinstance(self, ImageFile):
        return self
    file = ImageFile(**self.model_dump())
    file._set_stream(self._catalog, caching_enabled=self._caching_enabled)
    return file

as_text_file

as_text_file() -> TextFile

Convert the file to a TextFile object.

Source code in datachain/lib/file.py

def as_text_file(self) -> "TextFile":
    """Convert the file to a `TextFile` object."""
    if isinstance(self, TextFile):
        return self
    file = TextFile(**self.model_dump())
    file._set_stream(self._catalog, caching_enabled=self._caching_enabled)
    return file

as_video_file

as_video_file() -> VideoFile

Convert the file to a VideoFile object.

Source code in datachain/lib/file.py

def as_video_file(self) -> "VideoFile":
    """Convert the file to a `VideoFile` object."""
    if isinstance(self, VideoFile):
        return self
    file = VideoFile(**self.model_dump())
    file._set_stream(self._catalog, caching_enabled=self._caching_enabled)
    return file

at classmethod

at(
    uri: str | PathLike[str], session: Session | None = None
) -> Self

Construct a File from a full URI in one call.

Example

file = File.at("s3://bucket/path/to/output.png") with file.open("wb") as f: ...

Source code in datachain/lib/file.py

@classmethod
def at(
    cls, uri: str | os.PathLike[str], session: "Session | None" = None
) -> "Self":
    """Construct a File from a full URI in one call.

    Example:
        file = File.at("s3://bucket/path/to/output.png")
        with file.open("wb") as f: ...
    """
    from datachain.client.fsspec import Client
    from datachain.query.session import Session

    if session is None:
        session = Session.get()
    catalog = session.catalog
    uri_str = stringify_path(uri)
    if uri_str.endswith(("/", os.sep)):
        raise ValueError(
            f"File.at directory URL/path given (trailing slash), got: {uri_str}"
        )
    client_cls = Client.get_implementation(uri_str)
    uri_str = client_cls.path_to_uri(uri_str)
    source, rel_path = client_cls.split_url(uri_str)
    source_uri = client_cls.get_uri(source)
    file = cls(source=source_uri, path=rel_path)
    file._set_stream(catalog)
    return file

export

export(
    output: str | PathLike[str],
    placement: ExportPlacement = "fullpath",
    use_cache: bool = True,
    link_type: Literal["copy", "symlink"] = "copy",
    client_config: dict | None = None,
) -> None

Export file to new location.

Source code in datachain/lib/file.py

def export(
    self,
    output: str | os.PathLike[str],
    placement: ExportPlacement = "fullpath",
    use_cache: bool = True,
    link_type: Literal["copy", "symlink"] = "copy",
    client_config: dict | None = None,
) -> None:
    """Export file to new location."""
    self._caching_enabled = use_cache
    dst = self.get_destination_path(output, placement)
    dst_dir = os.path.dirname(dst)
    client: Client = self._catalog.get_client(dst_dir, **(client_config or {}))
    client.fs.makedirs(dst_dir, exist_ok=True)

    if link_type == "symlink":
        try:
            return self._symlink_to(dst)
        except OSError as exc:
            if exc.errno not in (errno.ENOTSUP, errno.EXDEV, errno.ENOSYS):
                raise

    self.save(dst, client_config=client_config)

get_destination_path

get_destination_path(
    output: str | PathLike[str], placement: ExportPlacement
) -> str

Returns full destination path of a file for exporting to some output based on export placement

Source code in datachain/lib/file.py

def get_destination_path(
    self, output: str | os.PathLike[str], placement: ExportPlacement
) -> str:
    """
    Returns full destination path of a file for exporting to some output
    based on export placement
    """
    if placement == "filename":
        path = unquote(self.name)
    elif placement == "etag":
        path = f"{self.etag}{self.get_file_suffix()}"
    elif placement == "fullpath":
        path = unquote(self.get_path_normalized())
        source = urlparse(self.source)
        if source.scheme and source.scheme != "file":
            path = posixpath.join(source.netloc, path)
    elif placement == "filepath":
        path = unquote(self.get_path_normalized())
    elif placement == "checksum":
        raise NotImplementedError("Checksum placement not implemented yet")
    else:
        raise ValueError(f"Unsupported file export placement: {placement}")
    return posixpath.join(output, path)  # type: ignore[union-attr]

get_file_ext

get_file_ext()

Returns last part of file name without ..

Source code in datachain/lib/file.py

def get_file_ext(self):
    """Returns last part of file name without `.`."""
    return PurePosixPath(self.path).suffix.lstrip(".")

get_file_stem

get_file_stem()

Returns file name without extension.

Source code in datachain/lib/file.py

def get_file_stem(self):
    """Returns file name without extension."""
    return PurePosixPath(self.path).stem

get_file_suffix

get_file_suffix()

Returns last part of file name with ..

Source code in datachain/lib/file.py

def get_file_suffix(self):
    """Returns last part of file name with `.`."""
    return PurePosixPath(self.path).suffix

get_fs

get_fs()

Returns fsspec filesystem for the file.

Source code in datachain/lib/file.py

def get_fs(self):
    """Returns `fsspec` filesystem for the file."""
    return self._catalog.get_client(self.source).fs

get_fs_path

get_fs_path() -> str

Returns file path with respect to the filescheme.

If normalize is True, the path is normalized to remove any redundant separators and up-level references.

If the file scheme is "file", the path is converted to a local file path using url2pathname. Otherwise, the original path with scheme is returned.

Source code in datachain/lib/file.py

def get_fs_path(self) -> str:
    """
    Returns file path with respect to the filescheme.

    If `normalize` is True, the path is normalized to remove any redundant
    separators and up-level references.

    If the file scheme is "file", the path is converted to a local file path
    using `url2pathname`. Otherwise, the original path with scheme is returned.
    """
    path = unquote(self.get_uri())
    path_parsed = urlparse(path)
    if path_parsed.scheme == "file":
        path = url2pathname(path_parsed.path)
    return path

get_full_name

get_full_name()

[DEPRECATED] Use file.path directly instead.

Returns name with parent directories.

Source code in datachain/lib/file.py

def get_full_name(self):
    """
    [DEPRECATED] Use `file.path` directly instead.

    Returns name with parent directories.
    """
    warnings.warn(
        "file.get_full_name() is deprecated and will be removed "
        "in a future version. Use `file.path` directly.",
        DeprecationWarning,
        stacklevel=2,
    )
    return self.path

get_local_path

get_local_path() -> str | None

Return path to a file in a local cache.

Returns None if file is not cached. Raises an exception if cache is not setup.

Source code in datachain/lib/file.py

def get_local_path(self) -> str | None:
    """Return path to a file in a local cache.

    Returns None if file is not cached.
    Raises an exception if cache is not setup.
    """
    if self._catalog is None:
        raise RuntimeError(
            "cannot resolve local file path because catalog is not setup"
        )
    return self._catalog.cache.get_path(self)

get_uri

get_uri() -> str

Returns file URI.

Source code in datachain/lib/file.py

def get_uri(self) -> str:
    """Returns file URI."""
    return f"{self.source}/{self.get_path_normalized()}"

open

open(
    mode: str = "rb",
    *,
    client_config: dict[str, Any] | None = None,
    **open_kwargs
) -> Iterator[Any]

Open the file and return a file-like object.

Supports both read ("rb", "r") and write modes (e.g. "wb", "w", "ab"). When opened in a write mode, metadata is refreshed after closing.

Source code in datachain/lib/file.py

@contextmanager
def open(
    self,
    mode: str = "rb",
    *,
    client_config: dict[str, Any] | None = None,
    **open_kwargs,
) -> Iterator[Any]:
    """Open the file and return a file-like object.

    Supports both read ("rb", "r") and write modes (e.g. "wb", "w", "ab").
    When opened in a write mode, metadata is refreshed after closing.
    """
    writing = any(ch in mode for ch in "wax+")
    if self.location and writing:
        raise VFileError(
            "Writing to virtual file is not supported",
            self.source,
            self.path,
        )

    if self._catalog is None:
        raise RuntimeError("Cannot open file: catalog is not set")

    base_cfg = getattr(self._catalog, "client_config", {}) or {}
    merged_cfg = {**base_cfg, **(client_config or {})}
    client: Client = self._catalog.get_client(self.source, **merged_cfg)

    if not writing:
        if self.location:
            with VFileRegistry.open(self, self.location) as f:  # type: ignore[arg-type]
                yield self._wrap_text(f, mode, open_kwargs)
            return
        if self._caching_enabled:
            self.ensure_cached()
        with client.open_object(
            self, use_cache=self._caching_enabled, cb=self._download_cb
        ) as f:
            yield self._wrap_text(f, mode, open_kwargs)
        return

    # write path
    full_path = client.get_full_path(self.get_path_normalized())
    with client.fs.open(full_path, mode, **open_kwargs) as f:
        yield self._wrap_text(f, mode, open_kwargs)

    version_hint = self._extract_write_version(f)

    # refresh metadata pinned to the version that was just written
    refreshed = client.get_file_info(
        self.get_path_normalized(), version_id=version_hint
    )
    for k, v in refreshed.model_dump().items():
        setattr(self, k, v)

read

read(length: int = -1)

Returns file contents.

Source code in datachain/lib/file.py

def read(self, length: int = -1):
    """Returns file contents."""
    return self.read_bytes(length)

read_bytes

read_bytes(length: int = -1)

Returns file contents as bytes.

Source code in datachain/lib/file.py

def read_bytes(self, length: int = -1):
    """Returns file contents as bytes."""
    with self.open() as stream:
        return stream.read(length)

read_text

read_text(**open_kwargs)

Return file contents decoded as text.

**open_kwargs : Any Extra keyword arguments forwarded to open(mode="r", ...) (e.g. encoding="utf-8", errors="ignore")

Source code in datachain/lib/file.py

def read_text(self, **open_kwargs):
    """Return file contents decoded as text.

    **open_kwargs : Any
        Extra keyword arguments forwarded to ``open(mode="r", ...)``
        (e.g. ``encoding="utf-8"``, ``errors="ignore"``)
    """
    if self.location:
        raise VFileError(
            "Reading text from virtual file is not supported",
            self.source,
            self.path,
        )

    with self.open(mode="r", **open_kwargs) as stream:
        return stream.read()

rebase

rebase(
    old_base: str,
    new_base: str,
    suffix: str = "",
    extension: str = "",
) -> str

Rebase the file's URI from one base directory to another.

Parameters:

• old_base (str) –

  Base directory to remove from the file's URI

• new_base (str) –

  New base directory to prepend

• suffix (str, default: '' ) –

  Optional suffix to add before file extension

• extension (str, default: '' ) –

  Optional new file extension (without dot)

Returns:

• str ( str ) –

  Rebased URI with new base directory

Raises:

• ValueError –

  If old_base is not found in the file's URI

Examples:

>>> file = File(source="s3://bucket", path="data/2025-05-27/file.wav")
>>> file.rebase("s3://bucket/data", "s3://output-bucket/processed",                     extension="mp3")
's3://output-bucket/processed/2025-05-27/file.mp3'

>>> file.rebase("data/audio", "/local/output", suffix="_ch1",
        extension="npy")
'/local/output/file_ch1.npy'

Source code in datachain/lib/file.py

def rebase(
    self,
    old_base: str,
    new_base: str,
    suffix: str = "",
    extension: str = "",
) -> str:
    """
    Rebase the file's URI from one base directory to another.

    Args:
        old_base: Base directory to remove from the file's URI
        new_base: New base directory to prepend
        suffix: Optional suffix to add before file extension
        extension: Optional new file extension (without dot)

    Returns:
        str: Rebased URI with new base directory

    Raises:
        ValueError: If old_base is not found in the file's URI

    Examples:
        >>> file = File(source="s3://bucket", path="data/2025-05-27/file.wav")
        >>> file.rebase("s3://bucket/data", "s3://output-bucket/processed", \
                extension="mp3")
        's3://output-bucket/processed/2025-05-27/file.mp3'

        >>> file.rebase("data/audio", "/local/output", suffix="_ch1",
                extension="npy")
        '/local/output/file_ch1.npy'
    """
    return rebase_path(self.get_uri(), old_base, new_base, suffix, extension)

resolve

resolve() -> Self

Resolve a File object by checking its existence and updating its metadata.

Returns:

• File ( Self ) –

  The resolved File object with updated metadata.

Source code in datachain/lib/file.py

def resolve(self) -> "Self":
    """
    Resolve a File object by checking its existence and updating its metadata.

    Returns:
        File: The resolved File object with updated metadata.
    """
    if self._catalog is None:
        raise RuntimeError("Cannot resolve file: catalog is not set")

    try:
        client = self._catalog.get_client(self.source)
    except NotImplementedError as e:
        raise RuntimeError(
            f"Unsupported protocol for file source: {self.source}"
        ) from e

    try:
        normalized_path = self.get_path_normalized()
        info = client.fs.info(client.get_full_path(normalized_path))
        converted_info = client.info_to_file(info, normalized_path)
        res = type(self)(
            path=self.path,
            source=self.source,
            size=converted_info.size,
            etag=converted_info.etag,
            version=converted_info.version,
            is_latest=converted_info.is_latest,
            last_modified=converted_info.last_modified,
            location=self.location,
        )
        res._set_stream(self._catalog)
        return res
    except FileError as e:
        logger.warning(
            "File error when resolving %s/%s: %s", self.source, self.path, str(e)
        )
    except (FileNotFoundError, PermissionError, OSError) as e:
        logger.warning(
            "File system error when resolving %s/%s: %s",
            self.source,
            self.path,
            str(e),
        )

    res = type(self)(
        path=self.path,
        source=self.source,
        size=0,
        etag="",
        version="",
        is_latest=True,
        last_modified=TIME_ZERO,
        location=self.location,
    )
    res._set_stream(self._catalog)
    return res

save

save(destination: str, client_config: dict | None = None)

Writes it's content to destination

Source code in datachain/lib/file.py

def save(self, destination: str, client_config: dict | None = None):
    """Writes it's content to destination"""
    destination = stringify_path(destination)
    client: Client = self._catalog.get_client(destination, **(client_config or {}))

    if client.PREFIX == "file://" and not destination.startswith(client.PREFIX):
        destination = Path(destination).absolute().as_uri()

    client.upload(self.read(), destination)

FileError

FileError(message: str, source: str, path: str)

Bases: DataChainError

Source code in datachain/lib/file.py

def __init__(self, message: str, source: str, path: str):
    self.message = message
    self.source = source
    self.path = path
    super().__init__(f"Error in file '{source}/{path}': {message}")

TarVFile

Bases: VFile

Virtual file model for files extracted from tar archives.

open classmethod

open(file: File, location: list[dict])

Stream file from tar archive based on location in archive.

Source code in datachain/lib/file.py

@classmethod
def open(cls, file: "File", location: list[dict]):
    """Stream file from tar archive based on location in archive."""
    tar_file = cls.parent(file, location)

    loc = location[0]

    if (offset := loc.get("offset", None)) is None:
        raise VFileError("'offset' is not specified", file.source, file.path)

    if (size := loc.get("size", None)) is None:
        raise VFileError("'size' is not specified", file.source, file.path)

    client = file._catalog.get_client(tar_file.source)
    fd = client.open_object(tar_file, use_cache=file._caching_enabled)
    return FileSlice(fd, offset, size, file.name)