cf748690-046a-4f45-983b-b5ca63416eda

Build # cf748690-046a-4f45-983b-b5ca63416eda

Build Type

circle-ci

Committed by jarulraj

Commit Message

checkpoint

Pull Request Pull Request #582: server: asyncio refactoring

Run Details

150 of 150 new or added lines in 7 files covered. (100.0%)

7887 of 8618 relevant lines covered (91.52%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

27.69

/eva/server/interpreter.py

# coding=utf-8
# Copyright 2018-2022 EVA
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
import os
import sys
from asyncio import StreamReader, StreamWriter
from collections import deque
from typing import Dict

from eva.server.db_api import EVAConnection
from eva.utils.logging_manager import logger

# version.py defines the VERSION and VERSION_SHORT variables
VERSION_DICT: Dict[str, str] = {}

current_file_dir = os.path.dirname(__file__)
current_file_parent_dir = os.path.join(current_file_dir, os.pardir)
version_file_path = os.path.join(current_file_parent_dir, "version.py")

with open(version_file_path, "r") as version_file:
    exec(version_file.read(), VERSION_DICT)


async def read_line(stdin_reader: StreamReader) -> str:
    delete_char = b"\x7f"
    input_buffer = deque()
    while True:
        input_char = await stdin_reader.read(1)
        if input_char == b";":
            break
        # If the input character is backspace, remove the last character
        if input_char == delete_char:
            if len(input_buffer) > 0:
                input_buffer.pop()
        # Else, append it to the buffer and echo.
        else:
            input_buffer.append(input_char)
    message = b"".join(input_buffer).decode()
    return message


async def create_stdin_reader() -> StreamReader:
    stream_reader = asyncio.StreamReader()
    protocol = asyncio.StreamReaderProtocol(stream_reader)
    loop = asyncio.get_running_loop()
    await loop.connect_read_pipe(lambda: protocol, sys.stdin)
    return stream_reader


async def read_from_client_and_send_to_server(
    stdin_reader: StreamReader, writer: StreamWriter, server_reader: StreamReader
):
    VERSION = VERSION_DICT["VERSION"]
    intro = "eva (v " + VERSION + ')\nType "EXIT;" to exit the client' + "\n"
    sys.stdout.write(intro)
    sys.stdout.flush()

    prompt = "eva=#"

    connection = EVAConnection(server_reader, writer)
    cursor = connection.cursor()

    while True:
        sys.stdout.write(prompt)
        sys.stdout.flush()
        query = await read_line(stdin_reader)
        logger.debug("Query: --|" + query + "|--")

        query = query.lstrip()
        query = query.rstrip()
        if query in ["EXIT", "QUIT"]:
            return

        await cursor.execute_async(query)
        response = await cursor.fetch_all_async()
        sys.stdout.write(str(response) + "\n")
        sys.stdout.flush()


async def start_cmd_client(host: str, port: int):
    """
    Start client
    """
    reader, writer = await asyncio.open_connection(host, port)
    stdin_reader = await create_stdin_reader()

    input_listener = asyncio.create_task(
        read_from_client_and_send_to_server(stdin_reader, writer, reader)
    )

    try:
        await asyncio.wait([input_listener], return_when=asyncio.FIRST_COMPLETED)
    except Exception as e:
        logger.error("Error.", exc_info=e)
        writer.close()
        await writer.wait_closed()
        raise e

1	# coding=utf-8
2	# Copyright 2018-2022 EVA
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15	import asyncio	1✔
16	import os	1✔
17	import sys	1✔
18	from asyncio import StreamReader, StreamWriter	1✔
19	from collections import deque	1✔
20	from typing import Dict	1✔
21
22	from eva.server.db_api import EVAConnection	1✔
23	from eva.utils.logging_manager import logger	1✔
24
25	# version.py defines the VERSION and VERSION_SHORT variables
26	VERSION_DICT: Dict[str, str] = {}	1✔
27
28	current_file_dir = os.path.dirname(__file__)	1✔
29	current_file_parent_dir = os.path.join(current_file_dir, os.pardir)	1✔
30	version_file_path = os.path.join(current_file_parent_dir, "version.py")	1✔
31
32	with open(version_file_path, "r") as version_file:	1✔
33	exec(version_file.read(), VERSION_DICT)	1✔
34
35
36	async def read_line(stdin_reader: StreamReader) -> str:	1✔
37	delete_char = b"\x7f"	×
38	input_buffer = deque()	×
39	while True:	×
40	input_char = await stdin_reader.read(1)	×
41	if input_char == b";":	×
42	break	×
43	# If the input character is backspace, remove the last character
44	if input_char == delete_char:	×
45	if len(input_buffer) > 0:	×
46	input_buffer.pop()	×
47	# Else, append it to the buffer and echo.
48	else:
49	input_buffer.append(input_char)	×
50	message = b"".join(input_buffer).decode()	×
51	return message	×
52
53
54	async def create_stdin_reader() -> StreamReader:	1✔
55	stream_reader = asyncio.StreamReader()	×
56	protocol = asyncio.StreamReaderProtocol(stream_reader)	×
57	loop = asyncio.get_running_loop()	×
58	await loop.connect_read_pipe(lambda: protocol, sys.stdin)	×
59	return stream_reader	×
60
61
62	async def read_from_client_and_send_to_server(	1✔
63	stdin_reader: StreamReader, writer: StreamWriter, server_reader: StreamReader
64	):
65	VERSION = VERSION_DICT["VERSION"]	×
66	intro = "eva (v " + VERSION + ')\nType "EXIT;" to exit the client' + "\n"	×
67	sys.stdout.write(intro)	×
68	sys.stdout.flush()	×
69
70	prompt = "eva=#"	×
71
72	connection = EVAConnection(server_reader, writer)	×
73	cursor = connection.cursor()	×
74
75	while True:	×
76	sys.stdout.write(prompt)	×
77	sys.stdout.flush()	×
78	query = await read_line(stdin_reader)	×
79	logger.debug("Query: --\|" + query + "\|--")	×
80
81	query = query.lstrip()	×
82	query = query.rstrip()	×
83	if query in ["EXIT", "QUIT"]:	×
84	return	×
85
86	await cursor.execute_async(query)	×
87	response = await cursor.fetch_all_async()	×
88	sys.stdout.write(str(response) + "\n")	×
89	sys.stdout.flush()	×
90
91
92	async def start_cmd_client(host: str, port: int):	1✔
93	"""
94	Start client
95	"""
96	reader, writer = await asyncio.open_connection(host, port)	×
97	stdin_reader = await create_stdin_reader()	×
98
99	input_listener = asyncio.create_task(	×
100	read_from_client_and_send_to_server(stdin_reader, writer, reader)
101	)
102
103	try:	×
104	await asyncio.wait([input_listener], return_when=asyncio.FIRST_COMPLETED)	×
105	except Exception as e:	×
106	logger.error("Error.", exc_info=e)	×
107	writer.close()	×
108	await writer.wait_closed()	×
109	raise e	×

georgia-tech-db / eva / cf748690-046a-4f45-983b-b5ca63416eda

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous