doodledashboard/cli.py
def view(action, dashboards, secrets):
    """View the output of the datafeeds and/or notifications used in your DASHBOARDS"""

    if secrets is None:
        secrets = os.path.join(os.path.expanduser("~"), "/.doodledashboard/secrets")

    try:
        loaded_secrets = try_read_secrets_file(secrets)
    except InvalidSecretsException as err:
        click.echo(get_error_message(err, default="Secrets file is invalid"), err=True)
        raise click.Abort()

    dashboard_config = DashboardConfigReader(initialise_component_loader(), loaded_secrets)

    read_configs = [read_file(f) for f in dashboards]
    dashboard = read_dashboard_from_config(dashboard_config, read_configs)

    try:
        messages = DashboardRunner(dashboard).poll_datafeeds()
    except SecretNotFound as err:
        click.echo(get_error_message(err, default="Datafeed didn't have required secret"), err=True)
        raise click.Abort()

    cli_output = {"source-data": messages}

    if action == "notifications":
        cli_output["notifications"] = []

        for notification in dashboard.notifications:
            notification_output = notification.create(messages)

            filtered_messages = messages

            if isinstance(notification, FilteredNotification):
                filtered_messages = notification.filter_messages(messages)

            cli_output["notifications"].append({
                "filtered-messages": filtered_messages,
                "notification": str(notification_output)
            })
    json_output = json.dumps(cli_output, sort_keys=True, indent=4, cls=MessageJsonEncoder)
    click.echo(json_output)
def view(action, dashboards, secrets):
    """View the output of the datafeeds and/or notifications used in your DASHBOARDS"""

    if secrets is None:
        secrets = os.path.join(os.path.expanduser("~"), "/.doodledashboard/secrets")

    try:
        loaded_secrets = try_read_secrets_file(secrets)
    except InvalidSecretsException as err:
        click.echo(get_error_message(err, default="Secrets file is invalid"), err=True)
        raise click.Abort()

    dashboard_config = DashboardConfigReader(initialise_component_loader(), loaded_secrets)

    read_configs = [read_file(f) for f in dashboards]
    dashboard = read_dashboard_from_config(dashboard_config, read_configs)

    try:
        messages = DashboardRunner(dashboard).poll_datafeeds()
    except SecretNotFound as err:
        click.echo(get_error_message(err, default="Datafeed didn't have required secret"), err=True)
        raise click.Abort()

    cli_output = {"source-data": messages}

    if action == "notifications":
        cli_output["notifications"] = []

        for notification in dashboard.notifications:
            notification_output = notification.create(messages)

            filtered_messages = messages

            if isinstance(notification, FilteredNotification):
                filtered_messages = notification.filter_messages(messages)

            cli_output["notifications"].append({
                "filtered-messages": filtered_messages,
                "notification": str(notification_output)
            })
    json_output = json.dumps(cli_output, sort_keys=True, indent=4, cls=MessageJsonEncoder)
    click.echo(json_output)
View the output of the datafeeds and/or notifications used in your DASHBOARDS