How to use the iredis.utils.timer function in iredis

To help you get started, we’ve selected a few iredis examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github laixintao / iredis / tests / test_utils.py View on Github external
def test_timer():
    with patch("iredis.utils.logger") as mock_logger:
        timer("foo")
        time.sleep(0.1)
        timer("bar")
        mock_logger.debug.assert_called()
        args, kwargs = mock_logger.debug.call_args
        matched = re.match(r"\[timer (\d)\] (0\.\d+) -> bar", args[0])

        assert matched.group(1) == str(3)
        assert 0.1 <= float(matched.group(2)) <= 0.2

        # --- test again ---
        timer("foo")
        time.sleep(0.2)
        timer("bar")
        mock_logger.debug.assert_called()
        args, kwargs = mock_logger.debug.call_args
        matched = re.match(r"\[timer (\d)\] (0\.\d+) -> bar", args[0])

        assert matched.group(1) == str(5)
        assert 0.2 <= float(matched.group(2)) <= 0.3
github laixintao / iredis / tests / test_utils.py View on Github external
def test_timer():
    with patch("iredis.utils.logger") as mock_logger:
        timer("foo")
        time.sleep(0.1)
        timer("bar")
        mock_logger.debug.assert_called()
        args, kwargs = mock_logger.debug.call_args
        matched = re.match(r"\[timer (\d)\] (0\.\d+) -> bar", args[0])

        assert matched.group(1) == str(3)
        assert 0.1 <= float(matched.group(2)) <= 0.2

        # --- test again ---
        timer("foo")
        time.sleep(0.2)
        timer("bar")
        mock_logger.debug.assert_called()
        args, kwargs = mock_logger.debug.call_args
        matched = re.match(r"\[timer (\d)\] (0\.\d+) -> bar", args[0])

        assert matched.group(1) == str(5)
github laixintao / iredis / iredis / commands_csv_loader.py View on Github external
group2commands = copy.deepcopy(group)

    # add lowercase commands
    group2command_res = {}
    for syntax in group.keys():
        commands = group[syntax]
        lower_commands = [command.lower() for command in commands]
        commands += lower_commands
        # Space in command cloud be mutiple spaces
        re_commands = [command.replace(" ", r"\s+") for command in commands]
        group2command_res[syntax] = "|".join(re_commands)

    return group2commands, group2command_res, command2callback


timer("[Loader] Start loading commands file...")
group2commands, group2command_res, command2callback = load_command()
# all redis command strings, in UPPER case
# NOTE: Must sort by length, to match longest command first
all_commands = sorted(
    list(command2callback.keys()) + ["HELP"] + ["HELP"],
    key=lambda x: len(x),
    reverse=True,
)
commands_summary = load_command_summary()
timer("[Loader] Finished loading commands.")
github laixintao / iredis / iredis / entry.py View on Github external
def repl(client, session, start_time):
    command_holder = UserInputCommand()
    timer(f"First REPL command enter, time cost: {time.time() - start_time}")

    while True:
        logger.info("↓↓↓↓" * 10)
        logger.info("REPL waiting for command...")
        if config.compiling != COMPILING_DONE:
            # auto refresh to display animation...
            _interval = 0.1
        else:
            _interval = None

        try:
            command = session.prompt(
                "{hostname}> ".format(hostname=str(client)),
                bottom_toolbar=BottomToolbar(command_holder).render,
                refresh_interval=_interval,
                input_processors=[GetCommandProcessor(command_holder)],
github laixintao / iredis / iredis / commands_csv_loader.py View on Github external
group2command_res[syntax] = "|".join(re_commands)

    return group2commands, group2command_res, command2callback


timer("[Loader] Start loading commands file...")
group2commands, group2command_res, command2callback = load_command()
# all redis command strings, in UPPER case
# NOTE: Must sort by length, to match longest command first
all_commands = sorted(
    list(command2callback.keys()) + ["HELP"] + ["HELP"],
    key=lambda x: len(x),
    reverse=True,
)
commands_summary = load_command_summary()
timer("[Loader] Finished loading commands.")