#
# Copyright © 2018-2026 Commonwealth Scientific and Industrial Research
# Organisation (CSIRO) ABN 41 687 119 230.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Error handling for the Pathling command line interface.
JVM exceptions raised through Py4J carry verbose Java stack traces that are
unhelpful at the command line. This module unwraps them to their root message
and maps recognised categories onto concise, actionable guidance. Stack traces
are shown only when the user passes ``--verbose``.
Author: John Grimes.
"""
from __future__ import annotations
import re
import traceback
from typing import Optional
# Exit code for a successful invocation.
EXIT_SUCCESS = 0
# Exit code for a runtime failure (a command failed while executing).
EXIT_RUNTIME = 1
# Exit code for a usage error (the command was invoked incorrectly).
EXIT_USAGE = 2
# Signals that an unwrapped root message describes an authentication failure
# rather than a connection, timeout, or server-side problem. The SMART
# backend-services setup surfaces "Failed to retrieve SMART configuration" when
# the token endpoint or credential is wrong, so that phrasing is included
# alongside the standard OAuth2 rejection codes.
_AUTH_PATTERN = re.compile(
r"\b401\b"
r"|\bunauthorized\b"
r"|invalid_client|invalid_grant|invalid_scope|unauthorized_client"
r"|smart configuration"
r"|smart auth"
r"|access[ _]token"
r"|token endpoint"
r"|authentication (?:failed|error)"
r"|failed to authenticate"
r"|credential",
re.IGNORECASE,
)
[docs]class CliError(Exception):
"""An error with a message that is safe to show the user directly.
:param message: the human-readable error message.
:param exit_code: the process exit code to use; defaults to a runtime
failure.
"""
def __init__(self, message: str, exit_code: int = EXIT_RUNTIME) -> None:
super().__init__(message)
self.message = message
self.exit_code = exit_code
[docs]def unwrap_java_exception(exc: BaseException) -> str:
"""Extracts the most useful message from an exception.
Py4J wraps Java exceptions, whose ``str`` representation includes the full
stack trace. Where the wrapped exception can report its own message, that
message is returned in full, because a library failure may deliberately span
several lines - the import failure that lists a release's language reference
sets, for example, is only useful if the operator can see the list. Otherwise
only the leading line is taken, since the rest would be stack frames. The
Java exception class prefix is stripped where present.
:param exc: the exception to unwrap.
:return: a message describing the underlying problem.
"""
java_exception = getattr(exc, "java_exception", None)
own_message = False
if java_exception is not None:
get_message = getattr(java_exception, "getMessage", None)
message = get_message() if callable(get_message) else None
own_message = bool(message)
if not message:
message = str(java_exception)
else:
message = str(exc)
lines = [line for line in message.splitlines() if line.strip()] or [message]
# Strip a leading fully-qualified Java exception class name, e.g.
# "java.lang.IllegalArgumentException: actual message".
match = re.match(r"^(?:[\w.$]+Exception|[\w.$]+Error):\s*(.*)$", lines[0])
lines[0] = match.group(1) if match and match.group(1) else lines[0]
if not own_message:
# The remaining lines are stack frames rather than part of the message.
return lines[0].strip()
return "\n".join([lines[0].strip(), *(line.rstrip() for line in lines[1:])])
def _categorise(root_message: str) -> Optional[str]:
"""Classifies a root message into a known error category.
:param root_message: the unwrapped root message.
:return: a category key, or None when the message is not recognised.
"""
lowered = root_message.lower()
if "connection refused" in lowered or "failed to connect" in lowered:
return "connection"
if "unknownhost" in lowered or "unknown host" in lowered:
return "connection"
if _AUTH_PATTERN.search(root_message):
return "auth"
# Require a FHIRPath-specific signal rather than the bare substring "parse
# error", which can appear in unrelated parse failures (FR-012).
if "fhirpath" in lowered:
return "fhirpath"
return None
[docs]def is_connection_error(exc: BaseException) -> bool:
"""Determines whether an exception represents a server connection failure.
:param exc: the exception to classify.
:return: True when the unwrapped message looks like a connection failure.
"""
return _categorise(unwrap_java_exception(exc)) == "connection"
[docs]def is_auth_error(exc: BaseException) -> bool:
"""Determines whether an exception represents an authentication failure.
Used to decide whether an export failure should be reported as an
authentication problem; connection, timeout, and server-side errors that
happen to occur while authentication is configured must not be misdiagnosed
as bad credentials.
:param exc: the exception to classify.
:return: True when the unwrapped message looks like an authentication
failure.
"""
return _categorise(unwrap_java_exception(exc)) == "auth"
[docs]def friendly_message(
exc: BaseException,
verbose: bool = False,
server_url: Optional[str] = None,
store_path: Optional[str] = None,
) -> str:
"""Builds a friendly, actionable message for an unexpected exception.
:param exc: the exception to describe.
:param verbose: when True, append the full traceback.
:param server_url: a server URL to name in connection errors, or None.
:param store_path: a local terminology store path to name in local-mode
failures, or None. When set, the message names the store and suggests
the import commands, and never names a terminology server (FR-011).
:return: the message to present to the user.
"""
root = unwrap_java_exception(exc)
category = _categorise(root)
# In local mode, name the store and point at the import commands rather than
# naming a server that was never contacted (FR-011). This takes precedence
# over the connection-error branch below.
if store_path is not None:
# The library failure usually already names the store path, so only
# restate it when the root cause has not, to avoid a repeated path.
if store_path in root:
message = (
f"{root} Check that it has been populated with 'pathling "
"import-snomed' or 'pathling import-fhir-terminology'."
)
else:
message = (
f"{root} The local terminology store at {store_path} could not "
"be used. Check that it exists and has been populated with "
"'pathling import-snomed' or 'pathling import-fhir-terminology'."
)
if verbose:
trace = "".join(
traceback.format_exception(type(exc), exc, exc.__traceback__)
)
message = message + "\n\n" + trace
return message
if category == "connection" and server_url:
message = (
f"Could not reach the server at {server_url}: {root}. "
"Check the URL and your network connection."
)
elif category == "fhirpath":
message = (
f"The FHIRPath expression could not be evaluated: {root}. "
"Check the expression syntax."
)
else:
message = root or f"{type(exc).__name__}"
if not verbose:
# A multi-line message ends in content the reader needs to see intact, so the hint goes
# on a line of its own rather than being run onto the last of them.
separator = "\n" if "\n" in message else " "
message = (
f"{message}{separator}Re-run with --verbose for the full stack trace."
)
if verbose:
# Use the three-argument form for Python 3.9 compatibility.
trace = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
message = message + "\n\n" + trace
return message