Skip to content

Module config.cli

None

None

View Source
import json

import logging

import os

import random

import re

from typing import Any, List

from cleo import Command

from requests.auth import HTTPBasicAuth, HTTPDigestAuth

from rich.console import Console

from rich.json import JSON

from rich.panel import Panel

from rich.status import Status

from rich.table import Table

from config.exceptions import RequestFailedException

from config.spring import ConfigClient

logging.disable(logging.ERROR)

console = Console()

class ConfigClientCommand(Command):

    """

    Interact with Spring Cloud Server via cli.

    client

        {app : Application name.}

        {filter? : Config selector.}

        {--a|address=http://localhost:8888 : ConfigServer address.}

        {--l|label=master : Branch config.}

        {--p|profile=development : Profile config.}

        {--file : Gets remote file from server and saves locally.}

        {--json : Save output as json.}

        {--all : Show all config.}

        {--auth= : Basic authentication credentials.}

        {--digest= : Digest authentication credentials.}

    """

    EMOJI_ERRORS: List[str] = [

        "\U0001f92f",

        "\U0001f635",

        "\U0001f92e",

        "\U0001f922",

        "\U0001f628",

        "\U0001f62d",

        "\U0001f4a9",

        "\U0001f494",

        "\U0001f4a5",

        "\U0001f525",

    ]

    EMOJI_SUCCESS: List[str] = [

        "\U0001f973",

        "\U0001f929",

        "\U0001f63b",

        "\U0001f496",

        "\U0001f389",

        "\U0001f38a",

    ]

    EMOJI_NOT_FOUND = [

        "\U0001f642",

        "\U0001f60c",

        "\U0001f928",

        "\U0001f643",

        "\U0001f605",

    ]

    def handle(self) -> None:

        filter_options = self.argument("filter") or ""

        client = ConfigClient(

            address=os.getenv("CONFIGSERVER_ADDRESS", self.option("address")),

            label=os.getenv("LABEL", self.option("label")),

            app_name=os.getenv("APP_NAME", self.argument("app")),

            profile=os.getenv("PROFILE", self.option("profile")),

            fail_fast=False,

        )

        if self.option("file"):

            self.request_file(client, filter_options)

            raise SystemExit(0)

        content = self.request_config(client, filter_options)

        if self.option("json"):

            self.save_json("output.json", content)

        else:

            self.std_output(filter_options, content)

    def request_config(self, client: ConfigClient, filter_options: str) -> Any:

        if self.io.verbosity:

            table = Table.grid(padding=(0, 1))

            table.add_column(style="cyan", justify="right")

            table.add_column(style="magenta")

            table.add_row("address[yellow]:[/yellow] ", client.address)

            table.add_row("label[yellow]:[/yellow] ", client.label)

            table.add_row("profile[yellow]:[/yellow] ", client.profile)

            table.add_row("URL[yellow]:[/yellow] ", client.url)

            console.print(

                Panel(

                    table,

                    title="[bold yellow]client info[/bold yellow]",

                    border_style="yellow",

                    expand=True,

                )

            )

        with Status("contacting server...", spinner="dots4") as status:

            emoji = random.choice(self.EMOJI_ERRORS)

            try:

                if self.option("auth"):

                    username, password = self.option("auth").split(":")

                    auth = HTTPBasicAuth(username, password)

                elif self.option("digest"):

                    username, password = self.option("digest").split(":")

                    auth = HTTPDigestAuth(username, password)  # type: ignore

                else:

                    auth = None

                client.get_config(auth=auth)

            except ValueError:

                console.print(

                    f"\nbad credentials format for auth method. Format expected: user:password {emoji}",

                    style="bold",

                )

                raise SystemExit(1)

            except ConnectionError:

                console.print(

                    f"\n[red]failed to contact server![/red] {emoji}", style="bold"

                )

                raise SystemExit(1)

            status.update("OK!")

        content = self.get_config(client, filter_options)

        self.has_content(content, filter_options)

        return content

    def request_file(self, client: ConfigClient, filter_options: str) -> None:

        with Status("contacting server...", spinner="dots4") as status:

            try:

                response = client.get_file(filter_options)

            except RequestFailedException:

                emoji = random.choice(self.EMOJI_ERRORS)

                console.print(

                    f"\n[red]failed to contact server![/red] {emoji}", style="bold"

                )

                raise SystemExit(1)

            with open(f"{filter_options}", "w", encoding="utf-8") as f:

                f.write(response)

            status.update("OK!")

        console.print(f"file saved: [cyan]{filter_options}[/cyan]", style="bold")

    def get_config(self, client: ConfigClient, filter_options: str) -> Any:

        if self.option("all"):

            content = client.config

        else:

            content = client.get(f"{filter_options}")

        return content

    def has_content(self, content, filter_options: str) -> None:

        if len(str(content)) == 0:

            emoji = random.choice(self.EMOJI_NOT_FOUND)

            console.print(

                f"\n{emoji} no result found for filter: [yellow]'[white bold]{filter_options}[/white bold]'[/yellow]",

            )

            raise SystemExit(0)

    def std_output(self, filter_options: str, content: str) -> None:

        if self.option("all"):

            filter_options = "all"

        console.print(

            Panel(

                JSON(json.dumps(content), indent=4, highlight=True, sort_keys=True),

                title=f"[bold green]report for filter: [yellow]'[white italic]{filter_options}[/white italic]'[/yellow][/bold green]",

                highlight=True,

                border_style="white",

                expand=True,

            )

        )

    def save_json(self, filename: str, content: str) -> None:

        with open(f"{filename}", "w", encoding="utf-8") as f:

            json.dump(content, f, indent=4, sort_keys=True)

        console.print(f"file saved: [cyan]{filename}[/cyan]", style="bold")

