Module isilon.commands.objects
View Source
import asyncio
from cleo import Command
import isilon
class ObjectsCommand(Command):
"""
Objects.
objects
{container : Container name.}
{object : Object name.}
{--headers=* : HTTP headers.}
{--data=? : Object data.}
{--c|create : Create or replace object.}
{--m|metadata : Show object metadata.}
{--u|update : Create or update object metadata.}
{--d|delete : Delete object.}
"""
def handle(self):
container_name = str(self.argument("container"))
object_name = str(self.argument("object"))
if self.option("create"):
data = self.option("data")
if not data:
self.line("<error>Please, provides a valid object data.</>")
raise SystemExit(1)
asyncio.run(isilon.objects.create_large(container_name, object_name, data))
self.line(f"\n<options=bold><comment>{object_name}</comment> created.</>")
elif self.option("metadata"):
resp = asyncio.run(
isilon.objects.show_metadata(container_name, object_name)
)
for meta_key, meta_value in resp.items():
self.line(f"<options=bold>{meta_key}</>: {meta_value}")
elif self.option("update"):
asyncio.run(isilon.objects.update_metadata(container_name, object_name))
self.line("<options=bold>metadata updated.</>")
elif self.option("delete"):
asyncio.run(isilon.objects.delete(container_name, object_name))
self.line(f"<options=bold><comment>{object_name}</comment> deleted.</>")
else:
asyncio.run(
isilon.objects.get_large(container_name, object_name, object_name)
)
self.line(f"<options=bold><comment>{object_name}</comment> saved.</>")
Classes
ObjectsCommand
class ObjectsCommand(
)
Objects.
objects {container : Container name.} {object : Object name.} {--headers=* : HTTP headers.} {--data=? : Object data.} {--c|create : Create or replace object.} {--m|metadata : Show object metadata.} {--u|update : Create or update object metadata.}
View Source
class ObjectsCommand(Command):
"""
Objects.
objects
{container : Container name.}
{object : Object name.}
{--headers=* : HTTP headers.}
{--data=? : Object data.}
{--c|create : Create or replace object.}
{--m|metadata : Show object metadata.}
{--u|update : Create or update object metadata.}
{--d|delete : Delete object.}
"""
def handle(self):
container_name = str(self.argument("container"))
object_name = str(self.argument("object"))
if self.option("create"):
data = self.option("data")
if not data:
self.line("<error>Please, provides a valid object data.</>")
raise SystemExit(1)
asyncio.run(isilon.objects.create_large(container_name, object_name, data))
self.line(f"\n<options=bold><comment>{object_name}</comment> created.</>")
elif self.option("metadata"):
resp = asyncio.run(
isilon.objects.show_metadata(container_name, object_name)
)
for meta_key, meta_value in resp.items():
self.line(f"<options=bold>{meta_key}</>: {meta_value}")
elif self.option("update"):
asyncio.run(isilon.objects.update_metadata(container_name, object_name))
self.line("<options=bold>metadata updated.</>")
elif self.option("delete"):
asyncio.run(isilon.objects.delete(container_name, object_name))
self.line(f"<options=bold><comment>{object_name}</comment> deleted.</>")
else:
asyncio.run(
isilon.objects.get_large(container_name, object_name, object_name)
)
self.line(f"<options=bold><comment>{object_name}</comment> saved.</>")
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.
:param text: The line to write :type text: str
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):
container_name = str(self.argument("container"))
object_name = str(self.argument("object"))
if self.option("create"):
data = self.option("data")
if not data:
self.line("<error>Please, provides a valid object data.</>")
raise SystemExit(1)
asyncio.run(isilon.objects.create_large(container_name, object_name, data))
self.line(f"\n<options=bold><comment>{object_name}</comment> created.</>")
elif self.option("metadata"):
resp = asyncio.run(
isilon.objects.show_metadata(container_name, object_name)
)
for meta_key, meta_value in resp.items():
self.line(f"<options=bold>{meta_key}</>: {meta_value}")
elif self.option("update"):
asyncio.run(isilon.objects.update_metadata(container_name, object_name))
self.line("<options=bold>metadata updated.</>")
elif self.option("delete"):
asyncio.run(isilon.objects.delete(container_name, object_name))
self.line(f"<options=bold><comment>{object_name}</comment> deleted.</>")
else:
asyncio.run(
isilon.objects.get_large(container_name, object_name, object_name)
)
self.line(f"<options=bold><comment>{object_name}</comment> saved.</>")
info
def info(
self,
text
)
Write a string as information output.
:param text: The line to write :type text: str
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
:param max: The maximum number of steps :type max: int
:rtype: ProgressBar
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.
:param text: The line to write :type text: str
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)