network

Network protocols.

This module implements support for the following network protocols:

• SNS network protocol

  • Authentication

  • Base support for accessing Scanco database records and files

• XF network protocol

  • Authentication

  • Base support for accessing XF database records as JSON

• FSS network protocol

  • File Storage Service access to bulk files

  • Access Tokens

  • Virtualpaths

  • Direct Access

class network.AuthorizationToken(token_base64: str)

An authorization token for the FSS protocol.

class AuthorizationScope(scope_string: str)

The scope of a authorization token.

get_xml_element_text(document: Element, element_name: str) → str

Get the XML element’s text value.

get_xml_optional_element_text(document: Element, element_name: str) → str | None

Get the optional XML element’s text value.

class network.Connection(host: str, port: int | str, auth: str)

A network connection to a SNS or XF server.

check_status(response: HTTPResponse, allow_status: List[int] | None = None) → None

Check HTTP status and raise exception for unexpected errors.

close() → None

Close the connection.

delete(url: str, allow_status: List[int] | None = None) → bytes

Send a HTTP DELETE request and read the response body.

get(url: str, allow_status: List[int] | None = None) → bytes

Send a HTTP GET request and read the response body.

get_error_response_json_message(response: HTTPResponse, response_body: bytes, default_message: str) → str

Try to extract a JSON Message from the error response.

get_json(url: str) → Any

Send a HTTP GET request an parse the JSON response.

head_response(url: str, allow_status: List[int] | None = None) → HTTPResponse

Send a HTTP HEAD request and return the processed (read) HTTP response object.

post(url: str, content_type: str, data: Any, allow_status: List[int] | None = None) → bytes

Send a HTTP POST request and read the response body.

post_json(url: str, json_data: Any) → Any

Send a HTTP JSON POST request an parse the JSON response.

put(url: str, content_type: str, data: Any, allow_status: List[int] | None = None) → bytes

Send a HTTP PUT request and read the response body.

put_json(url: str, json_data: Any) → Any

Send a HTTP JSON PUT request an parse the JSON response.

request_with_retry(method: str, url: str, data: bytes | None, headers: Dict[str, str], allow_status: List[int] | None) → HTTPResponse

Send a HTTP request and return the unprocessed HTTP response object. Retry once if there is a connection error.

unprocessed_delete_response(url: str, allow_status: List[int] | None = None) → HTTPResponse

Send a HTTP DELETE request and return the unprocessed HTTP response object.

unprocessed_get_response(url: str, allow_status: List[int] | None = None) → HTTPResponse

Send a HTTP GET request and return the unprocessed HTTP response object.

unprocessed_head_response(url: str, allow_status: List[int] | None = None) → HTTPResponse

Send a HTTP HEAD request and return the unprocessed HTTP response object.

unprocessed_post_response(url: str, content_type: str, data: Any, allow_status: List[int] | None = None) → HTTPResponse

Send a HTTP POST request and return the unprocessed HTTP response object.

unprocessed_put_response(url: str, content_type: str, data: Any, allow_status: List[int] | None = None) → HTTPResponse

Send a HTTP PUT request and return the unprocessed HTTP response object.

class network.DirectAccessHeader(protocol: str, name: str, path: str)

A parsed X-FSS-DirectAccess header value.

is_direct_access_possible() → bool

Is direct access possible?

static try_parse_response_header(response: HTTPResponse) → DirectAccessHeader | None

Try to parse a direct access header value.

exception network.UnexpectedStatusError

Unexpected HTTP status code.

network.basic_authentication(username_password: str) → str

Create a HTTP Basic Authentication header value.

network.bearer_authentication(token: AuthorizationToken) → str

Create a HTTP Bearer Authentication header value.

network.delete_empty_directory_on_filestorage(token: AuthorizationToken, *, missing_ok: bool = False) → bool

Delete an empty directory on FSS. Return True if it was empty or missing (if ok), False if not.

network.delete_file_on_filestorage(token: AuthorizationToken, *, allow_direct_access: bool = True) → None

Delete a file on FSS, or just delete it directly locally if possible.

network.file_exists_on_filestorage(token: AuthorizationToken) → bool

Check if a file exists on FSS.

network.get_file_from_filestorage(token: AuthorizationToken, filename: _AnyPath, *, copy_even_on_direct_access: bool = True, allow_direct_access: bool = True, max_block_size: int = 8192) → Path

Download a file from FSS, or just access it directly locally if possible.

network.get_file_size_from_filestorage(token: AuthorizationToken) → int

Get the size of a file from FSS.

network.get_sns_meas_dir_token(conn: Connection, sample_index: str, measurement_index: str, scope: str = 'list', project_serial: str | None = None) → AuthorizationToken

Request a FSS auth token to access a SNS measurement directory.

network.get_sns_meas_file_token(conn: Connection, sample_index: str, measurement_index: str, filename: str, scope: str = 'read', project_serial: str | None = None) → AuthorizationToken

Request a FSS auth token to access SNS a measurement file.

network.get_sns_token(conn: Connection, scopes: List[Tuple[str, str]], hierarchy: str | None = None) → AuthorizationToken

Request a generic SNS FSS auth token.

network.get_virtualpath_dirname(path: str) → str

Extract directory name from a virtualpath.

network.get_virtualpath_filename(path: str) → str

Extract filename from a virtualpath.

network.list_files_and_folders_on_filestorage(token: AuthorizationToken, *, missing_ok: bool = False) → Tuple[List[Dict], List[Dict]]

List the files and folders on FSS.

network.list_files_on_filestorage(token: AuthorizationToken, *, missing_ok: bool = False) → List[Dict]

List the files on FSS.

network.percent_encode_relaxed_virtualpath(path: str) → str

Percent-encode relaxed characters in the virtualpath.

network.put_file_to_filestorage(filename: _AnyPath, token: AuthorizationToken, content_type: str = 'application/octet-stream', *, allow_direct_access: bool = True, max_block_size: int = 100000000) → None

Upload a file to FSS, or just access it directly locally if possible.

network.sns_translate_logical(conn: Connection, name: str, project_serial: str | None = None) → str

Translate a logical via SNS.

network.try_get_direct_access_path(token: AuthorizationToken) → Path | None

Check if directly accessing the path locally is possible and return its path.

network.try_parse_virtualpath_schema_sns_measurement(virtualpath: str) → Tuple[str, str, str, str] | None

Try to parse a @sns/measurement/ virtualpath.

network.virtualpath_join_file(*paths: str) → str

Combines virtualpath elements to a file path.

network.virtualpath_join_folder(*paths: str) → str

Combines virtualpath elements to a folder path.

network.virtualpath_parent_folder(path: str) → str

Extract virtualpath of the parent folder path.