#758

Committed 04 Sep 2023 08:37PM UTC coverage: 0.0% (-78.3%) from 78.333%

Build # #758

Build Type

push

circle-ci

Committed by

hershd23

Commit Message

Increased underline length in at line 75 in text_summarization.rst
	modified:   docs/source/benchmarks/text_summarization.rst

Run Details

0 of 11303 relevant lines covered (0.0%)

0.0 hits per line

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

0.0

/evadb/catalog/services/column_catalog_service.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.
from typing import List

from sqlalchemy.orm import Session
from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.sql.expression import select

from evadb.catalog.models.column_catalog import ColumnCatalog
from evadb.catalog.models.utils import ColumnCatalogEntry, TableCatalogEntry
from evadb.catalog.services.base_service import BaseService


class ColumnCatalogService(BaseService):
    def __init__(self, db_session: Session):
        super().__init__(ColumnCatalog, db_session)

    def filter_entry_by_table_id_and_name(
        self, table_id, column_name
    ) -> ColumnCatalogEntry:
        entry = self.session.execute(
            select(self.model).filter(
                self.model._table_id == table_id,
                self.model._name == column_name,
            )
        ).scalar_one_or_none()
        if entry:
            return entry.as_dataclass()
        return entry

    def filter_entries_by_table_id(self, table_id: int) -> List[ColumnCatalogEntry]:
        """return all the columns for table table_id"""
        entries = (
            self.session.execute(
                select(self.model).filter(
                    self.model._table_id == table_id,
                )
            )
            .scalars()
            .all()
        )
        return [entry.as_dataclass() for entry in entries]

    def get_entry_by_id(
        self, col_id: int, return_alchemy=False
    ) -> List[ColumnCatalogEntry]:
        entry = self.session.execute(
            select(self.model).filter(self.model._row_id == col_id)
        ).scalar_one_or_none()
        if entry:
            return entry if return_alchemy else entry.as_dataclass()
        return entry

    def insert_entries(self, column_list: List[ColumnCatalogEntry]):
        catalog_column_objs = [
            self.model(
                name=col.name,
                type=col.type,
                is_nullable=col.is_nullable,
                array_type=col.array_type,
                array_dimensions=col.array_dimensions,
                table_id=col.table_id,
            )
            for col in column_list
        ]
        saved_column_objs = []
        for column in catalog_column_objs:
            saved_column_objs.append(column.save(self.session))
        return [obj.as_dataclass() for obj in saved_column_objs]

    def filter_entries_by_table(
        self, table: TableCatalogEntry
    ) -> List[ColumnCatalogEntry]:
        try:
            entries = (
                self.session.execute(
                    select(self.model).filter(self.model._table_id == table.row_id)
                )
                .scalars()
                .all()
            )
            return [entry.as_dataclass() for entry in entries]

        except NoResultFound:
            return None

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	from typing import List	×
16
17	from sqlalchemy.orm import Session	×
18	from sqlalchemy.orm.exc import NoResultFound	×
19	from sqlalchemy.sql.expression import select	×
20
21	from evadb.catalog.models.column_catalog import ColumnCatalog	×
22	from evadb.catalog.models.utils import ColumnCatalogEntry, TableCatalogEntry	×
23	from evadb.catalog.services.base_service import BaseService	×
24
25
26	class ColumnCatalogService(BaseService):	×
27	def __init__(self, db_session: Session):	×
28	super().__init__(ColumnCatalog, db_session)	×
29
30	def filter_entry_by_table_id_and_name(	×
31	self, table_id, column_name
32	) -> ColumnCatalogEntry:
33	entry = self.session.execute(	×
34	select(self.model).filter(
35	self.model._table_id == table_id,
36	self.model._name == column_name,
37	)
38	).scalar_one_or_none()
39	if entry:	×
40	return entry.as_dataclass()	×
41	return entry	×
42
43	def filter_entries_by_table_id(self, table_id: int) -> List[ColumnCatalogEntry]:	×
44	"""return all the columns for table table_id"""
45	entries = (	×
46	self.session.execute(
47	select(self.model).filter(
48	self.model._table_id == table_id,
49	)
50	)
51	.scalars()
52	.all()
53	)
54	return [entry.as_dataclass() for entry in entries]	×
55
56	def get_entry_by_id(	×
57	self, col_id: int, return_alchemy=False
58	) -> List[ColumnCatalogEntry]:
59	entry = self.session.execute(	×
60	select(self.model).filter(self.model._row_id == col_id)
61	).scalar_one_or_none()
62	if entry:	×
63	return entry if return_alchemy else entry.as_dataclass()	×
64	return entry	×
65
66	def insert_entries(self, column_list: List[ColumnCatalogEntry]):	×
67	catalog_column_objs = [	×
68	self.model(
69	name=col.name,
70	type=col.type,
71	is_nullable=col.is_nullable,
72	array_type=col.array_type,
73	array_dimensions=col.array_dimensions,
74	table_id=col.table_id,
75	)
76	for col in column_list
77	]
78	saved_column_objs = []	×
79	for column in catalog_column_objs:	×
80	saved_column_objs.append(column.save(self.session))	×
81	return [obj.as_dataclass() for obj in saved_column_objs]	×
82
83	def filter_entries_by_table(	×
84	self, table: TableCatalogEntry
85	) -> List[ColumnCatalogEntry]:
86	try:	×
87	entries = (	×
88	self.session.execute(
89	select(self.model).filter(self.model._table_id == table.row_id)
90	)
91	.scalars()
92	.all()
93	)
94	return [entry.as_dataclass() for entry in entries]	×
95
96	except NoResultFound:
97	return None

georgia-tech-db / eva / #758

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