fs_ops

This package contains classes and related functionality for working with file and directories, on different underlying file systems. The main idea is to standardize the interface for interacting with file systems, regardless of if it is local, s3, gcs, or sftp etc. Right now we only support local, gcs, and sftp, but s3 soon.

Path Objects

Top level classes that we interact with.

pydantic model FilePath

general file path, can have any file extension.

field class_id: _ClassIdStr = None: this is the field we dispatch the different sub-classes on. Concrete classes will constrain this to be only one specific value.

field file_system: _file_systems.AnyFileSystemT = None

the file system object for this file. if not specified, we’ll infer it from the path.

Constraints:

discriminator = class_id

field path: _path_strs.AbsoluteFilePathStr [Required]: the path string on the underlying file system. Must be absolute. You can use the file system-specific format, e.g. ‘gs://bucket/file.txt’ and we will convert it to /bucket/file.txt’ and make sure the file_system aligns

as_str() → str: return a contextualized str representation of the path, meaning the file system-specific prefix is used.

classmethod build_dispatched_ann() → Type[SelfTV]: build a type annotation that is equivalent to Union[all concrete subclasses of this class] using class_id as a discriminator. This can be used as an annotation for a field in another pydantic class.

classmethod build_dispatcher_type_adapter() → TypeAdapter[SelfTV]: use the type annotation created by build_dispatched_ann in a pydantic TypeAdapter that allows us to get the “discriminated union” parsing functionality outside a pydantic class context. see pydantic type adapter docs

build_like(path: t.Union[str, _path_strs.AbsoluteFilePathStr, _path_strs.AbsoluteDirPathStr]) → FileOrDirPathT: build a new file or dir path instance using the specified file path and the file system tied to this instance

delete(if_non_existent: Literal['raise', 'return'] = 'return') → None

download_locally(dest_dir: AbsoluteDirPathStr) → AbsoluteFilePathStr: download the file to the local dir specified

ensure_correct_file_ext(choices: list[str]) → None

ensure the file path has one of the specified extensions :param choices: the valid extensions

Returns: Nothing

Raises:: FilePathInvalidException – if the file path has no extension of it isn’t one of the choices

ensure_correct_mime_type(choices: List[MimeType]) → None

ensure the file path is one of the specified mime types :param choices: the valid mime types

Returns: Nothing

Raises:: InvalidFileMimeTypeException – if we cannot determine the mime type, or it is not one of the specified choices

ensure_exists() → None: raise error if the path doesn’t exist

ensure_nonexistent() → None: raise an error if the path does exist

exists() → bool: returns True if this file system object exists, false otherwise

classmethod get_all_concrete_subclass_ids() → List[ClassId]

classmethod get_class_id() → ClassId

classmethod get_class_id_path() → Tuple[ClassId, ...]

classmethod get_class_type() → ClassType

classmethod get_cls_name() → str

get_file_ext() → FileExt | None: get the file extension from the file path if there is one

get_mime_type() → MimeType | None: get the mime type of the file from its file extension, if we have it

get_name(include_suffixes: bool = False) → str: get the name of the file, optionally stripping the name of the suffixes

get_parent() → DirPath: return the dir path instance of the parent of the current dir or file path instance

classmethod init_from_str(raw_str: str) → SelfTV

parse a string into a FilePath instance, inferring the file system from the prefix of the string path.

If the str starts with a ‘$’ we will parse it as a DeferredFilePath instance and evaluate it right way into a FilePath inst.

classmethod is_concrete() → bool

classmethod iter_concrete_subclasses() → Iterator[Type[SelfTV]]

json_dict(use_str_fallback: bool = True, **pydantic_kwargs) → dict

Converts a pydantic instance to a dict, going through json first. This is a convenience function for when you want a dict, but want to use the json encoders connected to the pydantic object.

Parameters:

use_str_fallback – if true, we cast anything we cannot encode to a string
**pydantic_kwargs – any kwargs available for pydantic’s json method. see their docs

Returns:

a natively json-encodeable dict

Return type:

dict

Examples

simple class with custom date encoder:

class Data(PydanticBase):
    class Config:
        json_encoders = {
            dt.date: lambda date_obj: date_obj.strftime("%d/%m/%Y")
        }
    x: int
    date: dt.date