class DecryptCommand(Command):

    """

    Decrypt a input via Spring Cloud Config.

    decrypt

        {data : Data to decrypt.}

        {--a|address=http://localhost:8888 : ConfigServer address.}

        {--p|path=/decrypt : decrypt path.}

    """

    def handle(self):

        client = ConfigClient(

            address=os.getenv("CONFIGSERVER_ADDRESS", self.option("address")),

            fail_fast=False,

        )

        try:

            data = re.match(r"^.?{cipher}?(?P<name>\w.*)", self.argument("data")).group(

                "name"

            )

        except AttributeError:

            data = self.argument("data")

        try:

            resp = client.decrypt(data, path=self.option("path"))

        except Exception:

            console.print("\n[red]failed to contact server![/red]", style="bold")

            raise SystemExit(1)

        console.print(f"[cyan]{resp}[/cyan]")

class EncryptCommand(Command):

    """

    Encrypt a input via Spring Cloud Config.

    encrypt

        {data : Data to encrypt.}

        {--a|address=http://localhost:8888 : ConfigServer address.}

        {--p|path=/encrypt : encrypt path.}

        {--raw : Format output including {cipher}?}

    """

    def handle(self):

        client = ConfigClient(

            address=os.getenv("CONFIGSERVER_ADDRESS", self.option("address")),

            fail_fast=False,

        )

        try:

            resp = client.encrypt(self.argument("data"), path=self.option("path"))

        except Exception:

            console.print("[red]failed to contact server![/red]", style="bold")

            raise SystemExit(1)

        if not self.option("raw"):

            console.print(

                f"[yellow]'[white]{{cipher}}{resp}[/white]'[/yellow]",

                style="bold",

            )

        else:

            console.print(resp)

Variables

console

Classes

ConfigClientCommand

class ConfigClientCommand(

)
View Source
class ConfigClientCommand(Command):

    """

    Interact with Spring Cloud Server via cli.

    client

        {app : Application name.}

        {filter? : Config selector.}

        {--a|address=http://localhost:8888 : ConfigServer address.}

        {--l|label=master : Branch config.}

        {--p|profile=development : Profile config.}

        {--file : Gets remote file from server and saves locally.}

        {--json : Save output as json.}

        {--all : Show all config.}

        {--auth= : Basic authentication credentials.}

        {--digest= : Digest authentication credentials.}

    """

    EMOJI_ERRORS: List[str] = [

        "\U0001f92f",

        "\U0001f635",

        "\U0001f92e",

        "\U0001f922",

        "\U0001f628",

        "\U0001f62d",

        "\U0001f4a9",

        "\U0001f494",

        "\U0001f4a5",

        "\U0001f525",

    ]

    EMOJI_SUCCESS: List[str] = [

        "\U0001f973",

        "\U0001f929",

        "\U0001f63b",

        "\U0001f496",

        "\U0001f389",

        "\U0001f38a",

    ]

    EMOJI_NOT_FOUND = [

        "\U0001f642",

        "\U0001f60c",

        "\U0001f928",

        "\U0001f643",

        "\U0001f605",

    ]

    def handle(self) -> None:

        filter_options = self.argument("filter") or ""

        client = ConfigClient(

            address=os.getenv("CONFIGSERVER_ADDRESS", self.option("address")),

            label=os.getenv("LABEL", self.option("label")),

            app_name=os.getenv("APP_NAME", self.argument("app")),

            profile=os.getenv("PROFILE", self.option("profile")),

            fail_fast=False,

        )

        if self.option("file"):

            self.request_file(client, filter_options)

            raise SystemExit(0)

        content = self.request_config(client, filter_options)

        if self.option("json"):

            self.save_json("output.json", content)

        else:

            self.std_output(filter_options, content)

    def request_config(self, client: ConfigClient, filter_options: str) -> Any:

        if self.io.verbosity:

            table = Table.grid(padding=(0, 1))

            table.add_column(style="cyan", justify="right")

            table.add_column(style="magenta")

            table.add_row("address[yellow]:[/yellow] ", client.address)

            table.add_row("label[yellow]:[/yellow] ", client.label)

            table.add_row("profile[yellow]:[/yellow] ", client.profile)

            table.add_row("URL[yellow]:[/yellow] ", client.url)

            console.print(

                Panel(

                    table,

                    title="[bold yellow]client info[/bold yellow]",

                    border_style="yellow",

                    expand=True,

                )

            )

        with Status("contacting server...", spinner="dots4") as status:

            emoji = random.choice(self.EMOJI_ERRORS)

            try:

                if self.option("auth"):

                    username, password = self.option("auth").split(":")

                    auth = HTTPBasicAuth(username, password)

                elif self.option("digest"):

                    username, password = self.option("digest").split(":")

                    auth = HTTPDigestAuth(username, password)  # type: ignore

                else:

                    auth = None

                client.get_config(auth=auth)

            except ValueError:

                console.print(

                    f"\nbad credentials format for auth method. Format expected: user:password {emoji}",

                    style="bold",

                )

                raise SystemExit(1)

            except ConnectionError:

                console.print(

                    f"\n[red]failed to contact server![/red] {emoji}", style="bold"

                )

                raise SystemExit(1)

            status.update("OK!")

        content = self.get_config(client, filter_options)

        self.has_content(content, filter_options)

        return content

    def request_file(self, client: ConfigClient, filter_options: str) -> None:

        with Status("contacting server...", spinner="dots4") as status:

            try:

                response = client.get_file(filter_options)

            except RequestFailedException:

                emoji = random.choice(self.EMOJI_ERRORS)

                console.print(

                    f"\n[red]failed to contact server![/red] {emoji}", style="bold"

                )

                raise SystemExit(1)

            with open(f"{filter_options}", "w", encoding="utf-8") as f:

                f.write(response)

            status.update("OK!")

        console.print(f"file saved: [cyan]{filter_options}[/cyan]", style="bold")

    def get_config(self, client: ConfigClient, filter_options: str) -> Any:

        if self.option("all"):

            content = client.config

        else:

            content = client.get(f"{filter_options}")

        return content

    def has_content(self, content, filter_options: str) -> None:

        if len(str(content)) == 0:

            emoji = random.choice(self.EMOJI_NOT_FOUND)

            console.print(

                f"\n{emoji} no result found for filter: [yellow]'[white bold]{filter_options}[/white bold]'[/yellow]",

            )

            raise SystemExit(0)

    def std_output(self, filter_options: str, content: str) -> None:

        if self.option("all"):

            filter_options = "all"

        console.print(

            Panel(

                JSON(json.dumps(content), indent=4, highlight=True, sort_keys=True),

                title=f"[bold green]report for filter: [yellow]'[white italic]{filter_options}[/white italic]'[/yellow][/bold green]",

                highlight=True,

                border_style="white",

                expand=True,

            )

        )

    def save_json(self, filename: str, content: str) -> None:

        with open(f"{filename}", "w", encoding="utf-8") as f:

            json.dump(content, f, indent=4, sort_keys=True)

        console.print(f"file saved: [cyan]{filename}[/cyan]", style="bold")

