Homec4science
Diffusion blackdynamite (features/zeo) Browse

R3127/BlackDynamite/selector.py
840aec0f91fefeatures/zeo

Last Change
Raw File

selector.py
View Options

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# -*- py-which-shell: "python"; -*-
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
################################################################
from __future__ import print_function
################################################################
import copy
from . import constraints as BDcons
################################################################
from . import bdlogging
################################################################
print = bdlogging.invalidPrint
logger = bdlogging.getLogger(__name__)
################################################################
class Selector(object):
def __init__(self, base):
self.base = base
def buildList(self, curs, sqlobjs):
logger.debug(sqlobjs)
if not isinstance(sqlobjs, list):
sqlobjs = [sqlobjs]
col_infos = []
sqlobjs2 = []
for sqlobj in sqlobjs:
if isinstance(sqlobj, type):
sqlobj = sqlobj(self.base)
sqlobjs2.append(sqlobj)
col_infos.append(self.base.getColumnProperties(sqlobj))
sqlobjs = sqlobjs2
list_objects = []
for entries in curs:
# print(entries)
objs = []
offset = 0
logger.debug(sqlobjs)
for index, sqlobj in enumerate(sqlobjs):
obj = copy.deepcopy(sqlobj)
for col_name, size in col_infos[index]:
logger.debug((col_name, entries[offset]))
obj[col_name] = entries[offset]
offset += 1
objs.append(obj)
if len(objs) == 1:
list_objects.append(objs[0])
else:
list_objects.append(tuple(objs))
return list_objects
def select(self, _types, constraints=None, sort_by=None):
if (sort_by is not None) and (not isinstance(sort_by, str)):
raise RuntimeError(
'sort_by argument is not correct: {0}'.format(sort_by))
const = BDcons.BDconstraints(self.base, constraints)
condition, params = const.getMatchingCondition()
if not isinstance(_types, list):
_types = [_types]
selected_tables = ['{0}.{1}'.format(self.base.schema, t.table_name)
for t in _types]
selected_tables = ','.join(selected_tables)
request = "SELECT * FROM {0}".format(selected_tables)
if condition:
request += " WHERE " + condition
# print (sort_by)
if sort_by:
request += " ORDER BY " + sort_by
logger.debug(request)
logger.debug(params)
curs = self.base.performRequest(request, params)
obj_list = self.buildList(curs, _types)
return obj_list
c4science · Help