inst = Data(x=1, date=dt.date(year=2023, month=1, day=1))

# regular dict
inst.dict() == {"x": 1, "date": dt.date(year=2023, month=1, day=1)}

# this method
inst.json_dict() == {"x": 1, "date": "01/01/2023"}
# note that the date is the json encoding tied to the encoder for this pydantic
# class. I just used a familiar string format, it could be whatever is specified

open(mode: Literal['ab', 'wb', 'rb'] = 'rb') → AbstractBufferedFile | TextIOWrapper

read_as_str(encoding: str = 'utf-8') → str: convenience method to read str data from a file

special_repr(verbose: bool = False) → str

write(data: str | bytes, mode: Literal['wb', 'ab'] = 'wb') → None: convenience method to write bytes or str data to a file

property file_system_type: ClassId: get the class id of the file system tied to this path object

pydantic model DirPath

this class represents a directory on a file system, with the absolute full path string, and the file system instance

field class_id: _ClassIdStr = None: this is the field we dispatch the different sub-classes on. Concrete classes will constrain this to be only one specific value.

field file_system: _file_systems.AnyFileSystemT = None

the file system object for this file. if not specified, we’ll infer it from the path.

Constraints:

discriminator = class_id

field path: _path_strs.AbsoluteDirPathStr [Required]: the path string on the underlying file system. Must be absolute. You can use the file system-specific format, e.g. ‘gs://bucket/dir/’ and we will convert it to /bucket/dir/’ and make sure the file_system aligns

as_str() → str: return a contextualized str representation of the path, meaning the file system-specific prefix is used.

classmethod build_dispatched_ann() → Type[SelfTV]: build a type annotation that is equivalent to Union[all concrete subclasses of this class] using class_id as a discriminator. This can be used as an annotation for a field in another pydantic class.

classmethod build_dispatcher_type_adapter() → TypeAdapter[SelfTV]: use the type annotation created by build_dispatched_ann in a pydantic TypeAdapter that allows us to get the “discriminated union” parsing functionality outside a pydantic class context. see pydantic type adapter docs

build_like(path: t.Union[str, _path_strs.AbsoluteFilePathStr, _path_strs.AbsoluteDirPathStr]) → FileOrDirPathT: build a new file or dir path instance using the specified file path and the file system tied to this instance

delete(if_non_existent: Literal['raise', 'return'] = 'return', recursive: bool = True) → None: delete the directory. recursively deleting all contents if recursive is true.

download_locally(dest_dir: AbsoluteDirPathStr) → AbsoluteDirPathStr

ensure_exists() → None: raise error if the path doesn’t exist

ensure_nonexistent() → None: raise an error if the path does exist

exists() → bool: returns True if this file system object exists, false otherwise

classmethod get_all_concrete_subclass_ids() → List[ClassId]

classmethod get_class_id() → ClassId

classmethod get_class_id_path() → Tuple[ClassId, ...]

classmethod get_class_type() → ClassType

classmethod get_cls_name() → str

get_name() → str: get simple name, i.e. the most terminal chunk in the path

get_parent() → DirPath: return the dir path instance of the parent of the current dir or file path instance

classmethod init_from_str(raw_str: str) → DirPath

classmethod is_concrete() → bool

classmethod iter_concrete_subclasses() → Iterator[Type[SelfTV]]

iter_dir() → t.Iterator[t.Union[FilePath, DirPath]]

iter_dir_contents_files_only(recursive: bool = True) → list[FilePath]

joinpath(relative_path: RelFilePathStr) → FilePath
joinpath(relative_path: RelDirPathStr) → DirPath
joinpath(relative_path: str) → t.Union[FilePath, DirPath]: get a new FilePath or DirPath by combining this DirPath instance with a provided relative path.

json_dict(use_str_fallback: bool = True, **pydantic_kwargs) → dict

Converts a pydantic instance to a dict, going through json first. This is a convenience function for when you want a dict, but want to use the json encoders connected to the pydantic object.

Parameters:

use_str_fallback – if true, we cast anything we cannot encode to a string
**pydantic_kwargs – any kwargs available for pydantic’s json method. see their docs

Returns:

a natively json-encodeable dict

Return type:

dict

Examples

simple class with custom date encoder:

class Data(PydanticBase):
    class Config:
        json_encoders = {
            dt.date: lambda date_obj: date_obj.strftime("%d/%m/%Y")
        }
    x: int
    date: dt.date

inst = Data(x=1, date=dt.date(year=2023, month=1, day=1))

# regular dict
inst.dict() == {"x": 1, "date": dt.date(year=2023, month=1, day=1)}

# this method
inst.json_dict() == {"x": 1, "date": "01/01/2023"}
# note that the date is the json encoding tied to the encoder for this pydantic
# class. I just used a familiar string format, it could be whatever is specified

make_self(if_exists: Literal['raise', 'return'] = 'return') → None: equivalent of mkdir

special_repr(verbose: bool = False) → str

property file_system_type: ClassId: get the class id of the file system tied to this path object

FileOrDirPathT: alias of FilePath | TextFilePath | JsonFilePath | YamlFilePath | DirPath[FilePath | TextFilePath | JsonFilePath | YamlFilePath | DirPath]

FileOrDirPathTypeAdapter: alias of <pydantic.type_adapter.TypeAdapter object>

pydantic model DeferredFilePath

a file path, where part of the path is contained in an environment variable, that is looked up when we call ‘evaluate’ method to convert this instance to a FilePath instance.

field class_id: _ClassIdStr = None: this is the field we dispatch the different sub-classes on. Concrete classes will constrain this to be only one specific value.

field env_var_key: str [Required]

field rel_path: _path_strs.RelFilePathStr [Required]

classmethod build_dispatched_ann() → Type[SelfTV]: build a type annotation that is equivalent to Union[all concrete subclasses of this class] using class_id as a discriminator. This can be used as an annotation for a field in another pydantic class.

classmethod build_dispatcher_type_adapter() → TypeAdapter[SelfTV]: use the type annotation created by build_dispatched_ann in a pydantic TypeAdapter that allows us to get the “discriminated union” parsing functionality outside a pydantic class context. see pydantic type adapter docs

evaluate() → FilePath

classmethod get_all_concrete_subclass_ids() → List[ClassId]

classmethod get_class_id() → ClassId

classmethod get_class_id_path() → Tuple[ClassId, ...]

classmethod get_class_type() → ClassType

classmethod get_cls_name() → str

classmethod is_concrete() → bool

classmethod iter_concrete_subclasses() → Iterator[Type[SelfTV]]

json_dict(use_str_fallback: bool = True, **pydantic_kwargs) → dict

Converts a pydantic instance to a dict, going through json first. This is a convenience function for when you want a dict, but want to use the json encoders connected to the pydantic object.

Parameters:

use_str_fallback – if true, we cast anything we cannot encode to a string
**pydantic_kwargs – any kwargs available for pydantic’s json method. see their docs

Returns:

a natively json-encodeable dict

Return type:

dict

Examples

simple class with custom date encoder:

class Data(PydanticBase):
    class Config:
        json_encoders = {
            dt.date: lambda date_obj: date_obj.strftime("%d/%m/%Y")
        }
    x: int
    date: dt.date

inst = Data(x=1, date=dt.date(year=2023, month=1, day=1))

# regular dict
inst.dict() == {"x": 1, "date": dt.date(year=2023, month=1, day=1)}

# this method
inst.json_dict() == {"x": 1, "date": "01/01/2023"}
# note that the date is the json encoding tied to the encoder for this pydantic
# class. I just used a familiar string format, it could be whatever is specified

pydantic model DeferredDirPath

a dir path, where part or all of the path is contained in an environment variable, that is looked up when we call ‘evaluate’ method to convert this instance to a DirPath instance.

field class_id: _ClassIdStr = None: this is the field we dispatch the different sub-classes on. Concrete classes will constrain this to be only one specific value.

field env_var_key: str [Required]

field rel_path: t.Optional[_path_strs.RelDirPathStr] [Required]

classmethod build_dispatched_ann() → Type[SelfTV]: build a type annotation that is equivalent to Union[all concrete subclasses of this class] using class_id as a discriminator. This can be used as an annotation for a field in another pydantic class.

classmethod build_dispatcher_type_adapter() → TypeAdapter[SelfTV]: use the type annotation created by build_dispatched_ann in a pydantic TypeAdapter that allows us to get the “discriminated union” parsing functionality outside a pydantic class context. see pydantic type adapter docs