Ancestors (in MRO)

  • cleo.commands.command.Command
  • cleo.commands.base_command.BaseCommand

Class variables

EMOJI_ERRORS
EMOJI_NOT_FOUND
EMOJI_SUCCESS
TABLE_STYLES
aliases
arguments
commands
description
enabled
help
hidden
name
options
signature
validation

Instance variables

application
config
io

Methods

add_style

def add_style(
    self,
    name,
    fg=None,
    bg=None,
    options=None
)

Adds a new style

View Source
    def add_style(self, name, fg=None, bg=None, options=None):

        """

        Adds a new style

        """

        style = Style(name)

        if fg is not None:

            style.fg(fg)

        if bg is not None:

            style.bg(bg)

        if options is not None:

            if "bold" in options:

                style.bold()

            if "underline" in options:

                style.underlined()

        self._io.output.formatter.add_style(style)

        self._io.error_output.formatter.add_style(style)

add_sub_command

def add_sub_command(
    self,
    command
)
View Source
    def add_sub_command(self, command):  # type: (BaseCommand) -> None

        self._config.add_sub_command_config(command.config)

        command.set_application(self.application)

anonymous

def anonymous(
    self
)
View Source
    def anonymous(self):  # type: () -> BaseCommand

        self._config.anonymous()

        return self

argument

def argument(
    self,
    key=None
)

Get the value of a command argument.

View Source
    def argument(self, key=None):

        """

        Get the value of a command argument.

        """

        if key is None:

            return self._args.arguments()

        return self._args.argument(key)

ask

def ask(
    self,
    question,
    default=None
)

Prompt the user for input.

View Source
    def ask(self, question, default=None):

        """

        Prompt the user for input.

        """

        if isinstance(question, Question):

            return self._io.ask_question(question)

        return self._io.ask(question, default)

call

def call(
    self,
    name,
    args=None
)

Call another command.

View Source
    def call(self, name, args=None):  # type: (str, Optional[str]) -> int

        """

        Call another command.

        """

        if args is None:

            args = ""

        args = StringArgs(args)

        command = self.application.get_command(name)

        return command.run(args, self.io)

call_silent

def call_silent(
    self,
    name,
    args=None
)

Call another command.

View Source
    def call_silent(self, name, args=None):  # type: (str, Optional[str]) -> int

        """

        Call another command.

        """

        if args is None:

            args = ""

        args = StringArgs(args)

        command = self.application.get_command(name)

        return command.run(args, NullIO())

choice

def choice(
    self,
    question,
    choices,
    default=None,
    attempts=None,
    multiple=False
)

Give the user a single choice from an list of answers.

View Source
    def choice(self, question, choices, default=None, attempts=None, multiple=False):

        """

        Give the user a single choice from an list of answers.

        """

        question = ChoiceQuestion(question, choices, default)

        question.set_max_attempts(attempts)

        question.set_multi_select(multiple)

        return self._io.ask_question(question)

comment

def comment(
    self,
    text
)

Write a string as comment output.

Parameters:

Name Type Description Default
text None The line to write None
View Source
    def comment(self, text):

        """

        Write a string as comment output.

        :param text: The line to write

        :type text: str

        """

        self.line(text, "comment")

confirm

def confirm(
    self,
    question,
    default=False,
    true_answer_regex='(?i)^y'
)

Confirm a question with the user.

View Source
    def confirm(self, question, default=False, true_answer_regex="(?i)^y"):

        """

        Confirm a question with the user.

        """

        return self._io.confirm(question, default, true_answer_regex)

create_question

def create_question(
    self,
    question,
    type=None,
    **kwargs
)

Returns a Question of specified type.

View Source
    def create_question(self, question, type=None, **kwargs):

        """

        Returns a Question of specified type.

        """

        if not type:

            return Question(question, **kwargs)

        if type == "choice":

            return ChoiceQuestion(question, **kwargs)

        if type == "confirmation":

            return ConfirmationQuestion(question, **kwargs)

default

def default(
    self,
    default=True
)
View Source
    def default(self, default=True):  # type: (bool) -> BaseCommand

        self._config.default(default)

        return self

get_config

def get_config(
    self,
    client: config.spring.ConfigClient,
    filter_options: str
) -> Any
View Source
    def get_config(self, client: ConfigClient, filter_options: str) -> Any:

        if self.option("all"):

            content = client.config

        else:

            content = client.get(f"{filter_options}")

        return content

handle

def handle(
    self
) -> None

Executes the command.

