def display(id):
''' Display build information by build id. '''
history = load_history()
build = get_build_info(history, id or history['current'])
is_current = build['id'] == history['current']
timestamp = local_timestamp(build['timestamp'])
table = SingleTable([
[green('Build ' + build['id'])],
['ID: ' + green(build['id'])],
['Commit: ' + green(build['commit'])],
['Branch: ' + green(build['branch'])],
['Stage: ' + green(build['stage'])],
['Created By: ' + green(build['createdBy'])],
['Path: ' + green(build['path'])],
['Current Build: ' + green('Yes' if is_current else 'No')],
['Timestamp: ' + green(timestamp)]
])
print(table.table)
python类SingleTable()的实例源码
def display_list(history):
''' Display build history. '''
if not history['builds']:
remote_info('No builds have been deployed yet.')
return
remote_info('Showing recent builds')
# Map build data into tabular format
data = map(row_mapper_wrt(history['current']), history['builds'])
# Prepend heading rows
data.insert(0, [
' ', 'ID', 'Commit',
'Branch', 'Created By', 'Timestamp'
])
table = SingleTable(data)
print('')
print(table.table)
def print_run_table(table_data):
table = SingleTable(table_data)
table.justify_columns = {0: 'left', 1: 'center', 2: 'left'}
table.inner_heading_row_border = False
table.inner_column_border = False
table.outer_border = False
max_width = table.column_max_width(2)
for index, row in enumerate(table_data):
table.table_data[index][2] = str(row[2][0:max_width].splitlines()[0])
if row[1] == 0:
table.table_data[index][1] = colored(str(row[1]), 'green')
elif row[1] == 1:
table.table_data[index][2] = colored(str(row[1]), 'yellow')
elif row[1] == 3:
table.table_data[index][2] = colored(str(row[1]), 'grey')
else:
table.table_data[index][2] = colored(str(row[1]), 'red')
print table.table
def show(args, client, unparsed=None):
parser = argparse.ArgumentParser(description="Show a ticket and its replies.")
parser.add_argument('ticketid', metavar='TICKETID', type=int,
help="The ticket to show.")
args = parser.parse_args(args=unparsed, namespace=args)
t = linode.SupportTicket(client, args.ticketid)
try:
t.summary
except:
print("No ticket found with ID {}".format(args.ticketid))
sys.exit(0)
data = [ [ t.summary, t.opened ] ]
data.append([ "Regrading {}".format(t.entity.label) if t.entity else '', "Status: {}".format(t.status) ])
data.append([ t.description ])
print(SingleTable(data).table)
for r in t.replies:
data = [ [ "Reply from {}".format(r.created_by), r.created ] ]
data.append([ r.description ])
print(SingleTable(data).table)
def config_list(args, client, unparsed=None):
parser = argparse.ArgumentParser(description="List all configs (ports) for a specific NodeBalancer")
parser.add_argument('label', metavar='LABEL', type=str,
help="The NodeBalancer to list.")
args = parser.parse_args(args=unparsed, namespace=args)
n = _get_nodebalancer_or_die(client, args.label)
configs = n.configs
if not configs:
print("{} has no configs".format(n.label))
sys.exit(0)
data = [ [ "port", "protocol", "algorithm", "stickiness", "check", "node_status" ] ]
for c in configs:
data.append([ c.port, c.protocol, c.algorithm, c.stickiness, c.check, "{} UP, {} DOWN".format(c.nodes_status.up, c.nodes_status.down) ])
tab = SingleTable(data)
print(tab.table)
def list(args, client, unparsed=None):
linodes = client.linode.get_instances()
groups = {}
for l in linodes:
if not l.group in groups:
groups[l.group] = []
groups[l.group].append(l)
header = [ "label", "status", "location", "backups", "disk", "memory" ]
for k in sorted(groups.keys()):
if args.raw:
for l in groups[k]:
print(args.separator.join(_make_raw_linode_row(k,l)))
else:
print(k if k else 'Linode')
data = [ _make_linode_row(l) for l in groups[k] ]
data = [ header ] + data
tab = SingleTable(data)
print(tab.table)
def list_accounts(raw):
with Spinner('Fetching accounts: '):
accounts = newrelic.get_accounts()
if raw:
for account in accounts:
print(account[0])
data = [[
'ID',
'Name'
]]
for account in accounts:
data.append([account['id'], account['name']])
table = SingleTable(data)
table.title = click.style('Accounts', fg='black')
print(table.table)
def display(job, click):
"""
Display the job information.
:param job: A Job object provided by ScrapingHub
:param click: A click object used to print on the terminal
"""
if job:
main_info = get_job_main_info(job)
table_data = list(TABLE_JOB_MODEL)
table_data.append(
[main_info['spider'], main_info['started_time'], main_info['items_scraped'], main_info['tags'],
main_info['state'], main_info['close_reason'], main_info['errors_count'], main_info['version']]
)
table = SingleTable(table_data)
click.echo(table.table)
def display_jobs(jobs, click):
"""
Display all the jobs' information contained in a JobSet.
:param jobs: A JobSet object provided by ScrapingHub with all the jobs
:param click: A click object used to print on the terminal
"""
table_data = list(TABLE_JOBS_MODEL)
for job in jobs:
main_info = get_job_main_info(job)
table_data.append(
[main_info['id'], main_info['spider'], main_info['started_time'], main_info['items_scraped'],
main_info['tags'], main_info['state'], main_info['close_reason'], main_info['errors_count'],
main_info['version']]
)
table = SingleTable(table_data)
click.echo(table.table)
def buy(args):
path = args.config['book-path']
# Check if the file book.json exist and if not create it
# and iniate it with '[]'
if not os.path.isfile(path):
data = "[]"
file = open(path, 'a+')
file.write(data)
file.close()
order = Order(args)
data = json.loads(open(path).read())
data.append(order.toDict())
with open(path, 'w') as outfile:
json.dump(data, outfile, indent=4)
# Display the order with a nice table
table_data = [
['id', 'Exchange', 'Market', 'Price'],
[order.id , order.exchange, order.market, order.last]
]
table = SingleTable(table_data)
table.title = 'Buy Order'
print (table.table)
def close(args):
path = args.config['book-path']
if not os.path.isfile(path):
print('Order book empty, use buy command to fill it')
return
data = json.loads(open(path).read())
i = 0
while i < len(data):
if data[i]['id'] == args.id:
table_data = [['id', 'Exchange', 'Market', 'Price', 'Current', 'Profit']]
current = get_last_price_tmp(data[i]['market'])
profit = get_profit(data[i]['last'], current)
table_data.append([data[i]['id'], data[i]['exchange'],
data[i]['market'], data[i]['last'], current , profit])
table = SingleTable(table_data)
table.title = 'Close Order'
print (table.table)
data.pop(i)
break
i += 1
with open('./book.json', 'w') as outfile:
json.dump(data, outfile, indent=4)
def position(args):
path = args.config['book-path']
if not os.path.isfile(path):
print('Order book empty, use buy command to fill it')
return
data = json.loads(open(path).read())
table_data = [['id', 'Exchange', 'Market', 'Price', 'Current', 'Profit']]
i = 0
while i < len(data):
current = get_last_price_tmp(data[i]['market'])
profit = get_profit(data[i]['last'], current)
table_data.append([data[i]['id'], data[i]['exchange'],
data[i]['market'], data[i]['last'], current , profit])
i += 1
table = SingleTable(table_data)
if args.live:
return(table.table)
else:
print(table.table)
def list(item_type, **kwargs):
"""
List all available items of a specified type.
:param item_type: experiment/campaign/wsn-generation-algorithm
:param kwargs: simulation keyword arguments (see the documentation for more information)
"""
data, title = [['Name']], None
if item_type == 'experiments':
title = 'Available experiments'
data.extend([['- {}'.format(x).ljust(25)] for x in list_experiments()])
elif item_type == 'campaigns':
title = 'Available campaigns'
data.extend([['- {}'.format(x).ljust(25)] for x in list_campaigns()])
elif item_type == 'wsn-generation-algorithms':
title = 'Available WSN generation algorithms'
data.extend([['- {}'.format(x).ljust(25)] for x in list_wsn_gen_algorithms()])
if title is not None:
table = SingleTable(data, title)
print(table.table)
# ***************************************** SETUP COMMANDS ****************************************
def do_status(self, line):
"""
Display process pool status.
"""
self.clean_tasks()
# this prevents from re-displaying the same status table once ENTER is pressed
# (marker 'restart' is handled in emptyline() hereafter
if line == 'restart' and self.__last_tasklist is not None and \
hash(repr(self.tasklist)) == self.__last_tasklist:
return
self.__last_tasklist = hash(repr(copy(self.tasklist)))
if len(self.tasklist) == 0:
data = [['No task currently running']]
else:
data = [['Task', 'Status', 'Result']]
for task, info in sorted(self.tasklist.items(), key=lambda x: str(x[0])):
data.append([str(task).ljust(15), info['status'].ljust(10), str(info['result']).ljust(40)])
table = SingleTable(data, 'Status of opened tasks')
table.justify_columns = {0: 'center', 1: 'center', 2: 'center'}
print(table.table)
def as_table(self, line):
"""
:param line: Convert data which splited by '\n' or array into beautiful table splited by '\n'
:return: beautified table
>>> len(zpy(processor=None).as_table([[0, 1, 4, 9, 16], [25, 36, 49, 64, 81]])) > 0
True
"""
from terminaltables import AsciiTable, SingleTable
if isinstance(line, str):
arr = line.split('\n')
prepared_arr = [' '.join(x.split()).split(' ') for x in arr]
else: #Iterable
prepared_arr = line
##Todo refactor
#prepared_arr = [' '.join(x.split() if isinstance(x,str) else str(x)).split(' ') for x in arr]
return SingleTable(prepared_arr).table
def __init__(self, *args, **kwargs):
"""If we specified the column_widths already in the constructor, then set them right away"""
self.column_widths = kwargs.get("column_widths", None)
super(SingleTable, self).__init__(*args, **kwargs)
def column_widths(self):
"""If we set our own column width, then we take that one, otherwise we will use the default one"""
if self.__column_widths:
return self.__column_widths
else:
return super(SingleTable, self).column_widths
def print_table(rows, title):
print(SingleTable(rows, title).table)
def warn():
"""
Deprecation warning for old way to invoke script
"""
msg = [ [ Color('{yellow}WARNING{/yellow}') ], [ "The 'linode-beta' command has been deprecated and renamed to "
"'linode-cli'. Please invoke with that command to hide this warning." ] ]
print(SingleTable(msg).table)
main()
def list(args, client, unparsed=None):
tickets = client.support.get_tickets(linode.SupportTicket.closed == None)
# TODO linode.SupportTicket.status != 'closed')
data = [ [ "id", "summary", "regarding", "updated", "updated by" ] ]
for t in tickets:
data.append([ t.id, t.summary, t.entity.label if t.entity else '', t.updated,
_colorize_updater(t.updated_by) ])
tab = SingleTable(data)
print(tab.table)
def list(args, client, unparsed=None):
nodebalancers = client.get_nodebalancers()
header = [ "label", "region" ]
data = [ header ]
for n in nodebalancers:
data.append([ n.label, n.region.id ])
tab = SingleTable(data)
print(tab.table)
def list(args, client, unparsed=None):
zones = client.get_domains()
data = [ _make_domain_row(d) for d in zones ]
if args.raw:
for d in data:
print(args.separator.join(d))
else:
data = [ [ 'domain', 'type', 'soa email' ] ] + data
tab = SingleTable(data)
print(tab.table)
def record_list(args, client, unparsed=None):
parser = argparse.ArgumentParser(description="Create a Domain.")
parser.add_argument('label', metavar='LABEL', type=str,
help="The Domain to delete.")
parser.add_argument('-y','--type', metavar='TYPE', type=str,
help="Optional. Allows domain record filtering by type. "
"One of: NS, MX, A, AAAA, CNAME, TXT, or SRV")
args = parser.parse_args(args=unparsed, namespace=args)
z = _get_domain_or_die(client, args.label)
if not args.raw:
print("Domain records for {}".format(z.domain))
if not z.records:
print("No records to list.")
return
data = []
for r in z.records:
if args.type and not r.zone_record_type == args.type:
continue
data.append(_make_domain_record_row(r))
if args.raw:
for d in data:
print(args.separator.join(d))
else:
data = [ ['type', 'name', 'target', 'port' ] ] + data
tab = SingleTable(data)
print(tab.table)
def list(args, client, unparsed=None):
stackscripts = client.linode.get_stackscripts(mine_only=True)
data = [ _make_stackscript_row(s) for s in stackscripts ]
data = [ [ 'id', 'label', 'public', 'revision note' ] ] + data
tab = SingleTable(data)
print(tab.table)
def list(args, client, unparsed=None):
events = client.account.get_events()
header = [ " ", "entity", "action", "user", "status", "time" ]
data = [ header ]
for e in events[:10]:
data.append(_make_event_row(e))
tab = SingleTable(data)
print(tab.table)
def list(args, client, unparsed=None):
volumes = client.linode.get_volumes()
header = [ "label", "status", "size", "location", "attached to" ]
data = [ header ] + [ _make_volume_row(v) for v in volumes ]
tab = SingleTable(data)
print(tab.table)
def as_table(tlv, title=None, redact=False):
res = [['Tag', 'Name', 'Value']]
for tag, value in tlv.items():
res.append([format_bytes(tag.id),
tag.name or '',
'\n'.join(textwrap.wrap(render_element(tag, value, redact=redact), 80))])
table = SingleTable(res)
if title is not None:
table.title = title
return table.table
# Function called by the emvtool shim installed by setuptools
def card_info(self):
card = self.get_reader()
apps = card.list_applications()
if type(apps) != list:
apps = [apps]
print("Available apps: %s" % (", ".join(render_element(Tag.APP_LABEL, app[Tag.APP_LABEL])
for app in apps)))
for app in apps:
print("\nApplication %s, DF Name: %s" % (
render_element(Tag.APP_LABEL, app[Tag.APP_LABEL]),
render_element(Tag.DF, app[Tag.ADF_NAME])))
data = card.select_application(app[Tag.ADF_NAME]).data
print(as_table(data[Tag.FCI][Tag.FCI_PROP], 'FCI Proprietary Data', redact=self.args.redact))
for i in range(1, 10):
try:
rec = card.read_record(1, sfi=i).data
except ErrorResponse as e:
break
print(as_table(rec[Tag.RECORD], 'File: %s' % i, redact=self.args.redact))
print("\nFetching card data...")
try:
tab = SingleTable(card.get_metadata().items())
tab.inner_heading_row_border = False
print(tab.table)
except ErrorResponse as e:
print("Unable to fetch card data: %s" % e)
def list_locations(ctx, raw):
with Spinner('Fetching locations: '):
locations = newrelic.get_locations(ctx.obj['ACCOUNT'])
if raw:
print(json.dumps(locations))
return
data = [[
'#',
'City',
'Continent',
'Code',
'Availability',
'Accessibility',
]]
for number, location in enumerate(locations.values()):
available = click.style(u'?', fg='green')
if not location['available']:
click.style(u'?', fg='red')
private = 'Private' if location['private'] else 'Public'
data.append([
number,
location['label'],
location['continent'],
location['name'],
available,
private,
])
table = SingleTable(data)
table.title = click.style('Locations', fg='black')
for i in [0, 4, 5]:
table.justify_columns[i] = 'right'
print(table.table)
def display_hc_jobs(jobs, click):
"""
Display all the jobs' information contained in a JobSet.
:param jobs: A JobSet object provided by ScrapingHub with all the jobs
:param click: A click object used to print on the terminal
"""
table_data = list(HC_TABLE_JOBS_MODEL)
for job in jobs:
main_info = get_hc_job_main_info(job)
table_data.append(
[main_info['id'], main_info['spider'], main_info['finished_time'], main_info['state'], main_info['close_reason'], main_info['errors_count'], main_info['logs'], main_info['version']]
)
table = SingleTable(table_data)
click.echo(table.table)