bf991521-4cdb-4552-b271-e6234b781c4c

Committed 31 Aug 2023 11:50PM UTC coverage: 79.906% (-2.0%) from 81.924%

Build # bf991521-4cdb-4552-b271-e6234b781c4c

Build Type

push

circle-ci

Committed by Ankith Reddy Chitti

Commit Message

sqlite handler

Run Details

51 of 51 new or added lines in 4 files covered. (100.0%)

8987 of 11247 relevant lines covered (79.91%)

1.32 hits per line

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

0.0

/evadb/third_party/databases/sqlite/sqlite_handler.py

# coding=utf-8
# Copyright 2018-2023 EvaDB
#
# 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 sqlite3

import pandas as pd

from evadb.third_party.databases.types import (
    DBHandler,
    DBHandlerResponse,
    DBHandlerStatus,
)


class SQLiteHandler(DBHandler):
    def __init__(self, name: str, **kwargs):
        """
        Initialize the handler.
        Args:
            name (str): name of the DB handler instance
            **kwargs: arbitrary keyword arguments for establishing the connection.
        """
        super().__init__(name)
        self.database = kwargs.get("database")

    def connect(self):
        """
        Set up the connection required by the handler.
        Returns:
            DBHandlerStatus
        """
        try:
            self.connection = sqlite3.connect(
                database=self.database, isolation_level=None  # Autocommit mode.
            )
            return DBHandlerStatus(status=True)
        except sqlite3.Error as e:
            return DBHandlerStatus(status=False, error=str(e))

    def disconnect(self):
        """
        Close any existing connections.
        """
        if self.connection:
            self.connection.close()

    def check_connection(self) -> DBHandlerStatus:
        """
        Check connection to the handler.
        Returns:
            DBHandlerStatus
        """
        if self.connection:
            return DBHandlerStatus(status=True)
        else:
            return DBHandlerStatus(status=False, error="Not connected to the database.")

    def get_tables(self) -> DBHandlerResponse:
        """
        Return the list of tables in the database.
        Returns:
            DBHandlerResponse
        """
        if not self.connection:
            return DBHandlerResponse(data=None, error="Not connected to the database.")

        try:
            query = "SELECT name AS table_name FROM sqlite_master WHERE type = 'table'"
            tables_df = pd.read_sql_query(query, self.connection)
            return DBHandlerResponse(data=tables_df)
        except sqlite3.Error as e:
            return DBHandlerResponse(data=None, error=str(e))

    def get_columns(self, table_name: str) -> DBHandlerResponse:
        """
        Returns the list of columns for the given table.
        Args:
            table_name (str): name of the table whose columns are to be retrieved.
        Returns:
            DBHandlerResponse
        """
        if not self.connection:
            return DBHandlerResponse(data=None, error="Not connected to the database.")
        """
        SQLite does not provide an in-built way to get the column names using a SELECT statement.
        Hence we have to use the PRAGMA command and filter the required columns.
        """
        try:
            query = f"PRAGMA table_info('{table_name}')"
            pragma_df = pd.read_sql_query(query, self.connection)
            columns_df = pragma_df[["name"]].copy()
            columns_df.columns = ["column_name"]
            return DBHandlerResponse(data=columns_df)
        except sqlite3.Error as e:
            return DBHandlerResponse(data=None, error=str(e))

    def _fetch_results_as_df(self, cursor):
        try:
            res = cursor.fetchall()
            res_df = pd.DataFrame(
                res,
                columns=[desc[0] for desc in cursor.description]
                if cursor.description
                else [],
            )
            return res_df
        except sqlite3.ProgrammingError as e:
            if str(e) == "no results to fetch":
                return pd.DataFrame({"status": ["success"]})
            raise e

    def execute_native_query(self, query_string: str) -> DBHandlerResponse:
        """
        Executes the native query on the database.
        Args:
            query_string (str): query in native format
        Returns:
            DBHandlerResponse
        """
        if not self.connection:
            return DBHandlerResponse(data=None, error="Not connected to the database.")
        try:
            cursor = self.connection.cursor()
            cursor.execute(query_string)
            return DBHandlerResponse(data=self._fetch_results_as_df(cursor))
        except sqlite3.Error as e:
            return DBHandlerResponse(data=None, error=str(e))

