Modules

Submodules

Jupyter Helper Module

app.jupyter_helper.event_stream(session, url)[source]

Generator yielding events from a JSON event stream.

Parameters:

session – Session object for making requests.
url (str) – URL of the event stream.

Yields:

dict – Event data.

app.jupyter_helper.get_user_status(user_name)[source]

Get detailed user status.

Parameters:: user_name (str) – Name of the user.
Returns:: User status data.
Return type:: dict
Raises:: HTTPException – If the request to fetch user status fails.

app.jupyter_helper.start_user_server(user_name, server_name='')[source]

Start a JupyterHub server and wait for it to be ready.

Parameters:

user_name (str) – Name of the user.
server_name (str, optional) – Name of the server. Defaults to “”.

Returns:

Server data.

Return type:

dict

Raises:

HTTPException – If the request to start the server fails.

app.jupyter_helper.stop_user_server(user_name, server_name='')[source]

Stop a server and wait for it to complete.

Parameters:

user_name (str) – Name of the user.
server_name (str, optional) – Name of the server. Defaults to “”.

Returns:

Server data.

Return type:

dict

Raises:

HTTPException – If the request to stop the server fails.

Rucio Helper Module

app.rucio_helper.check_storage_availability(account: str, rse_expression: str, rucio_token: str)[source]

Verify that data exists at a given scope and name.

Parameters:

scope (str) – Scope of the data.
name (str) – Name of the data.
rucio_token (str) – Rucio token.

Returns:

Data availability information.

Return type:

dict

Raises:

HTTPException – If the data is not found.

app.rucio_helper.get_rucio_headers(rucio_token: str)[source]

Generate headers with user-provided Rucio token.

Parameters:: rucio_token (str) – Rucio token.
Returns:: Headers with Rucio token.
Return type:: dict

app.rucio_helper.initiate_transfer(source_rse: str, dest_rse: str, scope: str, name: str, rucio_token: str)[source]

Initiate data transfer between RSEs.

Parameters:

source_rse (str) – Source RSE.
dest_rse (str) – Destination RSE.
scope (str) – Scope of the data.
name (str) – Name of the data.
rucio_token (str) – Rucio token.

Returns:

Transfer information.

Return type:

dict

Raises:

HTTPException – If the transfer fails.

app.rucio_helper.monitor_transfer(transfer_id: str, rucio_token: str)[source]

Initiate data transfer between RSEs.

Parameters:

source_rse (str) – Source RSE.
dest_rse (str) – Destination RSE.
scope (str) – Scope of the data.
name (str) – Name of the data.
rucio_token (str) – Rucio token.

Returns:

Transfer information.

Return type:

dict

Raises:

HTTPException – If the transfer fails.

app.rucio_helper.verify_data_availability(scope: str, name: str, rucio_token: str)[source]

Check available storage for the user.

Parameters:

account (str) – Account of the user.
rse_expression (str) – RSE expression.
rucio_token (str) – Rucio token.

Returns:

Storage availability information.

Return type:

dict

Raises:

HTTPException – If the storage is insufficient.

Staging Method Module

app.staging_methods.direct_download(local_path)[source]

Serves a file from the local storage using a direct download.

Parameters:

local_path (str) – Path to the file on the local storage.

Returns:

A FileResponse object containing the file contents.

Return type:

FileResponse

Raises:

FileNotFoundError – If the file does not exist at the provided path.
Exception – If any other error occurs during the direct download operation.

app.staging_methods.jupyter_copy(local_path, relative_path, username, token)[source]

Copies a file from the local storage to the user’s Jupyter server.

Parameters:

local_path (str) – Path to the file on the local storage.
relative_path (str) – Path to the file relative to the user area.
username (str) – Username of the user to copy the file for.
token (str) – Token to use for authentication with the Jupyter server.

Raises:

FileNotFoundError – If the file does not exist at the provided path.
Exception – If any other error occurs during the copy operation.

app.staging_methods.local_copy(local_path, relative_path)[source]

Copies a file or directory from the local storage to the user area.

Parameters:

local_path (str) – Path to the file or directory on the local storage.
relative_path (str) – Path to the file or directory relative to the user area.

Raises:

