fs_irods

Documentation about fs_irods.

Submodules

Classes

iRODSFS

iRODS PyFilesystem2 implementation.

Functions

can_create(→ bool)

Check if the mode implies creating a file.

Package Contents

class fs_irods.iRODSFS(session: irods.session.iRODSSession, root: str | None = None)[source]

Bases: fs.base.FS

iRODS PyFilesystem2 implementation.

The filesystem needs to be connected to an iRODS server via a sesseion, using valid login information.

_meta

_lock

_host

_port

_zone

_session

_finalizing = False

files

_root

wrap(path: str) → str[source]

Transform path into iRODSPath.

Parameters:: path (str) – Path to transform.
Returns:: Equivalent iRODSPath.
Return type:: str

parent(path: str) → str[source]

Get the parent directory of specified path.

Parameters:: path (str) – Path to retrieve parent for.
Returns:: Parent directory of the path.
Return type:: str

getinfo(path: str, namespaces: list | None = None) → fs.info.Info[source]

Get information about a resource on the filesystem.

Parameters:

path (str) – A path to a resource on the filesystem.
namespaces (list, optional) – Info namespaces to query. If namespaces is None, then all available namespaces are queried. Defaults to None.

Returns:

An Info object containing information about the: resource.

Return type:

Info

Raises:

ResourceNotFound – If the path does not exist.

listdir(path: str) → list[source]

List a directory on the filesystem.

Parameters:

path (str) – A path to a directory on the filesystem.

Returns:

A list of resources in the directory.

Return type:

list

Raises:

ResourceNotFound – If the path does not exist.
DirectoryExpected – If the path is not a directory.

makedir(path: str, permissions: fs.permissions.Permissions | None = None, recreate: bool = False)[source]

Make a directory on the filesystem.

Parameters:

path (str) – A path to a directory on the filesystem.
permissions (Permissions, optional) – A Permissions instance, or None to use default permissions. Defaults to None.
recreate (bool, optional) – If False (the default) raise an error if the directory already exists, if True do not raise an error. Defaults to False.

Raises:

DirectoryExists – If the directory already exists and recreate is False.
ResourceNotFound – If the path does not exist.

_finalize_files()[source]: Helper function to close file handles.

__del__()[source]: Deconstructor to close all dangling file handles.

open(path: str, mode: str = 'r', buffering: int = -1, encoding: str | None = None, errors: str | None = None, newline: str = '', **options)[source]

Open a file.

Stores weak references to open file handles that maintain a hard reference to the iRODSFS object. In this way, the iRODSFS can only be destructed once these file handles are gone.

Parameters:

path (str) – A path to a file on the filesystem.
mode (str) – Mode to open the file object with (defaults to r).
buffering (int) – Buffering policy (-1 to use default buffering, 0 to disable buffering, 1 to select line buffering, of any positive integer to indicate a buffer size).
encoding (str) – Encoding for text files (defaults to utf-8)
errors (str, optional) – What to do with unicode decode errors (see codecs module for more information).
newline (str) – Newline parameter.
**options – keyword arguments for any additional information required by the filesystem (if any).

Returns:

a file-like object.

Return type:

io.IOBase

Raises:

fs.errors.FileExpected – If the path is not a file.
fs.errors.FileExists – If the file exists, and exclusive mode is specified (x in the mode).
fs.errors.ResourceNotFound – If the path does not exist.

openbin(path: str, mode: str = 'r', buffering: int = -1, **options) → io.BufferedRandom[source]

Open a binary file-like object on the filesystem.

Parameters:

path (str) – A path to a file on the filesystem.
mode (str, optional) – The mode to open the file in, see the built-in open() function for details. Defaults to “r”.
buffering (int, optional) – The buffer size to use for the file, see the built-in open() function for details. Defaults to -1.
**options – Additional options to pass to the open() function.

Returns:

A file-like object representing the file.

Return type:

IO

Raises:

ResourceNotFound – If the path does not exist and mode does not imply creating the file, or if any ancestor of path does not exist.
FileExpected – If the path is not a file.
FileExists – If the path exists, and exclusive mode is specified (x in the mode).

remove(path: str)[source]

Remove a file from the filesystem.

Parameters:

path (str) – A path to a file on the filesystem.

Raises:

ResourceNotFound – If the path does not exist.
FileExpected – If the path is not a file.

