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, renew_token: Callable[[], AuthorizationToken] | None = None) → 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.