ValueError – If the path provided is invalid.
Exception – If any other error occurs during the copy operation.

app.staging_methods.local_symlink(local_path, relative_path)[source]

Creates a symlink from the local storage to the user area.

Parameters:

local_path (str) – Path to the file or directory on the local storage.
relative_path (str) – Path to the file or directory relative to the user area.

Raises:

ValueError – If the path provided is invalid.
Exception – If any other error occurs during the symlink operation.

Staging Service Module

class app.staging_service.DataItem(*, local_path_on_storage: str, relative_path: str)[source]

Bases: BaseModel

DataItem represents a single data mapping with local and relative paths.

local_path_on_storage

Local path on storage where the file currently resides.

Type:: str

relative_path

Relative path inside the user area where the file should be made available.

Type:: str

local_path_on_storage: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

relative_path: str

class app.staging_service.StagingRequest(*, data: DataItem)[source]

Bases: BaseModel

StagingRequest represents a request to stage data with a single data mapping.

data

Single data mapping with local and relative paths.

Type:: DataItem

Example

{: “local_path_on_storage”: “/mnt/storage_a/data1”, “relative_path”: “project/data1”

}

data: DataItem

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class app.staging_service.SuccessResponse(*, status: str, message: str)[source]

Bases: BaseModel

Successful response model.

status

Status of the response.

Type:: str

message

Message describing the response.

Type:: str

Example

{: “status”: “success”, “message”: “Data staged for user test_user with method local_copy”

}

message: str

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'example': {'message': 'Data staged for user test_user with method local_copy', 'status': 'success'}}}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

status: str

class app.staging_service.ValidationError(*, detail: str)[source]

Bases: BaseModel

Validation error model.

detail

Detailed error message.

Type:: str

Example

{: “detail”: “Invalid request”

}

detail: str

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'example': {'detail': 'Source path does not exist: /mnt/storage_a/data1 or Invalid Staging method'}}}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

async app.staging_service.create_file(filename: str = Body(PydanticUndefined), content: str = Body(PydanticUndefined), site_config: dict = Depends(get_config))[source]

Create a file in the source storage directory.

Parameters:

filename (str) – Name of the file to create.
content (str) – Content of the file.
site_config (dict) – Application configuration.

Returns:

Response.

Return type:

dict

app.staging_service.get_config()[source]

Fetch the application configuration.

Returns:: Application configuration.
Return type:: dict

async app.staging_service.get_logs()[source]

Expose application logs.

Returns:: Application logs.
Return type:: dict

async app.staging_service.get_site_allowed_methods(site_config: dict = Depends(get_config))[source]

Expose only the allowed methods.

Parameters:: site_config (dict) – Application configuration.
Returns:: Allowed methods.
Return type:: dict

async app.staging_service.get_site_config(site_config: dict = Depends(get_config))[source]

Expose the full site configuration.

Parameters:: site_config (dict) – Application configuration.
Returns:: Full site configuration.
Return type:: dict

async app.staging_service.stage_data(request: StagingRequest, method: str = Query(PydanticUndefined), username: str = Query(PydanticUndefined), jupyter_token: str | None = Query(None), site_config: dict = Depends(get_config))[source]

Stage data for a user.

Parameters:

request (StagingRequest) – Request to stage data.
method (str) – Method to use for staging data.
username (str) – Username of the requester.
jupyter_token (Optional[str]) – Optional JupyterHub API token for interaction with Jupyter.
site_config (dict) – Application configuration.

Returns:

Successful response.

Return type:

SuccessResponse

Utility Module

app.utility.ensure_user_exists(username: str)[source]

Ensure that a user with the specified username exists on the system.

Parameters:: username (str) – The username to check and potentially create.
Raises:: HTTPException – If there is an error in checking or creating the user.

app.utility.set_read_only(path, username)[source]

Set the given path (file or directory) to have read-only permissions (0o400 for files, 0o500 for directories) and change the ownership to the given username.

Parameters:

path (str) – The path to the file or directory to set read-only.
username (str) – The username to change the ownership to.

Raises:

HTTPException – If the username does not exist.

Modules

Submodules

Jupyter Helper Module

Rucio Helper Module

Staging Method Module

Staging Service Module

Utility Module

Module contents