1	# coding=utf-8
2	# Copyright 2018-2023 EvaDB
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 sqlite3	×
16
17	import pandas as pd	×
18
19	from evadb.third_party.databases.types import (	×
20	DBHandler,
21	DBHandlerResponse,
22	DBHandlerStatus,
23	)
24
25
26	class SQLiteHandler(DBHandler):	×
27	def __init__(self, name: str, **kwargs):	×
28	"""
29	Initialize the handler.
30	Args:
31	name (str): name of the DB handler instance
32	**kwargs: arbitrary keyword arguments for establishing the connection.
33	"""
34	super().__init__(name)	×
35	self.database = kwargs.get("database")	×
36
37	def connect(self):	×
38	"""
39	Set up the connection required by the handler.
40	Returns:
41	DBHandlerStatus
42	"""
43	try:	×
44	self.connection = sqlite3.connect(	×
45	database=self.database, isolation_level=None # Autocommit mode.
46	)
47	return DBHandlerStatus(status=True)	×
48	except sqlite3.Error as e:
49	return DBHandlerStatus(status=False, error=str(e))
50
51	def disconnect(self):	×
52	"""
53	Close any existing connections.
54	"""
55	if self.connection:	×
56	self.connection.close()	×
57
58	def check_connection(self) -> DBHandlerStatus:	×
59	"""
60	Check connection to the handler.
61	Returns:
62	DBHandlerStatus
63	"""
64	if self.connection:	×
65	return DBHandlerStatus(status=True)	×
66	else:
67	return DBHandlerStatus(status=False, error="Not connected to the database.")	×
68
69	def get_tables(self) -> DBHandlerResponse:	×
70	"""
71	Return the list of tables in the database.
72	Returns:
73	DBHandlerResponse
74	"""
75	if not self.connection:	×
76	return DBHandlerResponse(data=None, error="Not connected to the database.")	×
77
78	try:	×
79	query = "SELECT name AS table_name FROM sqlite_master WHERE type = 'table'"	×
80	tables_df = pd.read_sql_query(query, self.connection)	×
81	return DBHandlerResponse(data=tables_df)	×
82	except sqlite3.Error as e:
83	return DBHandlerResponse(data=None, error=str(e))
84
85	def get_columns(self, table_name: str) -> DBHandlerResponse:	×
86	"""
87	Returns the list of columns for the given table.
88	Args:
89	table_name (str): name of the table whose columns are to be retrieved.
90	Returns:
91	DBHandlerResponse
92	"""
93	if not self.connection:	×
94	return DBHandlerResponse(data=None, error="Not connected to the database.")	×
95	"""	×
96	SQLite does not provide an in-built way to get the column names using a SELECT statement.
97	Hence we have to use the PRAGMA command and filter the required columns.
98	"""
99	try:	×
100	query = f"PRAGMA table_info('{table_name}')"	×
101	pragma_df = pd.read_sql_query(query, self.connection)	×
102	columns_df = pragma_df[["name"]].copy()	×
103	columns_df.columns = ["column_name"]	×
104	return DBHandlerResponse(data=columns_df)	×
105	except sqlite3.Error as e:
106	return DBHandlerResponse(data=None, error=str(e))
107
108	def _fetch_results_as_df(self, cursor):	×
109	try:	×
110	res = cursor.fetchall()	×
111	res_df = pd.DataFrame(	×
112	res,
113	columns=[desc[0] for desc in cursor.description]
114	if cursor.description
115	else [],
116	)
117	return res_df	×
118	except sqlite3.ProgrammingError as e:
119	if str(e) == "no results to fetch":
120	return pd.DataFrame({"status": ["success"]})
121	raise e
122
123	def execute_native_query(self, query_string: str) -> DBHandlerResponse:	×
124	"""
125	Executes the native query on the database.
126	Args:
127	query_string (str): query in native format
128	Returns:
129	DBHandlerResponse
130	"""
131	if not self.connection:	×
132	return DBHandlerResponse(data=None, error="Not connected to the database.")	×
133	try:	×
134	cursor = self.connection.cursor()	×
135	cursor.execute(query_string)	×
136	return DBHandlerResponse(data=self._fetch_results_as_df(cursor))	×
137	except sqlite3.Error as e:
138	return DBHandlerResponse(data=None, error=str(e))

georgia-tech-db / eva / bf991521-4cdb-4552-b271-e6234b781c4c

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