View Source
    def handle(self) -> None:

        filter_options = self.argument("filter") or ""

        client = ConfigClient(

            address=os.getenv("CONFIGSERVER_ADDRESS", self.option("address")),

            label=os.getenv("LABEL", self.option("label")),

            app_name=os.getenv("APP_NAME", self.argument("app")),

            profile=os.getenv("PROFILE", self.option("profile")),

            fail_fast=False,

        )

        if self.option("file"):

            self.request_file(client, filter_options)

            raise SystemExit(0)

        content = self.request_config(client, filter_options)

        if self.option("json"):

            self.save_json("output.json", content)

        else:

            self.std_output(filter_options, content)

has_content

def has_content(
    self,
    content,
    filter_options: str
) -> None
View Source
    def has_content(self, content, filter_options: str) -> None:

        if len(str(content)) == 0:

            emoji = random.choice(self.EMOJI_NOT_FOUND)

            console.print(

                f"\n{emoji} no result found for filter: [yellow]'[white bold]{filter_options}[/white bold]'[/yellow]",

            )

            raise SystemExit(0)

info

def info(
    self,
    text
)

Write a string as information output.

Parameters:

Name Type Description Default
text None The line to write None
View Source
    def info(self, text):

        """

        Write a string as information output.

        :param text: The line to write

        :type text: str

        """

        self.line(text, "info")

line

def line(
    self,
    text,
    style=None,
    verbosity=None
)

Write a string as information output.

View Source
    def line(self, text, style=None, verbosity=None):

        """

        Write a string as information output.

        """

        if style:

            styled = "<%s>%s" % (style, text)

        else:

            styled = text

        self._io.write_line(styled, verbosity)

line_error

def line_error(
    self,
    text,
    style=None,
    verbosity=None
)

Write a string as information output to stderr.

View Source
    def line_error(self, text, style=None, verbosity=None):

        """

        Write a string as information output to stderr.

        """

        if style:

            styled = "<%s>%s" % (style, text)

        else:

            styled = text

        self._io.error_line(styled, verbosity)

option

def option(
    self,
    key=None
)

Get the value of a command option.

View Source
    def option(self, key=None):

        """

        Get the value of a command option.

        """

        if key is None:

            return self._args.options()

        return self._args.option(key)

overwrite

def overwrite(
    self,
    text,
    size=None
)

Overwrites the current line.

It will not add a new line so use line('') if necessary.

View Source
    def overwrite(self, text, size=None):

        """

        Overwrites the current line.

        It will not add a new line so use line('')

        if necessary.

        """

        self._io.overwrite(text, size=size)

progress_bar

def progress_bar(
    self,
    max=0
)

Creates a new progress bar

Parameters:

Name Type Description Default
max None The maximum number of steps None
View Source
    def progress_bar(self, max=0):

        """

        Creates a new progress bar

        :param max: The maximum number of steps

        :type max: int

        :rtype: ProgressBar

        """

        return self._io.progress_bar(max)

progress_indicator

def progress_indicator(
    self,
    fmt=None,
    interval=100,
    values=None
)

Creates a new progress indicator.

View Source
    def progress_indicator(self, fmt=None, interval=100, values=None):

        """

        Creates a new progress indicator.

        """

        return ProgressIndicator(self.io, fmt, interval, values)

question

def question(
    self,
    text
)

Write a string as question output.

Parameters:

Name Type Description Default
text None The line to write None
View Source
    def question(self, text):

        """

        Write a string as question output.

        :param text: The line to write

        :type text: str

        """

        self.line(text, "question")

render_table

def render_table(
    self,
    headers,
    rows,
    style=None
)

Format input to textual table.

View Source
    def render_table(self, headers, rows, style=None):

        """

        Format input to textual table.

        """

        table = self.table(headers, rows, style)

        table.render(self._io)

request_config

def request_config(
    self,
    client: config.spring.ConfigClient,
    filter_options: str
) -> Any
View Source
    def request_config(self, client: ConfigClient, filter_options: str) -> Any:

        if self.io.verbosity:

            table = Table.grid(padding=(0, 1))

            table.add_column(style="cyan", justify="right")

            table.add_column(style="magenta")

            table.add_row("address[yellow]:[/yellow] ", client.address)

            table.add_row("label[yellow]:[/yellow] ", client.label)

            table.add_row("profile[yellow]:[/yellow] ", client.profile)

            table.add_row("URL[yellow]:[/yellow] ", client.url)

            console.print(

                Panel(

                    table,

                    title="[bold yellow]client info[/bold yellow]",

                    border_style="yellow",

                    expand=True,

                )

            )

        with Status("contacting server...", spinner="dots4") as status:

            emoji = random.choice(self.EMOJI_ERRORS)

            try:

                if self.option("auth"):

                    username, password = self.option("auth").split(":")

                    auth = HTTPBasicAuth(username, password)

                elif self.option("digest"):

                    username, password = self.option("digest").split(":")

                    auth = HTTPDigestAuth(username, password)  # type: ignore

                else:

                    auth = None

                client.get_config(auth=auth)

            except ValueError:

                console.print(

                    f"\nbad credentials format for auth method. Format expected: user:password {emoji}",

                    style="bold",

                )

                raise SystemExit(1)

            except ConnectionError:

                console.print(

                    f"\n[red]failed to contact server![/red] {emoji}", style="bold"

                )

                raise SystemExit(1)

            status.update("OK!")

        content = self.get_config(client, filter_options)

        self.has_content(content, filter_options)

        return content

request_file

def request_file(
    self,
    client: config.spring.ConfigClient,
    filter_options: str
) -> None
View Source
    def request_file(self, client: ConfigClient, filter_options: str) -> None:

        with Status("contacting server...", spinner="dots4") as status:

            try:

                response = client.get_file(filter_options)

            except RequestFailedException:

                emoji = random.choice(self.EMOJI_ERRORS)

                console.print(

                    f"\n[red]failed to contact server![/red] {emoji}", style="bold"

                )

                raise SystemExit(1)

            with open(f"{filter_options}", "w", encoding="utf-8") as f:

                f.write(response)

            status.update("OK!")

        console.print(f"file saved: [cyan]{filter_options}[/cyan]", style="bold")

