Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, prompt='', on_change=None, on_move=None, on_accept=None, on_cancel=None,
on_complete_next=None, on_complete_prev=None, completer=None,
history_file=None, history_size=1000, **kwargs):
# Widgets
self._editw = urwid.Edit(prompt, wrap='clip')
self._candsw = CompletionCandidatesWidget(completer)
self._groupw = Group(cls=urwid.Pile)
self._groupw.add(name='candidates', widget=self._candsw, visible=False, options=10)
self._groupw.add(name='cmdline', widget=self._editw, visible=True, options='pack')
super().__init__(self._groupw)
# Callbacks
self._on_change = blinker.Signal()
self._on_move = blinker.Signal()
self._on_accept = blinker.Signal()
self._on_cancel = blinker.Signal()
self._on_complete_next = blinker.Signal()
self._on_complete_prev = blinker.Signal()
# Completion
self._completer = completer
self._on_complete_next.connect(self._cb_complete_next)
self._on_complete_prev.connect(self._cb_complete_prev)
self._on_change.connect(self._cb_change)
self._on_move.connect(self._cb_move)
self._user_input = None
# History
self._history = []
self._history_size = history_size
self._history_pos = -1
def __init__(self, app):
self.transactionPushed = Signal(object)
self.app = app
self.currentIndex = 0
self.stack = OrderedDict()
self.transaction = None
def __init__(self, name, parent, direction, **kwargs):
super(AnyPin, self).__init__(name, parent, direction, **kwargs)
self.typeChanged = Signal(str)
self.dataTypeBeenSet = Signal()
self.setDefaultValue(None)
self._isAny = True
# if True, setType and setDefault will work only once
self.singleInit = False
self.enableOptions(PinOptions.ChangeTypeOnConnection)
self._defaultSupportedDataTypes = self._supportedDataTypes = tuple([pin.__name__ for pin in getAllPinClasses() if pin.IsValuePin()])
self.canChange = True
self.super = None
import blinker
focus = blinker.Signal()
set_prompt = blinker.Signal()
set_prompt_yn = blinker.Signal()
set_message = blinker.Signal()
redraw_status = blinker.Signal()
call_delay = blinker.Signal()
self._state = ReportState(run_on_save=False, report_is_running=False)
# TODO: Add signal that file changed. Then subscribe to it from
# server.py
# TODO: Move to scriptrunner.py
self.state_changed = Signal(
doc="""Emitted when our state changes
Parameters
----------
state : ReportState
the ReportSession's current ReportState
""")
self.on_report_changed = Signal(
doc="""Emitted when our report's source_file is changed on disk,
and self.run_on_save is False. When this happens, the browser
alerts the user that the report has changed and prompts them to
re-run it.
"""
)
self.on_report_was_manually_stopped = Signal(
doc="""Emitted when our running report is manually stopped."""
)
self.set_client_connection(initial_client_connection)
def on_error(self):
"""Emitted when an error is received.
The signal sender is the consumer and the ``error`` is sent as an
argument.
"""
return blinker.Signal(doc='Emitted when a error is received.')
def __init__(self):
self.deferred: typing.Dict[str, str] = {}
self.changed = blinker.Signal()
self.errored = blinker.Signal()
# Options must be the last attribute here - after that, we raise an
# error for attribute assigment to unknown options.
self._options: typing.Dict[str, typing.Any] = {}
import urwid
import blinker
import textwrap
from mitmproxy.tools.console import layoutwidget
from mitmproxy.tools.console import signals
HELP_HEIGHT = 5
command_focus_change = blinker.Signal()
class CommandItem(urwid.WidgetWrap):
def __init__(self, walker, cmd, focused):
self.walker, self.cmd, self.focused = walker, cmd, focused
super().__init__(None)
self._w = self.get_widget()
def get_widget(self):
parts = [
("focus", ">> " if self.focused else " "),
("text", "async " if self.cmd.asyncf else " "),
("title", self.cmd.path),
("text", " "),
("text", " ".join(self.cmd.paramnames())),
]
Parameters
----------
state : ReportState
the ReportSession's current ReportState
""")
self.on_report_changed = Signal(
doc="""Emitted when our report's source_file is changed on disk,
and self.run_on_save is False. When this happens, the browser
alerts the user that the report has changed and prompts them to
re-run it.
"""
)
self.on_report_was_manually_stopped = Signal(
doc="""Emitted when our running report is manually stopped."""
)
self.set_client_connection(initial_client_connection)