Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def normalize_formatting(code: str) -> str:
"""Returns a string of the code with normalized formatting(spaces, indents, newlines) for easier compares"""
return autopep8.fix_code(code, options={"aggressive": 2})
batch_exec_pattern = re.compile("(def\s+execute\s*\(\s*self\s*,\s*\*\*kwargs\s*\)\s*:)")
online_exec_pattern = re.compile("(def\s+execute\s*\(\s*self\s*,\s*input_message\s*,\s*\*\*kwargs\s*\)\s*:)")
CLAZZES = {
"acquisitor": "AcquisitorAndCleaner",
"tpreparator": "TrainingPreparator",
"trainer": "Trainer",
"evaluator": "MetricsEvaluator",
"ppreparator": "PredictionPreparator",
"predictor": "Predictor"
}
for cell in cells:
if cell['cell_type'] == 'code' and cell["metadata"].get("marvin_cell", False):
source = cell["source"]
new_source = autopep8.fix_code(source, options={'max_line_length': 160})
marvin_action = cell["metadata"]["marvin_cell"]
marvin_action_clazz = getattr(__import__(Config.get("package")), CLAZZES[marvin_action])
source_path = inspect.getsourcefile(marvin_action_clazz)
fnew_source_lines = []
for new_line in new_source.split("\n"):
fnew_line = " " + new_line + "\n" if new_line.strip() else "\n"
if not new_line.startswith("import") and not new_line.startswith("from") and not new_line.startswith("print"):
for artifact in artifacts.keys():
fnew_line = re.sub(artifacts[artifact], 'self.' + artifact, fnew_line)
fnew_source_lines.append(fnew_line)
if marvin_action == "predictor":
def format_autopep8(raw: str) -> str:
import autopep8
return autopep8.fix_code(raw, options={'max_line_length': 100})
:return:
"""
filename = os.path.join(target_path, filename)
dirname = os.path.dirname(filename)
if len(dirname) and not os.path.exists(dirname):
os.makedirs(dirname)
with open(filename, 'w') as f:
if not raw:
rendered = render_template(template_name, context=context)
else:
rendered = render_file(template_name)
if not NO_PEP and filename.endswith('.py'):
rendered = autopep8.fix_code(rendered)
f.write(rendered)
with open(file_path, 'r') as f:
file_contents = f.read()
if update_header:
file_contents = update_file_header(file_contents,
os.path.basename(file_path),
PYTHON_HEADER_FORMAT,
PYTHON_HEADER_REGEX)
if format_code:
# Use the autopep8 module to format the source code. autopep8 uses
# pycodestyle to detect the style errors it should fix and thus it
# should fix all (or most) of them, however, it does not use pylint
# so it may not fix all of its reported errors.
options = {"max_line_length": 100}
file_contents = autopep8.fix_code(file_contents, options=options)
with open(file_path, 'w') as f:
f.write(file_contents)
def fix_line_range(source_code, start, end, options):
"""Apply autopep8 (and docformatter) between the lines start and end of
source."""
# TODO confirm behaviour outside range (indexing starts at 1)
start = max(start, 1)
options.line_range = [start, end]
from autopep8 import fix_code
fixed = fix_code(source_code, options)
try:
if options.docformatter:
from docformatter import format_code
fixed = format_code(
fixed,
summary_wrap_length=options.max_line_length - 1,
description_wrap_length=(options.max_line_length
- 2 * options.indent_size),
pre_summary_newline=options.pre_summary_newline,
post_description_blank=options.post_description_blank,
force_wrap=options.force_wrap,
line_range=[start, end])
except AttributeError: # e.g. using autopep8.parse_args, pragma: no cover
pass
def onAutopep8(self, evt):
import autopep8
sline = self.GetFirstVisibleLine()
txt = self.GetText()
txt = autopep8.fix_code(txt)
self.SetText(txt)
self.ScrollToLine(sline)
evt.Skip()
def format_str(src_contents):
return autopep8.fix_code(src_contents)
if not result.tree_changed:
debug(lambda: 'Tree not changed')
continue
tree = working_tree
debug(lambda: 'Tree changed:\n{}'.format(dump(tree)))
dependencies.extend(result.dependencies)
try:
code = unparse(tree)
debug(lambda: 'Code changed:\n{}'.format(code))
except:
raise TransformationError(path, transformer,
dump(tree), format_exc())
return fix_code(code), dependencies
experiment_name=metadata['experiment_name'],
pipeline_name=metadata['pipeline_name'],
pipeline_description=metadata.get('pipeline_description', ''),
pipeline_arguments=pipeline_args,
pipeline_arguments_names=', '.join(pipeline_args_names),
docker_base_image=metadata.get('docker_image', ''),
volumes=volumes,
leaf_nodes=leaf_nodes,
working_dir=metadata.get('abs_working_dir', None),
marshal_volume=marshal_volume,
marshal_path=marshal_path,
auto_snapshot=auto_snapshot
)
# fix code style using pep8 guidelines
pipeline_code = autopep8.fix_code(pipeline_code)
return pipeline_code