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)
def view ( action , dashboards , secrets ) : 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
View the output of the datafeeds and / or notifications used in your DASHBOARDS