evaluate() → DirPath

classmethod get_all_concrete_subclass_ids() → List[ClassId]

classmethod get_class_id() → ClassId

classmethod get_class_id_path() → Tuple[ClassId, ...]

classmethod get_class_type() → ClassType

classmethod get_cls_name() → str

classmethod is_concrete() → bool

classmethod iter_concrete_subclasses() → Iterator[Type[SelfTV]]

json_dict(use_str_fallback: bool = True, **pydantic_kwargs) → dict

Converts a pydantic instance to a dict, going through json first. This is a convenience function for when you want a dict, but want to use the json encoders connected to the pydantic object.

Parameters:

use_str_fallback – if true, we cast anything we cannot encode to a string
**pydantic_kwargs – any kwargs available for pydantic’s json method. see their docs

Returns:

a natively json-encodeable dict

Return type:

dict

Examples

simple class with custom date encoder:

class Data(PydanticBase):
    class Config:
        json_encoders = {
            dt.date: lambda date_obj: date_obj.strftime("%d/%m/%Y")
        }
    x: int
    date: dt.date

inst = Data(x=1, date=dt.date(year=2023, month=1, day=1))

# regular dict
inst.dict() == {"x": 1, "date": dt.date(year=2023, month=1, day=1)}

# this method
inst.json_dict() == {"x": 1, "date": "01/01/2023"}
# note that the date is the json encoding tied to the encoder for this pydantic
# class. I just used a familiar string format, it could be whatever is specified

FileOrDirDeferredPathT: alias of DeferredFilePath | DeferredDirPath[DeferredFilePath | DeferredDirPath]

FileOrDirDeferredPathTypeAdapter: alias of <pydantic.type_adapter.TypeAdapter object>

File Systems

The file system-specific implementations of the interface. Instances of these classes live on FilePath and DirPath instances.

pydantic model AbstractFileSystem

base class for all fsspec file system wrappers

field class_id: _ClassIdStr = None: this is the field we dispatch the different sub-classes on. Concrete classes will constrain this to be only one specific value.

classmethod build_dispatched_ann() → Type[SelfTV]: build a type annotation that is equivalent to Union[all concrete subclasses of this class] using class_id as a discriminator. This can be used as an annotation for a field in another pydantic class.

classmethod build_dispatcher_type_adapter() → TypeAdapter[SelfTV]: use the type annotation created by build_dispatched_ann in a pydantic TypeAdapter that allows us to get the “discriminated union” parsing functionality outside a pydantic class context. see pydantic type adapter docs

abstract build_fsspec_file_system() → AbstractFileSystem

classmethod clean_raw_path_str(_AbstractFileSystem__raw_path: str) → AbsoluteFilePathStr | AbsoluteDirPathStr

strip the raw path of the file system-specific prefix, and cast it to a standardized directory or file path representation.

This is the inverse of ‘contextualize_abs_path’.

contextualize_abs_path(_AbstractFileSystem__path: AbsoluteFilePathStr | AbsoluteDirPathStr) → str: convert a standardized file or directory path representation into a str starting with the file system-specific prefix. This is the inverse of ‘clean_raw_path_str’.

delete_dir(_AbstractFileSystem__path: AbsoluteDirPathStr, *, recursive: bool = True, if_non_existent: Literal['raise', 'return'] = 'return') → None: delete the directory, if recursive is true, we delete all files and subdirectories

delete_file(_AbstractFileSystem__path: AbsoluteFilePathStr, *, if_non_existent: Literal['raise', 'return'] = 'return') → None: delete the file

download_dir_locally(_AbstractFileSystem__source_dir: AbsoluteDirPathStr, _AbstractFileSystem__local_dest_dir: AbsoluteDirPathStr) → AbsoluteDirPathStr

download directory from current file system into the local file system.

the downloaded directory will have the same name as the source directory, inside the specified local directory.

download_file_locally(_AbstractFileSystem__source_path: AbsoluteFilePathStr, _AbstractFileSystem__local_dest_dir: AbsoluteDirPathStr) → AbsoluteFilePathStr

download file from current file system into the local file system.

the downloaded file will have the same name as the source file, inside the specified local directory.

