♻️ servicelib.fastapi tools and rabbitmq.rpc errors interface (#5157)

Author: pcrespov

.env-devel

@@ -1,3 +1,4 @@
+# local development
 #
 # - Keep it alfphabetical order and grouped by prefix [see vscode cmd: Sort Lines Ascending]
 # - To expose:

packages/models-library/src/models_library/api_schemas_payments/errors.py

@@ -0,0 +1,65 @@
+from pydantic.errors import PydanticErrorMixin
+
+
+class _BaseRpcApiError(PydanticErrorMixin, ValueError):
+    @classmethod
+    def get_full_class_name(cls) -> str:
+        # Can be used as unique code identifier
+        return f"{cls.__module__}.{cls.__name__}"
+
+
+#
+# service-wide errors
+#
+
+
+class PaymentServiceUnavailableError(_BaseRpcApiError):
+    msg_template = "Payments are currently unavailable: {human_readable_detail}"
+
+
+#
+# payment transactions errors
+#
+
+
+class PaymentsError(_BaseRpcApiError):
+    msg_template = "Error in payment transaction '{payment_id}'"
+
+
+class PaymentNotFoundError(PaymentsError):
+    msg_template = "Payment transaction '{payment_id}' was not found"
+
+
+class PaymentAlreadyExistsError(PaymentsError):
+    msg_template = "Payment transaction '{payment_id}' was already initialized"
+
+
+class PaymentAlreadyAckedError(PaymentsError):
+    msg_template = "Payment transaction '{payment_id}' cannot be changes since it was already closed."
+
+
+#
+# payment-methods errors
+#
+
+
+class PaymentsMethodsError(_BaseRpcApiError):
+    ...
+
+
+class PaymentMethodNotFoundError(PaymentsMethodsError):
+    msg_template = "The specified payment method '{payment_method_id}' does not exist"
+
+
+class PaymentMethodAlreadyAckedError(PaymentsMethodsError):
+    msg_template = (
+        "Cannot create payment-method '{payment_method_id}' since it was already closed"
+    )
+
+
+class PaymentMethodUniqueViolationError(PaymentsMethodsError):
+    msg_template = "Payment method '{payment_method_id}' aready exists"
+
+
+class InvalidPaymentMethodError(PaymentsMethodsError):
+    msg_template = "Invalid payment method '{payment_method_id}'"

packages/service-library/src/servicelib/fastapi/app_state.py

@@ -0,0 +1,36 @@
+import logging
+
+from fastapi import FastAPI
+
+_logger = logging.getLogger(__name__)
+
+
+class SingletonInAppStateMixin:
+    """
+    Mixin to get, set and delete an instance of 'self' from/to app.state
+    """
+
+    app_state_name: str  # Name used in app.state.$(app_state_name)
+    frozen: bool = True  # Will raise if set multiple times
+
+    @classmethod
+    def get_from_app_state(cls, app: FastAPI):
+        return getattr(app.state, cls.app_state_name)
+
+    def set_to_app_state(self, app: FastAPI):
+        if (exists := getattr(app.state, self.app_state_name, None)) and self.frozen:
+            msg = f"An instance of {type(self)} already in app.state.{self.app_state_name}={exists}"
+            raise ValueError(msg)
+
+        setattr(app.state, self.app_state_name, self)
+        return self.get_from_app_state(app)
+
+    @classmethod
+    def pop_from_app_state(cls, app: FastAPI):
+        """
+        Raises:
+            AttributeError: if instance is not in app.state
+        """
+        old = getattr(app.state, cls.app_state_name)
+        delattr(app.state, cls.app_state_name)
+        return old

packages/service-library/src/servicelib/fastapi/http_client.py

@@ -58,73 +58,3 @@ async def check_liveness(self) -> LivenessResult:
             return IsResponsive(elapsed=response.elapsed)
         except httpx.RequestError as err:
             return IsNonResponsive(reason=f"{err}")
-
-
-class AppStateMixin:
-    """
-    Mixin to get, set and delete an instance of 'self' from/to app.state
-    """
-
-    app_state_name: str  # Name used in app.state.$(app_state_name)
-    frozen: bool = True  # Will raise if set multiple times
-
-    @classmethod
-    def get_from_app_state(cls, app: FastAPI):
-        return getattr(app.state, cls.app_state_name)
-
-    def set_to_app_state(self, app: FastAPI):
-        if (exists := getattr(app.state, self.app_state_name, None)) and self.frozen:
-            msg = f"An instance of {type(self)} already in app.state.{self.app_state_name}={exists}"
-            raise ValueError(msg)
-
-        setattr(app.state, self.app_state_name, self)
-        return self.get_from_app_state(app)
-
-    @classmethod
-    def pop_from_app_state(cls, app: FastAPI):
-        """
-        Raises:
-            AttributeError: if instance is not in app.state
-        """
-        old = getattr(app.state, cls.app_state_name)
-        delattr(app.state, cls.app_state_name)
-        return old
-
-
-def to_curl_command(request: httpx.Request, *, use_short_options: bool = True) -> str:
-    """Composes a curl command from a given request
-
-    Can be used to reproduce a request in a separate terminal (e.g. debugging)
-    """
-    # Adapted from https://github.com/marcuxyz/curlify2/blob/master/curlify2/curlify.py
-    method = request.method
-    url = request.url
-
-    # https://curl.se/docs/manpage.html#-X
-    # -X, --request {method}
-    _x = "-X" if use_short_options else "--request"
-    request_option = f"{_x} {method}"
-
-    # https://curl.se/docs/manpage.html#-d
-    # -d, --data <data>          HTTP POST data
-    data_option = ""
-    if body := request.read().decode():
-        _d = "-d" if use_short_options else "--data"
-        data_option = f"{_d} '{body}'"
-
-    # https://curl.se/docs/manpage.html#-H
-    # H, --header <header/@file> Pass custom header(s) to server
-
-    headers_option = ""
-    headers = []
-    for key, value in request.headers.items():
-        if "secret" in key.lower() or "pass" in key.lower():
-            headers.append(f'"{key}: *****"')
-        else:
-            headers.append(f'"{key}: {value}"')
-
-    if headers:
-        _h = "-H" if use_short_options else "--header"
-        headers_option = f"{_h} {f' {_h} '.join(headers)}"
-
-    return f"curl {request_option} {headers_option} {data_option} {url}"

packages/service-library/src/servicelib/fastapi/httpx_utils.py

@@ -0,0 +1,84 @@
+import httpx
+
+
+def _is_secret(k: str) -> bool:
+    return "secret" in k.lower() or "pass" in k.lower()
+
+
+def _get_headers_safely(request: httpx.Request) -> dict[str, str]:
+    return {k: "*" * 5 if _is_secret(k) else v for k, v in request.headers.items()}
+
+
+def to_httpx_command(
+    request: httpx.Request, *, use_short_options: bool = True, multiline: bool = False
+) -> str:
+    """Command with httpx CLI
+
+    $ httpx --help
+
+    NOTE: Particularly handy as an alternative to curl (e.g. when docker exec in osparc containers)
+    SEE https://www.python-httpx.org/
+    """
+    cmd = [
+        "httpx",
+    ]
+
+    #  -m, --method METHOD
+    cmd.append(f'{"-m" if use_short_options else "--method"} {request.method}')
+
+    # -c, --content TEXT  Byte content to include in the request body.
+    if content := request.read().decode():
+        cmd.append(f'{"-c" if use_short_options else "--content"} \'{content}\'')
+
+    # -h, --headers <NAME VALUE> ...  Include additional HTTP headers in the request.
+    if headers := _get_headers_safely(request):
+        cmd.extend(
+            [
+                f'{"-h" if use_short_options else "--headers"} "{name}" "{value}"'
+                for name, value in headers.items()
+            ]
+        )
+
+    cmd.append(f"{request.url}")
+    separator = " \\\n" if multiline else " "
+    return separator.join(cmd)
+
+
+def to_curl_command(
+    request: httpx.Request, *, use_short_options: bool = True, multiline: bool = False
+) -> str:
+    """Composes a curl command from a given request
+
+    $ curl --help
+
+    NOTE: Handy reproduce a request in a separate terminal (e.g. debugging)
+    """
+    # Adapted from https://github.com/marcuxyz/curlify2/blob/master/curlify2/curlify.py
+    cmd = [
+        "curl",
+    ]
+
+    # https://curl.se/docs/manpage.html#-X
+    # -X, --request {method}
+    cmd.append(f'{"-X" if use_short_options else "--request"} {request.method}')
+
+    # https://curl.se/docs/manpage.html#-H
+    # H, --header <header/@file> Pass custom header(s) to server
+    if headers := _get_headers_safely(request):
+        cmd.extend(
+            [
+                f'{"-H" if use_short_options else "--header"} "{k}: {v}"'
+                for k, v in headers.items()
+            ]
+        )
+
+    # https://curl.se/docs/manpage.html#-d
+    # -d, --data <data>          HTTP POST data
+    if body := request.read().decode():
+        _d = "-d" if use_short_options else "--data"
+        cmd.append(f"{_d} '{body}'")
+
+    cmd.append(f"{request.url}")
+
+    separator = " \\\n" if multiline else " "
+    return separator.join(cmd)

packages/service-library/src/servicelib/rabbitmq/_client_rpc.py

@@ -12,6 +12,7 @@
 
 from ..logging_utils import log_context
 from ._client_base import RabbitMQClientBase
+from ._constants import RPC_REQUEST_DEFAULT_TIMEOUT_S
 from ._errors import RemoteMethodNotRegisteredError, RPCNotInitializedError
 from ._models import RPCNamespacedMethodName
 from ._rpc_router import RPCRouter
@@ -65,7 +66,7 @@ async def request(
         namespace: RPCNamespace,
         method_name: RPCMethodName,
         *,
-        timeout_s: PositiveInt | None = 5,
+        timeout_s: PositiveInt | None = RPC_REQUEST_DEFAULT_TIMEOUT_S,
         **kwargs,
     ) -> Any:
         """

packages/service-library/src/servicelib/rabbitmq/_constants.py

@@ -1,4 +1,7 @@
 from typing import Final
 
+from pydantic import PositiveInt
+
 BIND_TO_ALL_TOPICS: Final[str] = "#"
+RPC_REQUEST_DEFAULT_TIMEOUT_S: Final[PositiveInt] = PositiveInt(5)
 RPC_REMOTE_METHOD_TIMEOUT_S: Final[int] = 30

packages/service-library/src/servicelib/rabbitmq/_rpc_router.py

@@ -6,48 +6,73 @@
 from typing import Any, TypeVar
 
 from models_library.rabbitmq_basic_types import RPCMethodName
-from pydantic import SecretStr
 
 from ..logging_utils import log_context
 from ._errors import RPCServerError
 
 DecoratedCallable = TypeVar("DecoratedCallable", bound=Callable[..., Any])
 
+# NOTE: this is equivalent to http access logs
 _logger = logging.getLogger("rpc.access")
 
-_RPC_CUSTOM_ENCODER: dict[Any, Callable[[Any], Any]] = {
-    SecretStr: SecretStr.get_secret_value
-}
+
+def _create_func_msg(func, args: tuple[Any, ...], kwargs: dict[str, Any]) -> str:
+    msg = f"{func.__name__}("
+
+    if args_msg := ", ".join(map(str, args)):
+        msg += args_msg
+
+    if kwargs_msg := ", ".join({f"{name}={value}" for name, value in kwargs.items()}):
+        if args:
+            msg += ", "
+        msg += kwargs_msg
+
+    return f"{msg})"
 
 
 @dataclass
 class RPCRouter:
     routes: dict[RPCMethodName, Callable] = field(default_factory=dict)
 
-    def expose(self) -> Callable[[DecoratedCallable], DecoratedCallable]:
-        def decorator(func: DecoratedCallable) -> DecoratedCallable:
+    def expose(
+        self,
+        *,
+        reraise_if_error_type: tuple[type[Exception], ...] | None = None,
+    ) -> Callable[[DecoratedCallable], DecoratedCallable]:
+        def _decorator(func: DecoratedCallable) -> DecoratedCallable:
             @functools.wraps(func)
-            async def wrapper(*args, **kwargs):
+            async def _wrapper(*args, **kwargs):
+
                 with log_context(
+                    # NOTE: this is intentionally analogous to the http access log traces.
+                    # To change log-level use getLogger("rpc.access").set_level(...)
                     _logger,
                     logging.INFO,
-                    msg=f"calling {func.__name__} with {args}, {kwargs}",
+                    msg=f"RPC call {_create_func_msg(func, args, kwargs)}",
+                    log_duration=True,
                 ):
                     try:
                         return await func(*args, **kwargs)
+
                     except asyncio.CancelledError:
                         _logger.debug("call was cancelled")
                         raise
+
                     except Exception as exc:  # pylint: disable=broad-except
+                        if reraise_if_error_type and isinstance(
+                            exc, reraise_if_error_type
+                        ):
+                            raise
+
                         _logger.exception("Unhandled exception:")
                         # NOTE: we do not return internal exceptions over RPC
                         raise RPCServerError(
                             method_name=func.__name__,
-                            exc_type=f"{type(exc)}",
+                            exc_type=f"{exc.__class__.__module__}.{exc.__class__.__name__}",
                             msg=f"{exc}",
                         ) from None
 
-            self.routes[RPCMethodName(func.__name__)] = wrapper
+            self.routes[RPCMethodName(func.__name__)] = _wrapper
             return func
 
-        return decorator
+        return _decorator