save_json

def save_json(
    self,
    filename: str,
    content: str
) -> None
View Source
    def save_json(self, filename: str, content: str) -> None:

        with open(f"{filename}", "w", encoding="utf-8") as f:

            json.dump(content, f, indent=4, sort_keys=True)

        console.print(f"file saved: [cyan]{filename}[/cyan]", style="bold")

secret

def secret(
    self,
    question
)

Prompt the user for input but hide the answer from the console.

View Source
    def secret(self, question):

        """

        Prompt the user for input but hide the answer from the console.

        """

        return self._io.ask_hidden(question)

set_application

def set_application(
    self,
    application
)
View Source
    def set_application(self, application):

        self._application = application

        for command in self.commands:

            command.set_application(application)

spin

def spin(
    self,
    start_message,
    end_message,
    fmt=None,
    interval=100,
    values=None
)

Automatically spin a progress indicator.

View Source
    def spin(self, start_message, end_message, fmt=None, interval=100, values=None):

        """

        Automatically spin a progress indicator.

        """

        spinner = ProgressIndicator(self.io, fmt, interval, values)

        return spinner.auto(start_message, end_message)

std_output

def std_output(
    self,
    filter_options: str,
    content: str
) -> None
View Source
    def std_output(self, filter_options: str, content: str) -> None:

        if self.option("all"):

            filter_options = "all"

        console.print(

            Panel(

                JSON(json.dumps(content), indent=4, highlight=True, sort_keys=True),

                title=f"[bold green]report for filter: [yellow]'[white italic]{filter_options}[/white italic]'[/yellow][/bold green]",

                highlight=True,

                border_style="white",

                expand=True,

            )

        )

table

def table(
    self,
    header=None,
    rows=None,
    style=None
)

Return a Table instance.

View Source
    def table(self, header=None, rows=None, style=None):

        """

        Return a Table instance.

        """

        if style is not None:

            style = self.TABLE_STYLES[style]

        table = Table(style)

        if header:

            table.set_header_row(header)

        if rows:

            table.set_rows(rows)

        return table

wrap_handle

def wrap_handle(
    self,
    args,
    io,
    command
)
View Source
    def wrap_handle(

        self, args, io, command

    ):  # type: (Args, IO, CliKitCommand) -> Optional[int]

        self._args = args

        self._io = io

        self._command = command

        return self.handle()

write

def write(
    self,
    text,
    style=None
)

Writes a string without a new line.

Useful if you want to use overwrite().

View Source
    def write(self, text, style=None):

        """

        Writes a string without a new line.

        Useful if you want to use overwrite().

        """

        if style:

            styled = "<%s>%s" % (style, text)

        else:

            styled = text

        self._io.write(styled)

DecryptCommand

class DecryptCommand(

)
View Source
class DecryptCommand(Command):

    """

    Decrypt a input via Spring Cloud Config.

    decrypt

        {data : Data to decrypt.}

        {--a|address=http://localhost:8888 : ConfigServer address.}

        {--p|path=/decrypt : decrypt path.}

    """

    def handle(self):

        client = ConfigClient(

            address=os.getenv("CONFIGSERVER_ADDRESS", self.option("address")),

            fail_fast=False,

        )

        try:

            data = re.match(r"^.?{cipher}?(?P<name>\w.*)", self.argument("data")).group(

                "name"

            )

        except AttributeError:

            data = self.argument("data")

        try:

            resp = client.decrypt(data, path=self.option("path"))

        except Exception:

            console.print("\n[red]failed to contact server![/red]", style="bold")

            raise SystemExit(1)

        console.print(f"[cyan]{resp}[/cyan]")

Ancestors (in MRO)

  • cleo.commands.command.Command
  • cleo.commands.base_command.BaseCommand

Class variables

TABLE_STYLES
aliases
arguments
commands
description
enabled
help
hidden
name
options
signature
validation

Instance variables

application
config
io

Methods

add_style

def add_style(
    self,
    name,
    fg=None,
    bg=None,
    options=None
)

Adds a new style

View Source
    def add_style(self, name, fg=None, bg=None, options=None):

        """

        Adds a new style

        """

        style = Style(name)

        if fg is not None:

            style.fg(fg)

        if bg is not None:

            style.bg(bg)

        if options is not None:

            if "bold" in options:

                style.bold()

            if "underline" in options:

                style.underlined()

        self._io.output.formatter.add_style(style)

        self._io.error_output.formatter.add_style(style)

add_sub_command

def add_sub_command(
    self,
    command
)
View Source
    def add_sub_command(self, command):  # type: (BaseCommand) -> None

        self._config.add_sub_command_config(command.config)

        command.set_application(self.application)

anonymous

def anonymous(
    self
)
View Source
    def anonymous(self):  # type: () -> BaseCommand

        self._config.anonymous()

        return self

argument

def argument(
    self,
    key=None
)

Get the value of a command argument.

View Source
    def argument(self, key=None):

        """

        Get the value of a command argument.

        """

        if key is None:

            return self._args.arguments()

        return self._args.argument(key)

ask

def ask(
    self,
    question,
    default=None
)

Prompt the user for input.

View Source
    def ask(self, question, default=None):

        """

        Prompt the user for input.

        """

        if isinstance(question, Question):

            return self._io.ask_question(question)

        return self._io.ask(question, default)

call

def call(
    self,
    name,
    args=None
)

Call another command.

