Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def pull():
cwd = os.getcwd()
utils.save_history([{}, "", cwd])
data = utils.drive_data()
if cwd not in data.keys():
click.secho(
"following directory has not been tracked: \nuse drive add_remote or drive clone ", fg='red')
sys.exit(0)
fid = data[cwd]['id']
current_root = utils.get_file(fid)
click.secho("checking for changes in '" +
current_root['name'] + "' ....", fg='magenta')
utils.pull_content(cwd, fid)
click.secho(current_root['name'] +
" is up to date with drive", fg='yellow')
Black's output is shown directly to the user, so even in events of errors it is
expected that the users sees the error directly.
We need this function to work differently than the rest of ``esss_fix_format`` (which processes
each file individually) because we want to call black only once, reformatting all files
given to it in the command-line.
:return: a pair (would_be_formatted, black_failed)
"""
py_files = [x for x in files if should_format(str(x), ['*.py'], exclude_patterns)[0]]
black_failed = False
would_be_formatted = False
if py_files:
if check:
click.secho(f'Checking black on {len(py_files)} files...', fg='cyan')
else:
click.secho(f'Running black on {len(py_files)} files...', fg='cyan')
# on Windows there's a limit on the command-line size, so we call black in batches
# this should only be an issue when executing fix-format over the entire repository,
# not on day to day usage
chunk_size = 100
for chunked_files in boltons.iterutils.chunked(py_files, chunk_size):
args = ['black']
if check:
args.append('--check')
if verbose:
args.append('--verbose')
args.extend(str(x) for x in chunked_files)
status = subprocess.call(args)
if not black_failed:
black_failed = not check and status != 0
def determine_release_version(ctx: Context) -> None:
ctx.release_version = (
datetime.datetime.now(datetime.timezone.utc)
.astimezone(tz.gettz("US/Pacific"))
.strftime("%Y.%m.%d")
)
if ctx.release_version in ctx.last_release_version:
click.secho(
f"The release version {ctx.release_version} is already used.", fg="red"
)
ctx.release_version = click.prompt("Please input another version: ")
click.secho(f"Releasing {ctx.release_version}.")
from samtranslator.translator.managed_policy_translator import ManagedPolicyLoader
from samcli.commands.exceptions import UserException
from samcli.commands.local.cli_common.user_exceptions import InvalidSamTemplateException
from .lib.exceptions import InvalidSamDocumentException
from .lib.sam_template_validator import SamTemplateValidator
sam_template = _read_sam_file(template)
iam_client = boto3.client("iam")
validator = SamTemplateValidator(sam_template, ManagedPolicyLoader(iam_client))
try:
validator.is_valid()
except InvalidSamDocumentException as e:
click.secho("Template provided at '{}' was invalid SAM Template.".format(template), bg="red")
raise InvalidSamTemplateException(str(e))
except NoCredentialsError as e:
raise UserException("AWS Credentials are required. Please configure your credentials.")
click.secho("{} is a valid SAM Template".format(template), fg="green")
def after_upgrade_success(self):
click.secho(
"…and we're done. Good job, everyone! " + click.style("(◕‿◕✿)", bold=True),
fg="green",
)
def run(ctx, docker_run_args):
"""Run a Docker container"""
tag = ctx.obj['tag']
mount_volume_from_host = True
cmd = get_run_cmd(tag, mount_volume_from_host, docker_run_args)
click.secho('Invoking: %s' % ' '.join(cmd), fg='yellow')
status = subprocess.call(cmd)
click.echo()
for name, data in self.buckets.viewitems():
if not data['exists']:
self.create_bucket(name)
data['exists'] = True
for bucket, data in self.rclone_queue.viewitems():
click.echo(cformat('Copying %{cyan}{}%{reset} files (%{cyan}{}%{reset}) to %{cyan}{}%{reset} via rclone')
.format(data['files'], do_filesizeformat(data['bytes']), bucket))
start = datetime.now()
try:
subprocess.check_call([
'rclone', 'copy', '--copy-links',
data['path'], '{}:{}'.format(self.rclone_remote, bucket)
])
except subprocess.CalledProcessError:
click.secho('\nError while running rclone', fg='red')
raise
duration = (datetime.now() - start)
click.echo('...finished after {}'.format(format_human_timedelta(duration, 'minutes', narrow=True)))
rmlinktree(data['path'])
self.rclone_queue.clear()
try:
who = whois.whois(domain)
results = {}
# Check if info was returned before proceeding because sometimes records are protected
if who.registrar:
results['domain_name'] = who.domain_name
results['registrar'] = who.registrar
results['expiration_date'] = who.expiration_date
results['registrant'] = who.name
results['org'] = who.org
results['admin_email'] = who.emails[0]
results['tech_email'] = who.emails[1]
results['address'] = "{}, {}{}, {}, {}".format(who.address,who.city,who.zipcode,who.state,who.country)
results['dnssec'] = who.dnssec
else:
click.secho("[*] WHOIS record for {} came back empty. You might try looking at dnsstuff.com.".format(domain),fg="yellow")
return results
except Exception as error:
click.secho("[!] The WHOIS lookup for {} failed!".format(domain),fg="red")
click.secho("L.. Details: {}".format(error),fg="red")
def new_user(username, email, password, group):
"""Creates a new user. Omit any options to use the interactive mode."""
try:
user = create_user(username, password, email, group)
click.secho("[+] User {} with Email {} in Group {} created.".format(
user.username, user.email, user.primary_group.name), fg="cyan"
)
except IntegrityError:
raise FlaskBBCLIError("Couldn't create the user because the "
"username or email address is already taken.",
fg="red")
if test_data:
click.secho("[+] Adding some test data...", fg="cyan")
create_test_data()
if bulk_data:
timer = time.time()
topic_count, post_count = insert_bulk_data(int(topics), int(posts))
elapsed = time.time() - timer
click.secho("[+] It took {} seconds to create {} topics and {} posts"
.format(elapsed, topic_count, post_count), fg="cyan")
# this just makes the most sense for the command name; use -i to
# init the db as well
if not test_data:
click.secho("[+] Populating the database with some defaults...",
fg="cyan")
create_default_groups()
create_default_settings()