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)