ensure_exists(_AbstractFileSystem__path: AbsoluteFilePathStr | AbsoluteDirPathStr) → None

ensure the file or dir path exists, raising an exception if not

Parameters:: __path – the path to the file or directory

Returns: Nothing

Raises:: NonExistentPathException – if the path doesn’t exist, or we cannot access it

ensure_nonexistent(_AbstractFileSystem__path: AbsoluteFilePathStr | AbsoluteDirPathStr) → None

ensure the file or dir path does exist, raising an exception if it does

Parameters:: __path – the path to the file or directory

Returns: Nothing

Raises:: PathAlreadyExistsException – if the path does exist

exists(_AbstractFileSystem__path: AbsoluteFilePathStr | AbsoluteDirPathStr) → bool: returns True if this file system object exists, false otherwise

classmethod get_all_concrete_subclass_ids() → List[ClassId]

classmethod get_class_id() → ClassId

classmethod get_class_id_path() → Tuple[ClassId, ...]

classmethod get_class_type() → ClassType

abstract classmethod get_fs_specific_prefix() → str

abstract get_home_dir() → AbsoluteDirPathStr

classmethod is_concrete() → bool

classmethod iter_concrete_subclasses() → Iterator[Type[SelfTV]]

iter_dir_contents(_AbstractFileSystem__path: AbsoluteDirPathStr) → Iterator[AbsoluteFilePathStr | AbsoluteDirPathStr]: iterate over the top level dir contents

iter_dir_contents_files_only(_AbstractFileSystem__path: AbsoluteDirPathStr, *, recursive: bool = True) → Iterator[AbsoluteFilePathStr]: iterate over every file in a directory, recursing into subdirectories if recursive is True

list_dir_contents(_AbstractFileSystem__path: AbsoluteDirPathStr) → list[AbsoluteFilePathStr | AbsoluteDirPathStr]: exhaust the iterator built from ‘iter_dir_contents’ returning the items in a list

make_dir(_AbstractFileSystem__path: AbsoluteDirPathStr, *, if_exists: Literal['raise', 'return'] = 'return', create_parents: bool = True) → None: create a new directory, and parents if the parent directory doesn’t exist

open_file(_AbstractFileSystem__path: AbsoluteFilePathStr, mode: Literal['ab', 'wb', 'rb']) → AbstractBufferedFile | TextIOWrapper

return a readable open file object. todo: revisit type annotations of the return type here

Parameters:

__path – the file path str
mode – the mode we are opening in

Returns:

the readable open file object

special_repr() → str

property fsspec_obj: AbstractFileSystem

pydantic model LocalFileSystem

file system implementation for the local file system

field class_id: _ClassIdStr = None: this is the field we dispatch the different sub-classes on. Concrete classes will constrain this to be only one specific value.

classmethod build_dispatched_ann() → Type[SelfTV]: build a type annotation that is equivalent to Union[all concrete subclasses of this class] using class_id as a discriminator. This can be used as an annotation for a field in another pydantic class.

classmethod build_dispatcher_type_adapter() → TypeAdapter[SelfTV]: use the type annotation created by build_dispatched_ann in a pydantic TypeAdapter that allows us to get the “discriminated union” parsing functionality outside a pydantic class context. see pydantic type adapter docs

build_fsspec_file_system() → LocalFileSystem

classmethod clean_raw_path_str(_AbstractFileSystem__raw_path: str) → AbsoluteFilePathStr | AbsoluteDirPathStr

strip the raw path of the file system-specific prefix, and cast it to a standardized directory or file path representation.

This is the inverse of ‘contextualize_abs_path’.

contextualize_abs_path(_AbstractFileSystem__path: AbsoluteFilePathStr | AbsoluteDirPathStr) → str: convert a standardized file or directory path representation into a str starting with the file system-specific prefix. This is the inverse of ‘clean_raw_path_str’.

delete_dir(_AbstractFileSystem__path: AbsoluteDirPathStr, *, recursive: bool = True, if_non_existent: Literal['raise', 'return'] = 'return') → None: delete the directory, if recursive is true, we delete all files and subdirectories

delete_file(_AbstractFileSystem__path: AbsoluteFilePathStr, *, if_non_existent: Literal['raise', 'return'] = 'return') → None: delete the file