_check_isfile(path: str)[source]

Check if a path points to a file and raise an FileExpected error if not.

Parameters:

path (str) – A path to a file on the filesystem.

Raises:

ResourceNotFound – If the path does not exist.
FileExpected – If the path is not a file.

removedir(path: str)[source]

Remove a directory from the filesystem.

Parameters:

path (str) – A path to a directory on the filesystem.

Raises:

ResourceNotFound – If the path does not exist.
DirectoryExpected – If the path is not a directory.
RemoveRootError – If the path is the root directory.
DirectoryNotEmpty – If the directory is not empty.

_is_root(path: str) → bool[source]

Check if path points to root of the filesystem.

Parameters:: path (str) – Path to a directory.
Returns:: True if path points to root.
Return type:: bool

removetree(path: str)[source]

Recursively remove a directory and all its contents.

This method is similar to removedir, but will remove the contents of the directory if it is not empty.

Parameters:

path (str) – A path to a directory on the filesystem.

Raises:

ResourceNotFound – If the path does not exist.
DirectoryExpected – If the path is not a directory.

_check_isdir(path: str)[source]

Check if a path is a directory.

Parameters:

path (str) – A path to a resource on the filesystem.

Raises:

ResourceNotFound – If the path does not exist.
DirectoryExpected – If the path is not a directory.

setinfo(path: str, info: dict) → None[source]

Set metadata for a file or directory.

Parameters:

path (str) – Path to a file or directory on the filesystem.
info (dict) – Dictionary with metadata. Format:
{"details" – {“modified”: <int>, “created”: <int>, “expiry”: <str>, “comments”: <str>}}

Raises:

ResourceNotFound – If the path does not exist.
ValueError – If field values are invalid.

_validate_and_format_timestamp(value, field_name: str) → int[source]

Validate that value can be parsed as a non-negative int timestamp.

Parameters:

value (dict) – The value to validate and format.
field_name (str) – The name of the field being validated (for error messages).

Returns:

The integer timestamp.

Return type:

int

Raises:

ValueError – If value is not an integer or is negative.

_check_exists(path: str)[source]

Check if a resource exists.

Parameters:: path (str) – A path to a resource on the filesystem.
Raises:: ResourceNotFound – If the path does not exist.

isfile(path: str) → bool[source]

Check if a path is a file.

Parameters:: path (str) – A path to a resource on the filesystem.
Returns:: True if the path is a file, False otherwise.
Return type:: bool

isdir(path: str) → bool[source]

Check if a path is a directory.

Parameters:: path (str) – A path to a resource on the filesystem.
Returns:: True if the path is a directory, False otherwise.
Return type:: bool

create(path: str)[source]

Create a file on the filesystem.

Parameters:

path (str) – A path to a file on the filesystem.

Raises:

ResourceNotFound – If any ancestor of path does not exist.
FileExists – If the path exists.

_check_points_into_collection(path: str)[source]

Check if a path points to a location inside a collection.

Parameters:: path (str) – Path to check.
Raises:: ResourceNotFound – If the path does not point to a location inside a collection.

points_into_collection(path: str) → bool[source]

Return true if the path is located inside a collection, aka the parent is a collection.

Parameters:: path (str) – Path to check
Returns:: True if the parent of path is a collection.
Return type:: bool

exists(path: str) → bool[source]

Check if a resource exists.

Parameters:: path (str) – A path to a resource on the filesystem.
Returns:: True if the path exists, False otherwise.
Return type:: bool

move(src_path: str, dst_path: str, overwrite: bool = False, preserve_time: bool = False) → None[source]

Move a file to the specified location.

Parameters:

src_path (str) – Path to the current location of the file
dst_path (str) – Path to the target location of the file
overwrite (bool, optional) – Set to True to overwrite an existing destination file. Defaults to False.
preserve_time (bool, optional) – Set to True to preserve the original modification time. Defaults to False.

Raises:

ResourceNotFound – If the source path does not exist, or if a parent directory of dst_path does not exist.
FileExpected – If the source path is not a file.
DestinationExists – If destination path exists and overwrite is False.

_preserve_modified_time(src_path: str, dst_path: str) → None[source]

Copy the modified time field from src to dst if present.

Parameters:

src_path (str) – Source path to copy modified time from
dst_path (str) – Destination path to copy modified time to