View Source
    def call(self, name, args=None):  # type: (str, Optional[str]) -> int

        """

        Call another command.

        """

        if args is None:

            args = ""

        args = StringArgs(args)

        command = self.application.get_command(name)

        return command.run(args, self.io)

call_silent

def call_silent(
    self,
    name,
    args=None
)

Call another command.

View Source
    def call_silent(self, name, args=None):  # type: (str, Optional[str]) -> int

        """

        Call another command.

        """

        if args is None:

            args = ""

        args = StringArgs(args)

        command = self.application.get_command(name)

        return command.run(args, NullIO())

choice

def choice(
    self,
    question,
    choices,
    default=None,
    attempts=None,
    multiple=False
)

Give the user a single choice from an list of answers.

View Source
    def choice(self, question, choices, default=None, attempts=None, multiple=False):

        """

        Give the user a single choice from an list of answers.

        """

        question = ChoiceQuestion(question, choices, default)

        question.set_max_attempts(attempts)

        question.set_multi_select(multiple)

        return self._io.ask_question(question)

comment

def comment(
    self,
    text
)

Write a string as comment output.

Parameters:

Name Type Description Default
text None The line to write None
View Source
    def comment(self, text):

        """

        Write a string as comment output.

        :param text: The line to write

        :type text: str

        """

        self.line(text, "comment")

confirm

def confirm(
    self,
    question,
    default=False,
    true_answer_regex='(?i)^y'
)

Confirm a question with the user.

View Source
    def confirm(self, question, default=False, true_answer_regex="(?i)^y"):

        """

        Confirm a question with the user.

        """

        return self._io.confirm(question, default, true_answer_regex)

create_question

def create_question(
    self,
    question,
    type=None,
    **kwargs
)

Returns a Question of specified type.

View Source
    def create_question(self, question, type=None, **kwargs):

        """

        Returns a Question of specified type.

        """

        if not type:

            return Question(question, **kwargs)

        if type == "choice":

            return ChoiceQuestion(question, **kwargs)

        if type == "confirmation":

            return ConfirmationQuestion(question, **kwargs)

default

def default(
    self,
    default=True
)
View Source
    def default(self, default=True):  # type: (bool) -> BaseCommand

        self._config.default(default)

        return self

handle

def handle(
    self
)

Executes the command.

View Source
    def handle(self):

        client = ConfigClient(

            address=os.getenv("CONFIGSERVER_ADDRESS", self.option("address")),

            fail_fast=False,

        )

        try:

            data = re.match(r"^.?{cipher}?(?P<name>\w.*)", self.argument("data")).group(

                "name"

            )

        except AttributeError:

            data = self.argument("data")

        try:

            resp = client.decrypt(data, path=self.option("path"))

        except Exception:

            console.print("\n[red]failed to contact server![/red]", style="bold")

            raise SystemExit(1)

        console.print(f"[cyan]{resp}[/cyan]")

info

def info(
    self,
    text
)

Write a string as information output.

Parameters:

Name Type Description Default
text None The line to write None
View Source
    def info(self, text):

        """

        Write a string as information output.

        :param text: The line to write

        :type text: str

        """

        self.line(text, "info")

line

def line(
    self,
    text,
    style=None,
    verbosity=None
)

Write a string as information output.

View Source
    def line(self, text, style=None, verbosity=None):

        """

        Write a string as information output.

        """

        if style:

            styled = "<%s>%s" % (style, text)

        else:

            styled = text

        self._io.write_line(styled, verbosity)

line_error

def line_error(
    self,
    text,
    style=None,
    verbosity=None
)

Write a string as information output to stderr.

View Source
    def line_error(self, text, style=None, verbosity=None):

        """

        Write a string as information output to stderr.

        """

        if style:

            styled = "<%s>%s" % (style, text)

        else:

            styled = text

        self._io.error_line(styled, verbosity)

option

def option(
    self,
    key=None
)

Get the value of a command option.

View Source
    def option(self, key=None):

        """

        Get the value of a command option.

        """

        if key is None:

            return self._args.options()

        return self._args.option(key)

overwrite

def overwrite(
    self,
    text,
    size=None
)

Overwrites the current line.

It will not add a new line so use line('') if necessary.

View Source
    def overwrite(self, text, size=None):

        """

        Overwrites the current line.

        It will not add a new line so use line('')

        if necessary.

        """

        self._io.overwrite(text, size=size)

progress_bar

def progress_bar(
    self,
    max=0
)

Creates a new progress bar

Parameters:

Name Type Description Default
max None The maximum number of steps None
View Source
    def progress_bar(self, max=0):

        """

        Creates a new progress bar

        :param max: The maximum number of steps

        :type max: int

        :rtype: ProgressBar

        """

        return self._io.progress_bar(max)

progress_indicator

def progress_indicator(
    self,
    fmt=None,
    interval=100,
    values=None
)

Creates a new progress indicator.

View Source
    def progress_indicator(self, fmt=None, interval=100, values=None):

        """

        Creates a new progress indicator.

        """

        return ProgressIndicator(self.io, fmt, interval, values)

question

def question(
    self,
    text
)

Write a string as question output.

Parameters:

Name Type Description Default
text None The line to write None
View Source
    def question(self, text):

        """

        Write a string as question output.

        :param text: The line to write

        :type text: str

        """

        self.line(text, "question")

render_table

def render_table(
    self,
    headers,
    rows,
    style=None
)

Format input to textual table.

View Source
    def render_table(self, headers, rows, style=None):

        """

        Format input to textual table.

        """

        table = self.table(headers, rows, style)

        table.render(self._io)

secret

def secret(
    self,
    question
)

Prompt the user for input but hide the answer from the console.

View Source
    def secret(self, question):

        """

        Prompt the user for input but hide the answer from the console.

        """

        return self._io.ask_hidden(question)

set_application

