XRootDTargets

XRootDTargets#

To ease the work with files stored on the grid, b2luigi provides an Target implementation using XRootD. In the background, this relies on the Python bindings of the XRootD client and of course requires a working XRootD client installation.

Hint

For Belle II users, the XRootD client is already installed in basf2 environments.

Hint

To access XRootD storage you will need a valid VOMS proxy.

To use the targets, you will have to pass the XRootDTarget class and the keyword arguments to the add_to_output function. This requires some additional setup compared to the standard LocalTarget.

First you need to create a FileSystem object, which is used to connect to the XRootD server. For that you need to provide the server address like so: root://<server_address>.

Optionally, one can also set a scratch_dir which will be used to store temporary files when using the temporary_path context. (see the luigi documentation)

When entering this context, a temporary path will be created in the scratch_dir. At leaving the context, the file will then be copied to the final location on the XRootD storage.

A full task using XRootDTargets could look like this:

from b2luigi import XRootDSystem, XrootDTarget
from b2luigi.core.utils import create_output_filename
import b2luigi

class MyTask(b2luigi.Task):
    def run(self):
        file_name = "Hello_world.txt"
        target = self._get_output_file_target(file_name)
        with target.temporary_path() as temp_path:
            with open(temp_path, "w") as f:
                f.write("Hello World")

    def output(self):
        fs = XRootDSystem("root://eospublic.cern.ch")
        yield self.add_to_output("Hello_world.txt", XRootDTarget, file_system=fs)

Another example could be:

import b2luigi

class MyTask(b2luigi.Task):
    @property
    def task_file_dir(self):
        return "/path/on/local"

    @property
    def result_dir(self):
        return "/path/on/server"

    @property
    def xrootdsystem(self):
        return b2luigi.XRootDSystem("root://eospublic.cern.ch")

    @property
    def default_task_target_class(self):
        return b2luigi.XRootDTarget

    @property
    def target_class_kwargs(self):
        return {
            "file_system": self.xrootdsystem,
        }

    def output(self):
        yield self.add_to_output("Hello_world.txt")

    @b2luigi.on_temporary_files
    def run(self):
        with open(self.get_output_file_name("Hello_world.txt"), "w") as f:
            f.write("Hello World")
class b2luigi.XRootDSystem(server_path: str)[source]#

Bases: FileSystem

XRootD file system for b2luigi Targets. Inspiration taken from RHofsaess. It implements some standard file system operations, which can be used by the XRootDTarget. The error handling is done by assertions, since XRootD does not raise exceptions.

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

Implementation of the exists function for the XRootDSystem. Will return True if the file or directory exists and False if it can not be found. This might also include cases, where the server is not reachable.

Parameters:

path (str) – The path to check for existence.

Returns:

True if the path exists, False otherwise..

Return type:

bool

copy_file_to_remote(local_path: str, remote_path: str, force: bool = False) → None[source]#

Function to copy a file from the local file system to the remote file system. In case the copy fails, a warning will be printed and a assertion will fail.

Parameters:

  • local_path (str) – The path to the file on the local file system.

  • remote_path (str) – The destination path for the file on the remote file system.

  • force (bool, optional) – If True, overwrites the file on the remote system if it already exists. Defaults to False.

Raises:

AssertionError – If the file copy operation fails.

copy_file_from_remote(remote_path: str, local_path: str, force: bool = False) → None[source]#

Function to copy a file from the remote file system to the local file system. In case the copy fails, a warning will be printed and a assertion will fail.

Parameters:

  • remote_path – Path to the file on the remote file system.

  • local_path – Path to the file on the local file system.

  • force – If True, the file will be overwritten if it already exists. Default is False.

This method uses the client to perform the file transfer. If the copy operation fails, a warning message is logged, and an assertion error is raised.

Parameters:

  • remote_path (str) – The path to the file on the remote file system.

  • local_path (str) – The destination path for the file on the local file system.

  • force (bool, optional) – If True, overwrites the file on the local system if it already exists. Defaults to False.

Raises:

AssertionError – If the file copy operation fails.

copy_dir_from_remote(remote_path: str, local_path: str, force: bool = False) → None[source]#

Copies a directory and its files from the remote file system to the local file system.

