Source code for maestral.cli.common

from __future__ import annotations

import functools
import sys
from typing import Callable, Any, TypeVar, TYPE_CHECKING
from typing_extensions import ParamSpec

import click

from .core import ConfigName
from .output import warn
from .utils import get_term_size
from ..constants import DEFAULT_CONFIG_NAME

    from ..daemon import MaestralProxy
    from ..main import Maestral

[docs] P = ParamSpec("P")
[docs] T = TypeVar("T")
[docs] def convert_api_errors(func: Callable[P, T]) -> Callable[P, T]: """ Decorator that catches a MaestralApiError and prints a formatted error message to stdout before exiting. Calls ``sys.exit(1)`` after printing the error to stdout. """ from ..exceptions import MaestralApiError @functools.wraps(func) def wrapper(*args: P.args, **kwargs: P.kwargs) -> T: try: return func(*args, **kwargs) except MaestralApiError as exc: warn(f"{exc.title}. {exc.message}") sys.exit(1) return wrapper
[docs] def check_for_fatal_errors(m: MaestralProxy | Maestral) -> bool: """ Checks the given Maestral instance for fatal errors such as revoked Dropbox access, deleted Dropbox folder etc. Prints a nice representation to the command line. :param m: Proxy to Maestral daemon or Maestral instance. :returns: True in case of fatal errors, False otherwise. """ import textwrap maestral_err_list = m.fatal_errors if len(maestral_err_list) > 0: size = get_term_size() err = maestral_err_list[0] wrapped_msg = textwrap.fill(err.message, width=size.columns) click.echo("") click.secho(err.title, fg="red") click.secho(wrapped_msg, fg="red") click.echo("") return True else: return False
[docs] config_option = click.option( "-c", "--config-name", default=DEFAULT_CONFIG_NAME, type=ConfigName(existing=False), is_eager=True, expose_value=True, help="Run command with the given configuration.", )
[docs] existing_config_option = click.option( "-c", "--config-name", default=DEFAULT_CONFIG_NAME, type=ConfigName(), is_eager=True, expose_value=True, help="Run command with the given configuration.", )
[docs] def inject_proxy( fallback: bool, existing_config: bool ) -> Callable[[Callable[P, T]], Callable[P, Any]]: def decorator(f: Callable[P, T]) -> Callable[P, Any]: def wrapper(*args: P.args, **kwargs: P.kwargs) -> Any: from ..daemon import MaestralProxy, CommunicationError ctx = click.get_current_context() config_name = ctx.params.pop("config_name", "maestral") kwargs.pop("config_name", None) try: proxy = ctx.with_resource(MaestralProxy(config_name, fallback=fallback)) except CommunicationError: click.echo("Maestral daemon is not running.") ctx.exit(0) else: return ctx.invoke(f, proxy, *args, **kwargs) if existing_config: f = existing_config_option(f) else: f = config_option(f) return functools.update_wrapper(wrapper, f) return decorator