def set_application(
    self,
    application
)
View Source
    def set_application(self, application):

        self._application = application

        for command in self.commands:

            command.set_application(application)

spin

def spin(
    self,
    start_message,
    end_message,
    fmt=None,
    interval=100,
    values=None
)

Automatically spin a progress indicator.

View Source
    def spin(self, start_message, end_message, fmt=None, interval=100, values=None):

        """

        Automatically spin a progress indicator.

        """

        spinner = ProgressIndicator(self.io, fmt, interval, values)

        return spinner.auto(start_message, end_message)

table

def table(
    self,
    header=None,
    rows=None,
    style=None
)

Return a Table instance.

View Source
    def table(self, header=None, rows=None, style=None):

        """

        Return a Table instance.

        """

        if style is not None:

            style = self.TABLE_STYLES[style]

        table = Table(style)

        if header:

            table.set_header_row(header)

        if rows:

            table.set_rows(rows)

        return table

wrap_handle

def wrap_handle(
    self,
    args,
    io,
    command
)
View Source
    def wrap_handle(

        self, args, io, command

    ):  # type: (Args, IO, CliKitCommand) -> Optional[int]

        self._args = args

        self._io = io

        self._command = command

        return self.handle()

write

def write(
    self,
    text,
    style=None
)

Writes a string without a new line.

Useful if you want to use overwrite().

View Source
    def write(self, text, style=None):

        """

        Writes a string without a new line.

        Useful if you want to use overwrite().

        """

        if style:

            styled = "<%s>%s" % (style, text)

        else:

            styled = text

        self._io.write(styled)

EncryptCommand

class EncryptCommand(

)
View Source
class EncryptCommand(Command):

    """

    Encrypt a input via Spring Cloud Config.

    encrypt

        {data : Data to encrypt.}

        {--a|address=http://localhost:8888 : ConfigServer address.}

        {--p|path=/encrypt : encrypt path.}

        {--raw : Format output including {cipher}?}

    """

    def handle(self):

        client = ConfigClient(

            address=os.getenv("CONFIGSERVER_ADDRESS", self.option("address")),

            fail_fast=False,

        )

        try:

            resp = client.encrypt(self.argument("data"), path=self.option("path"))

        except Exception:

            console.print("[red]failed to contact server![/red]", style="bold")

            raise SystemExit(1)

        if not self.option("raw"):

            console.print(

                f"[yellow]'[white]{{cipher}}{resp}[/white]'[/yellow]",

                style="bold",

            )

        else:

            console.print(resp)

Ancestors (in MRO)

  • cleo.commands.command.Command
  • cleo.commands.base_command.BaseCommand

Class variables

TABLE_STYLES
aliases
arguments
commands
description
enabled
help
hidden
name
options
signature
validation

Instance variables

application
config
io

Methods

add_style

def add_style(
    self,
    name,
    fg=None,
    bg=None,
    options=None
)

Adds a new style

View Source
    def add_style(self, name, fg=None, bg=None, options=None):

        """

        Adds a new style

        """

        style = Style(name)

        if fg is not None:

            style.fg(fg)

        if bg is not None:

            style.bg(bg)

        if options is not None:

            if "bold" in options:

                style.bold()

            if "underline" in options:

                style.underlined()

        self._io.output.formatter.add_style(style)

        self._io.error_output.formatter.add_style(style)

add_sub_command

def add_sub_command(
    self,
    command
)
View Source
    def add_sub_command(self, command):  # type: (BaseCommand) -> None

        self._config.add_sub_command_config(command.config)

        command.set_application(self.application)

anonymous

def anonymous(
    self
)
View Source
    def anonymous(self):  # type: () -> BaseCommand

        self._config.anonymous()

        return self

argument

def argument(
    self,
    key=None
)

Get the value of a command argument.

View Source
    def argument(self, key=None):

        """

        Get the value of a command argument.

        """

        if key is None:

            return self._args.arguments()

        return self._args.argument(key)

ask

def ask(
    self,
    question,
    default=None
)

Prompt the user for input.

View Source
    def ask(self, question, default=None):

        """

        Prompt the user for input.

        """

        if isinstance(question, Question):

            return self._io.ask_question(question)

        return self._io.ask(question, default)

call

def call(
    self,
    name,
    args=None
)

Call another command.

View Source
    def call(self, name, args=None):  # type: (str, Optional[str]) -> int

        """

        Call another command.

        """

        if args is None:

            args = ""

        args = StringArgs(args)

        command = self.application.get_command(name)

        return command.run(args, self.io)

call_silent

def call_silent(
    self,
    name,
    args=None
)

Call another command.

View Source
    def call_silent(self, name, args=None):  # type: (str, Optional[str]) -> int

        """

        Call another command.

        """

        if args is None:

            args = ""

        args = StringArgs(args)

        command = self.application.get_command(name)

        return command.run(args, NullIO())

choice

def choice(
    self,
    question,
    choices,
    default=None,
    attempts=None,
    multiple=False
)

Give the user a single choice from an list of answers.

View Source
    def choice(self, question, choices, default=None, attempts=None, multiple=False):

        """

        Give the user a single choice from an list of answers.

        """

        question = ChoiceQuestion(question, choices, default)

        question.set_max_attempts(attempts)

        question.set_multi_select(multiple)

        return self._io.ask_question(question)

comment

def comment(
    self,
    text
)

Write a string as comment output.

Parameters:

Name Type Description Default
text None The line to write None
View Source
    def comment(self, text):

        """

        Write a string as comment output.

        :param text: The line to write

        :type text: str

        """

        self.line(text, "comment")

confirm

def confirm(
    self,
    question,
    default=False,
    true_answer_regex='(?i)^y'
)

Confirm a question with the user.

View Source
    def confirm(self, question, default=False, true_answer_regex="(?i)^y"):

        """

        Confirm a question with the user.

        """

        return self._io.confirm(question, default, true_answer_regex)

create_question

def create_question(
    self,
    question,
    type=None,
    **kwargs
)

Returns a Question of specified type.

View Source
    def create_question(self, question, type=None, **kwargs):

        """

        Returns a Question of specified type.

        """

        if not type:

            return Question(question, **kwargs)

        if type == "choice":

            return ChoiceQuestion(question, **kwargs)

        if type == "confirmation":

            return ConfirmationQuestion(question, **kwargs)

default

def default(
    self,
    default=True
)
View Source
    def default(self, default=True):  # type: (bool) -> BaseCommand

        self._config.default(default)

        return self

handle

def handle(
    self
)

Executes the command.

View Source
    def handle(self):

        client = ConfigClient(

            address=os.getenv("CONFIGSERVER_ADDRESS", self.option("address")),

            fail_fast=False,

        )

        try:

            resp = client.encrypt(self.argument("data"), path=self.option("path"))

        except Exception:

            console.print("[red]failed to contact server![/red]", style="bold")

            raise SystemExit(1)

        if not self.option("raw"):

            console.print(

                f"[yellow]'[white]{{cipher}}{resp}[/white]'[/yellow]",

                style="bold",

            )

        else:

            console.print(resp)

info

def info(
    self,
    text
)

Write a string as information output.

Parameters:

Name Type Description Default
text None The line to write None
View Source
    def info(self, text):

        """

        Write a string as information output.

        :param text: The line to write

        :type text: str

        """

        self.line(text, "info")

line

def line(
    self,
    text,
    style=None,
    verbosity=None
)

Write a string as information output.

View Source
    def line(self, text, style=None, verbosity=None):

        """

        Write a string as information output.

        """

        if style:

            styled = "<%s>%s" % (style, text)

        else:

            styled = text

        self._io.write_line(styled, verbosity)

line_error

def line_error(
    self,
    text,
    style=None,
    verbosity=None
)

Write a string as information output to stderr.

View Source
    def line_error(self, text, style=None, verbosity=None):

        """

        Write a string as information output to stderr.

        """

        if style:

            styled = "<%s>%s" % (style, text)

        else:

            styled = text

        self._io.error_line(styled, verbosity)

option

def option(
    self,
    key=None
)

Get the value of a command option.

View Source
    def option(self, key=None):

        """

        Get the value of a command option.

        """

        if key is None:

            return self._args.options()

        return self._args.option(key)

overwrite

def overwrite(
    self,
    text,
    size=None
)

Overwrites the current line.

It will not add a new line so use line('') if necessary.

View Source
    def overwrite(self, text, size=None):

        """

        Overwrites the current line.

        It will not add a new line so use line('')

        if necessary.

        """

        self._io.overwrite(text, size=size)

progress_bar

def progress_bar(
    self,
    max=0
)

Creates a new progress bar

Parameters:

Name Type Description Default
max None The maximum number of steps None
View Source
    def progress_bar(self, max=0):

        """

        Creates a new progress bar

        :param max: The maximum number of steps

        :type max: int

        :rtype: ProgressBar

        """

        return self._io.progress_bar(max)

progress_indicator

def progress_indicator(
    self,
    fmt=None,
    interval=100,
    values=None
)

Creates a new progress indicator.

View Source
    def progress_indicator(self, fmt=None, interval=100, values=None):

        """

        Creates a new progress indicator.

        """

        return ProgressIndicator(self.io, fmt, interval, values)

question

def question(
    self,
    text
)

Write a string as question output.

Parameters:

Name Type Description Default
text None The line to write None
View Source
    def question(self, text):

        """

        Write a string as question output.

        :param text: The line to write

        :type text: str

        """

        self.line(text, "question")

render_table

def render_table(
    self,
    headers,
    rows,
    style=None
)

Format input to textual table.

View Source
    def render_table(self, headers, rows, style=None):

        """

        Format input to textual table.

        """

        table = self.table(headers, rows, style)

        table.render(self._io)

secret

def secret(
    self,
    question
)

Prompt the user for input but hide the answer from the console.

View Source
    def secret(self, question):

        """

        Prompt the user for input but hide the answer from the console.

        """

        return self._io.ask_hidden(question)

set_application

def set_application(
    self,
    application
)
View Source
    def set_application(self, application):

        self._application = application

        for command in self.commands:

            command.set_application(application)

spin

def spin(
    self,
    start_message,
    end_message,
    fmt=None,
    interval=100,
    values=None
)

Automatically spin a progress indicator.

View Source
    def spin(self, start_message, end_message, fmt=None, interval=100, values=None):

        """

        Automatically spin a progress indicator.

        """

        spinner = ProgressIndicator(self.io, fmt, interval, values)

        return spinner.auto(start_message, end_message)

table

def table(
    self,
    header=None,
    rows=None,
    style=None
)

Return a Table instance.

View Source
    def table(self, header=None, rows=None, style=None):

        """

        Return a Table instance.

        """

        if style is not None:

            style = self.TABLE_STYLES[style]

        table = Table(style)

        if header:

            table.set_header_row(header)

        if rows:

            table.set_rows(rows)

        return table

wrap_handle

def wrap_handle(
    self,
    args,
    io,
    command
)
View Source
    def wrap_handle(

        self, args, io, command

    ):  # type: (Args, IO, CliKitCommand) -> Optional[int]

        self._args = args

        self._io = io

        self._command = command

        return self.handle()

write

def write(
    self,
    text,
    style=None
)

Writes a string without a new line.

Useful if you want to use overwrite().

View Source
    def write(self, text, style=None):

        """

        Writes a string without a new line.

        Useful if you want to use overwrite().

        """

        if style:

            styled = "<%s>%s" % (style, text)

        else:

            styled = text

        self._io.write(styled)