This method does not support nested directories. It iterates through the contents of the remote directory and copies each file to the specified local directory.

Parameters:

  • remote_path (str) – Path to the directory on the remote file system.

  • local_path (str) – Path to the directory on the local file system.

  • force (bool, optional) – If True, overwrites files on the local system if they already exist. Defaults to False.

move(source_path: str, dest_path: str) → None[source]#

A function to move a file from one location to another on the XRootD server. In case the move fails, a warning will be printed and a assertion will fail.

Parameters:

  • source_path (str) – The current path of the file on the remote file system.

  • dest_path (str) – The target path for the file on the remote file system.

Raises:

AssertionError – If the move operation fails.

mkdir(path: str) → None[source]#

A function to create a directory on the remote file system. In case the creation fails, a warning will be printed and a assertion will fail.

Parameters:

path (str) – The path of the directory to create.

Raises:

AssertionError – If the directory creation operation fails.

locate(path: str) → bool[source]#

Checks the location of a file on the remote file system.

This method queries the remote file system to locate the specified file. If the operation fails, a warning is logged, and an assertion error is raised.

Parameters:

path (str) – The path to the file on the remote file system.

Returns:

True if the file location is successfully determined.

Return type:

bool

Raises:

AssertionError – If the locate operation fails.

remove(path: str) → None[source]#

A function to remove a file from the remote file system. This function can not remove directories. Use remove_dir for that. In case the removal fails, a warning will be printed and a assertion will fail.

Parameters:

path – Path to the file on the remote file system.

This method deletes a single file and does not support directory removal.

Parameters:

path (str) – The path to the file on the remote file system.

Raises:

AssertionError – If the file removal operation fails.

listdir(path: str, print_entries: bool = False) → Tuple[Dict[str, int], Any][source]#

Lists the contents of a directory on the remote file system.

This method retrieves the directory listing and categorizes entries as files or directories. If the operation fails, a warning is logged, and an assertion error is raised.

Parameters:

path (str) – Path to the directory on the remote file system.

Returns:

A dictionary mapping paths to their type (1 for directories, 0 for files), and the raw listing object.

Return type:

Tuple[Dict[str, int], Any]

Raises:

AssertionError – If the directory listing operation fails.

remove_dir(path: str) → None[source]#

A function to iteratively remove a directory and all its content from the remote file system. In case the removal fails, a warning will be printed and a assertion will fail.

Parameters:

path (str) – Path to the directory on the remote file system.

Raises:

AssertionError – If the directory removal operation fails.

rename_dont_move(path: str, dest: str) → None[source]#

Overwriting a the luigi function used to handle the atomic write problem. (See spotify/luigi) In this case it is just an alias for copy_file_to_remote with force=True.

Parameters:

  • local_path – Path to the file on the local file system.

  • remote_path – Path to the file on the remote file system.

class b2luigi.XRootDTarget(path: str, file_system: XRootDSystem)[source]#

Bases: FileSystemTarget

Luigi target implementation for the XRootD file system.

property base_name: str#

Get the base name of the target path.

property fs: XRootDSystem#

Get the associated file system.

makedirs() → None[source]#

Create the target’s directory on the remote file system.

move() → None[source]#

Move the target to a new location.

get(path: str = '~') → str[source]#

A function to copy the file from the remote file system to the local file system.

Parameters:

path – Path to copy the file to.

Returns:

Path to the copied file.

open(mode: str) → None[source]#

Raise NotImplementedError as open is not supported.

get_temporary_input(task: Task | None = None) → Generator[str, None, None][source]#

Create a temporary local copy of a remote input file.

Downloads the remote file to a temporary local directory for processing, allowing safe concurrent access to the same remote input file by multiple tasks.

Parameters:

task (Optional[Task]) – Task instance used to determine scratch directory settings. If None, uses default settings.

Yields:

str – Absolute path to the temporary local copy of the remote input file.

Note

  • The temporary file and its parent directory are automatically cleaned up when exiting the context manager.

  • Files are downloaded to the scratch directory specified in settings (defaults to /tmp if not set).

Example

target = XRootDTarget("root://server/path/data.root", fs)
with target.get_temporary_input() as tmp_input:
    process_local_file(tmp_input)
# Temporary file is automatically cleaned up.
Contents