e4632a94-24ad-4257-b954-b31674e96477

Committed 17 Aug 2023 11:36PM UTC coverage: 94.379% (-0.9%) from 95.277%

Build # e4632a94-24ad-4257-b954-b31674e96477

Build Type

push

circle-ci

Committed by

gaurav274

Commit Message

pg_handler added

Run Details

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

10242 of 10852 relevant lines covered (94.38%)

1.89 hits per line

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

0.0

/evadb/third_party/databases/postgres/postgres.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 pandas as pd
import psycopg2

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


class PostgresHandler(DBHandler):
    def __init__(self, name: str, **kwargs):
        super().__init__(name)
        self.params = kwargs.get("params", None)

    def connect(self):
        try:
            self.connection = psycopg2.connect(
                host=self.params.get("host"),
                port=self.params.get("port"),
                user=self.params.get("username"),
                password=self.params.get("password"),
                database=self.params.get("database"),
            )
            return DBHandlerStatus(status=True)
        except psycopg2.Error as e:
            return DBHandlerStatus(status=False, error=str(e))

    def disconnect(self):
        if self.connection:
            self.connection.close()

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

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

        try:
            query = "SELECT table_name FROM information_schema.tables WHERE table_schema NOT IN ('information_schema', 'pg_catalog')"
            tables_df = pd.read_sql_query(query, self.connection)
            return DBHandlerResponse(data=tables_df)
        except psycopg2.Error as e:
            return DBHandlerResponse(data=None, error=str(e))

    def get_columns(self, table_name: str) -> DBHandlerResponse:
        if not self.connection:
            return DBHandlerResponse(data=None, error="Not connected to the database.")

        try:
            query = f"SELECT column_name FROM information_schema.columns WHERE table_name='{table_name}'"
            columns_df = pd.read_sql_query(query, self.connection)
            return DBHandlerResponse(data=columns_df)
        except psycopg2.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 pandas as pd	×
16	import psycopg2	×
17
18	from evadb.third_party.databases.types import (	×
19	DBHandler,
20	DBHandlerResponse,
21	DBHandlerStatus,
22	)
23
24
25	class PostgresHandler(DBHandler):	×
26	def __init__(self, name: str, **kwargs):	×
27	super().__init__(name)	×
28	self.params = kwargs.get("params", None)	×
29
30	def connect(self):	×
31	try:	×
32	self.connection = psycopg2.connect(	×
33	host=self.params.get("host"),
34	port=self.params.get("port"),
35	user=self.params.get("username"),
36	password=self.params.get("password"),
37	database=self.params.get("database"),
38	)
39	return DBHandlerStatus(status=True)	×
40	except psycopg2.Error as e:
41	return DBHandlerStatus(status=False, error=str(e))
42
43	def disconnect(self):	×
44	if self.connection:	×
45	self.connection.close()	×
46
47	def check_connection(self) -> DBHandlerStatus:	×
48	if self.connection:	×
49	return DBHandlerStatus(status=True)	×
50	else:
51	return DBHandlerStatus(status=False, error="Not connected to the database.")	×
52
53	def get_tables(self) -> DBHandlerResponse:	×
54	if not self.connection:	×
55	return DBHandlerResponse(data=None, error="Not connected to the database.")	×
56
57	try:	×
58	query = "SELECT table_name FROM information_schema.tables WHERE table_schema NOT IN ('information_schema', 'pg_catalog')"	×
59	tables_df = pd.read_sql_query(query, self.connection)	×
60	return DBHandlerResponse(data=tables_df)	×
61	except psycopg2.Error as e:
62	return DBHandlerResponse(data=None, error=str(e))
63
64	def get_columns(self, table_name: str) -> DBHandlerResponse:	×
65	if not self.connection:	×
66	return DBHandlerResponse(data=None, error="Not connected to the database.")	×
67
68	try:	×
69	query = f"SELECT column_name FROM information_schema.columns WHERE table_name='{table_name}'"	×
70	columns_df = pd.read_sql_query(query, self.connection)	×
71	return DBHandlerResponse(data=columns_df)	×
72	except psycopg2.Error as e:
73	return DBHandlerResponse(data=None, error=str(e))

georgia-tech-db / eva / e4632a94-24ad-4257-b954-b31674e96477

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