movedir(src_path: str, dst_path: str, overwrite: bool = False, preserve_time: bool = False) → None[source]

Move a directory to the specified location.

Parameters:

src_path (str) – Path to the current location of the directory
dst_path (str) – Path to the target location of the directory
overwrite (bool, optional) – Set to True to overwrite an existing destination directory. Defaults to False.
preserve_time (bool, optional) – Set to True to preserve the original modification time. Defaults to False.

Raises:

ResourceNotFound – If the source path does not exist.
DirectoryExpected – If the source path is not a directory.
DestinationExists – If destination path exists and overwrite is False.

_init_directory_tree_metadata(src_path: str) → dict[source]

Recursively collect modification time metadata for all files and directories in a directory tree.

Parameters:: src_path (str) – Root directory to collect metadata from
Returns:: Dictionary mapping relative paths to their modification times
Return type:: dict

_collect_entry_metadata(path: str, rel: str, entry) → dict[source]

Collect metadata for a file or directory entry.

Parameters:

path (str) – Current path during traversal
rel (str) – Path relative to source root
entry – Entry object (file or directory)

Returns:

Dictionary containing the entry’s metadata, or empty dict if no modified time

Return type:

dict

_apply_directory_tree_metadata(dst_path: str, metadata: dict) → None[source]

Apply collected metadata to a directory tree at the destination.

Parameters:

dst_path (str) – Root destination path where metadata should be applied
metadata (dict) – Dictionary mapping relative paths to their modification times

copy(src_path: str, dst_path: str, overwrite: bool = False, preserve_time: bool = False)[source]

Copy a file from one position to another.

Parameters:

src_path (str) – Path to source file to copy
dst_path (str) – Destination
overwrite (bool, optional) – Whether to overwrite if the destination exists. Defaults to False.
preserve_time (bool, optional) – Whether to preserve the original modification time. Defaults to False.

Raises:

DestinationExists – If dst_path exists and overwrite is False.
ResourceNotFound – If a parent directory of dst_path does not exist.
FileExpected – If src_path is not a file.

copydir(src_path: str, dst_path: str, create: bool = False, preserve_time: bool = False)[source]

Copy the contents of the folder src_path to dst_path.

Parameters:

src_path (str) – Source directory to copy.
dst_path (str) – Where to copy the folder to.
create (bool, optional) – Create the target directory if it does not exist. Defaults to False.
preserve_time (bool, optional) – Preserve the modification time. Defaults to False.

Raises:

ResourceNotFound – If the dst_path does not exist, and create is not True.
DirectoryExpected – If src_path is not a directory.

upload(path: str, file: io.IOBase | str, chunk_size: int | None = None, **options)[source]

Set a file to the contents of a binary file object.

This method copies bytes from an open binary file to a file on the filesystem. If the destination exists, it will first be truncated.

Parameters:

path (str) – A path on the filesystem.
file (io.IOBase or str) – a file object open for reading in binary mode or a path to a local file to upload.
chunk_size (int, optional) – Number of bytes to read at a time, if a simple copy is used, or None to use sensible default.
**options – Implementation specific options required to open the source file.

Raises:

ResourceNotFound – If a parent directory of path does not exist.

Note that the file object file will not be closed by this method. Take care to close it after this method completes (ideally with a context manager).

Example

>>> with open('~/movies/starwars.mov', 'rb') as read_file:
...     my_fs.upload('starwars.mov', read_file)

download(path: str, file: io.IOBase | str, chunk_size=None, **options)[source]

Copy a file from the filesystem to a file-like object.

This may be more efficient that opening and copying files manually if the filesystem supplies an optimized method.

Note that the file object file will not be closed by this method. Take care to close it after this method completes (ideally with a context manager).

Parameters:

path (str) – Path to a resource.
file (file-like) – A file-like object open for writing in binary mode.
chunk_size (int, optional) – Number of bytes to read at a time, if a simple copy is used, or None to use sensible default.
**options – Implementation specific options required to open the source file.

Example

>>> with open('starwars.mov', 'wb') as write_file:
...     my_fs.download('/Videos/starwars.mov', write_file)

Raises:: ResourceNotFound – if path does not exist.

fs_irods.can_create(mode: str) → bool[source]: Check if the mode implies creating a file.