def list_connectors()

in connectors/connectors_cli.py [0:0]

• 34 lines of code
• 4 McCabe index (conditional complexity)

def list_connectors(obj):
    connector = Connector(config=obj["config"]["elasticsearch"])
    coro = connector.list_connectors()

    try:
        connectors = asyncio.run(coro)
        click.echo("")
        if len(connectors) == 0:
            click.echo("No connectors found")
            return

        click.echo(f"Showing {len(connectors)} connectors \n")

        table_rows = []
        for connector in connectors:
            formatted_connector = [
                click.style(connector.id, fg="green"),
                click.style(connector.index_name, fg="white"),
                click.style(connector.service_type, fg="white"),
                click.style(connector.status.value, fg="white"),
                click.style(connector.last_sync_status.value, fg="white"),
            ]
            table_rows.append(formatted_connector)

        click.echo(
            tabulate(
                table_rows,
                headers=[
                    "ID",
                    "Index name",
                    "Service type",
                    "Status",
                    "Last sync job status",
                ],
            )
        )
    except asyncio.CancelledError as e:
        click.echo(e)