Объектное представление данных

Объектное предсталение данных

Добрый день. В этом блоге я хотел бы повести речь о представлении данных в виде набора однотипных объектов. Результат запроса к базе данных может быть представлен списком кортежей, либо в виде списка именованных последовательностей (словарей), которые, впоследствии, используются в приложении, а доступ к элементам одной последовательности происходит по индексу, либо по имени атрибута. Давайте попробуем в качестве результата выборки получить список объектов, обладающих:

• Атрибутами, имена которых совпадают с наименованием полей таблицы (именем колонки в запросе)
• Простыми методами обработки данных

Для сложных запросов применение этого механизма, скорее всего, будет не оправдано, но для работы со справочными данными вполне подойдет. В качестве языва программирования возьмем python.

Итак, приступим.

Начальные условия

В нашей базе данных имеем таблицу населенных пунктов. Для каждой записи определен первичный ключ, наименование, поправка на часовой пояс и признак активности строки.

CREATE TABLE dicts.points
(
  id_point serial NOT NULL, -- код населенного пункта
  "name" character varying(50) DEFAULT 'безымянный'::character varying, -- наименование
  sync_hour integer DEFAULT 0, -- поправка на часовой пояс...
  is_active boolean DEFAULT true, -- признак активности
  CONSTRAINT pk_points PRIMARY KEY (id_point)
);

Код приложения

Модуль подключения к базе данных

Будем использовать каноничный модуль pg для работы с PostgreSQL.

import pg
class Connection:
    """ Подключение к базе данных с использованием DB API2 """
    def __init__(self,dbname,user,password,host,port):
        self.dbname   = dbname
        self.user     = user
        self.password = password
        self.host     = host
        self.port     = port
        
        self.db = None                 # атрибут подключения к базе данных
        self.query_collector = None    # буфер для результатов выборки
        self.err = ''                  # буфер для фиксации ошибок

    def Connect(self):
        """ Подключение к базе данных """
        try:
            self.db = pg.DB(dbname=self.dbname,user=self.user,passwd=self.password,host=self.host,port=self.port)
        except Exception,err:
            raise Exception("Ошибка подключения к базе данных: %s" % err.message)

    def Disconnect(self):
        self.db.close()

    def SendQueryReturn(self,query):
        """ Выполнение SELECT-запросов """
        try:
            self.query_collector = self.db.query(query)
        except ProgrammingError,err:
            self.err = "Объект %s: Не удалось выполнить запрос\n %s" % (__name__,err.message)
            return -1
        else:
            return self.query_collector.ntuples()
        
    def SendQueryNoreturn(self,query):
        """ Выполнение запросов, не возвращающих значение """
        try:
            self.db.query(query)
        except ProgrammingError,err:
            self.err = "Объект %s: Не удалось выполнить запрос\n %s" % (__name__,err.message)
            return -1
        else:
            return 0

    def SendBEGIN(self):
        """ Начало транзакции """
        try:
            self.db.query("BEGIN")
        except ProgrammingError,err:
            self.err = "Объект %s: Не удалось выполнить запрос\n %s" % (__name__,err.message)
            return -1
        else:
            return 0
        
    def SendCOMMIT(self):
        """ Подтверждение транзакции """
        try:
            self.db.query("COMMIT")
        except ProgrammingError,err:
            self.err = "Объект %s: Не удалось завершить транзакцию\n %s" % (__name__,err.message)
            return -1
        else:
            return 0
        
    def SendROLLBACK(self):
        """ Откат транзакции """
        try:
            self.db.query("ROLLBACK")
        except ProgrammingError,err:
            self.err = "Объект %s: Не удалось выполнить откат транзакции\n %s" % (__name__,err.message)
            return -1
        else:
            return 0
        
    def GetTupleResult(self):
        """ Возвращение результата выборки в виде списка """
        try:
            res = self.query_collector.getresult()
        except:
            res = []
        self.query_collector = None
        return res
    
    def GetDictResult(self):
        """ Возвращение результата выборки в виде словаря """
        try:
            res = self.query_collector.dictresult()
        except:
            res = {}
        self.query_collector = None
        return res
    
    def GetError(self):
        """ Возвращение последней записи об ошибке """
        res = self.err
        self.err = ''
        return res
    
    def GetObjectStruct(self,name):
        try:
            return self.db.get_attnames(name)
        except Exception,err:
            self.err = "Объект %s: Не удалось загрузить структуру объекта\n%s"%(__name__,err.message)
            return ()

Разберем некоторые методы представленного класса. Методы SendQueryReturn и SendQueryNoreturn выделены для удобства, хотя, в принципе, можно остановиться на использовании чего-то одного. Первый метод предназначен для запросов, возвращающих результат выборки.

Он возвращает, в случае успешного выполнения запроса, количество строк. Результат запроса сохраняется в атрибут query_collector класса.

SendQueryNoreturn, соответственно, предназначен для выполнения запросов, не возвращающих наборы данных. В случае успешного выполнеия запроса, метод вернет 0.

В случае ошибки оба метода возвращают -1. Описание ошибки можно вернуть в программу при помощи GetError.

Методы GetTupleResult и GetDictResult возвращают результат полученной выборки в виде списка кортежей или в виде списка именованных последовательностей. И еще один метод, на котором стоит заострить внимание — GetObjectStruct. Этот метод возвращает словарь, состоящий из пар «имя поля»-«тип данных».

Уровень запросов

Запросы к базе данных я выделил в отдельный модуль. Используемые классы ничего не знают о типе СУБД или библиотеке подключения к базе данных. Выше было описано использование модуля pg, хотя ничто не мешает использовать что-то еще. Главное условие — классы уровня подключения к базе данных должны иметь идентичный набор методов.

def set_connection(conn):
    global connection
    
    if conn == 'pg':
        import connection.pg_connection as connection
    ...

Функция set_connection определяет тип подключения к базе данных и импортирует соответствующий модуль под псевдонимом connection.
Далее могут идти один или несколько классов, скрывающих механизмы обработки запросов:

from query_constants import *

class QueryLayout:
    def __init__(self,dbname,user,password,host='localhost',port=5432):
        global connection
        self.err_list = []
        if connection:
            self.conn = connection.Connection(dbname,user,password,host,port)
        else:
            self.CONNECTED = False
        self.err_list = []
        self.CONNECTED = self.SetConnection()
        
    def SetConnection(self):
        try:
            self.conn.Connect()
        except Exception,err:
            self.err_list.append(err.message)
            return False
        else:
            return True

    def CloseConnection(self):
        self.conn.Disconnect()

    def QueryNoreturn(self,query):
        """ Обработка результата запроса, не возвращающего значение """
        if self.conn.SendQueryNoreturn(query) == 0:
            return 1
        else:
            self.err_list.append(self.conn.GetError())
            return 0
        
    def QueryReturn(self,query):
        """ Обработка результата запроса, возвращающего значение """
        res = self.conn.SendQueryReturn(query)
        if res < 0:
            self.err_list.append(self.conn.GetError())
        return res
    
    def GetDataTuple(self):
        return self.conn.GetTupleResult()
    
    def GetDataDict(self):
        return self.conn.GetDictResult()
    
    def GetErrors(self):
        res = self.err_list
        self.err_list = []
        return res
    
    def GetObjectStruct(self,objname):
        return self.conn.GetObjectStruct(objname)

class CustomQuery(QueryLayout):
    def __init__(self,dbname,user,password,host='localhost',port=5432):
        QueryLayout.__init__(self,dbname,user,password,host,port)
        
    def BEGIN(self):
        if self.conn.SendBEGIN() < 0:
            err_list = self.conn.GetErrors()
            err_list.insert(0,u"Начало транзакции")
            return False
        return True
    
    def COMMIT(self):
        if self.conn.SendCOMMIT() < 0:
            err_list = self.conn.GetErrors()
            err_list.insert(0,u"Подтверждение транзакции")
            return False
        return True    

    def ROLLBACK(self):
        if self.conn.SendROLLBACK() < 0:
            err_list = self.conn.GetErrors()
            err_list.insert(0,u"Начало транзакции")
            return False
        return True

    def CustomGet(self,query,mode='dict',warn=False):
        nRes = self.QueryReturn(query)
        if nRes > 0:
            if mode == 'tuple':
                res = self.GetDataTuple()
            else:
                res = self.GetDataDict()
            return {'res':len(res),'err':[],'inf':res}
        elif nRes == 0:
            if warn:
                return {'res':-1,'err':[u"Отсутствуют данные"],'inf':[]}
            else:
                return {'res':0,'err':[],'inf':{}}
        else:
            err_list = self.GetErrors()
            err_list.insert(0,u"Ошибка времени выполнения запроса")
            return {'res':-1,'err':err_list,'inf':[]}

    def CustomSet(self,query):
        nRes = self.QueryNoreturn(query)
        if nRes == 1:
            return {'res':0,'err':[],'inf':[]}
        else:
            err_list = self.GetErrors()
            err_list.insert(0,u"Ошибка времени выполнения запроса")
            return {'res':-1,'err':err_list,'inf':[]}

В методах CustomGet и CustomSet являются унифицированной надстройкой над процессами выполнения запросов, возвращения результатов. В качестве возвращаемого значения выступает словарь. Первый элемент словаря — результат запроса. Второй — список возникших исключений. Третий — результат выполнения запроса. Метод CustomGet в качестве дополнительных параметров принимает вид возвращаемых данных и флаг обработки пустой выборки (в некоторых случаях это может восприниматься как ошибка).

Запросы и их выполнение

Теперь определим шаблоны запросов к нашей таблице

ADD_NEW_POINT   = "select * from dicts.insert_point('%s',%s)"
DELETE_POINT    = "select * from dicts.delete_point(%s)"
EDIT_POINT      = "select * from dicts.update_point(%s,'%s',%s)"
GET_ALL_POINTS  = "select * from dicts.get_all_points"

Можно использоваь запросы в явном виде, но я предпочитаю работать с представлениями и хранимыми процедурами.

Теперь можно представить сам механизм исполнения запросов на более высоком уровне:

class QueryCollector(CustomQuery):
    def __init__(self,dbname,user,password,host='localhost',port=5432):
        CustomQuery.__init__(self,dbname,user,password,host,port)
        
    def AddNewPoint(self,sPointName,nHour): return self.CustomSet(ADD_NEW_POINT%(sPointName,nHour))
    
    def DeletePoint(self,nIdPoint): return self.CustomSet(DELETE_POINT%nIdPoint)
    
    def EditPoint(self,nIdPoint,sPointName,nHour): return self.CustomSet(EDIT_POINT%(nIdPoint,sPointName,nHour))
    
    def GetAllPoints(self): return self.CustomGet(GET_ALL_POINTS)

Модели данных

Для использования моделей, нам придется определить некоторые константы.

COLUMN_TYPES = {'int':lambda:int(),'bool':lambda:bool(),'text':lambda:unicode()}
CAST_TYPES   = {'int':lambda x:int(x),'bool':lambda x:bool(x),'text':lambda x:unicode(x,'utf-8')}

POINTS = {'obj_name':'Point','struct':'dicts.get_all_points','select':'GetAllPoints','insert':'AddNewPoint','update':'EditPoint','delete':'DeletePoint'}

CONST_NAMES = {'Points':POINTS}

COLUMN_TYPES определяют базовый тип атрибутов. Он необходим на этапе описания класса будущей модели данных. CAST_TYPES — это попытка привести данные к необходимому типу на этапе создания объектов.
Словарь POINTS — это метаданные нашего объекта. Здесь храниться имя класса, а также имена методов, которые будут вызываться при определении структуры объекта, выполнении методов Select, Insert, Update и пр.

Создание объектов

По началу я каждый тип объекта, получаемого из базы данных, описывал отдельным классом, но вскоре пришел к более изящному решению — применение фабрики классов. В нашем примере будут созданы два типа объектов: объект-контейнер, который будет содержать коллекцию объектов-атомов, каждый из которых будет описывать полученный из базы населенный пункт.

class ModelManager:
    def __init__(self,dbname,user,password,host='localhost',port=5432):
        self.query_collector = query_layout.QueryCollector(dbname,user,password,host,port)
    
    def BuildModel(self,name,**props):
        """ model builder NAME - the name of data struct, props - additional properties """
        c_atts = self.GetMetaData(name) or {}
        struct = self.query_collector.GetObjectStruct(c_atts['struct']) or {}
        dctOptions = {}
        for i in struct.items():
            dctOptions[i[0]] = m_const.COLUMN_TYPES.get(i[1])()
        for i in props.items():
            dctOptions[i[0]] = i[1]
            
        return [struct,dctOptions]
    
    def GetMetaData(self,name):
        """ get meta data for loading struct """
        return m_const.CONST_NAMES.get(name)
    
    def GetInlineMethods(self,name,**methods):
        """ get's methods from QueryCollector object"""
        c_atts = self.GetMetaData(name)
        dctMethods = {}
        if methods:
            dctMethods.update(methods)
        if c_atts:
            try:
                dctMethods['Update'] = getattr(self.query_collector,c_atts['update'])
            except:
                dctMethods['Update'] = lambda:Warning("This method is not implemented!")
            try:
                dctMethods['Delete'] = getattr(self.query_collector,c_atts['delete'])
            except:
                dctMethods['Delete'] = lambda:Warning("This method is not implemented!")
        return dctMethods
    
    def GetCollectMethods(self,name,**methods):
        """ get's methods from QueryCollector to Collection object """
        c_atts = self.GetMetaData(name)
        dctMethods = {}
        if methods:
            dctMethods.update(methods)
        if c_atts:
            try:
                dctMethods['Insert'] = getattr(self.query_collector,c_atts['insert'])
            except:
                dctMethods['Insert'] = lambda:Warning("This method is not implemented!")
            try:
                dctMethods['Select'] = getattr(self.query_collector,c_atts['select'])
            except:
                dctMethods['Select'] = lambda:Warning("This method is not implemented!")
        return dctMethods

    def CreateClass(self,name,i_methods={},props={}):
        """ creates a new object """
        c_atts = self.GetMetaData(name)
        print c_atts
        o_meth = self.GetInlineMethods(name,**i_methods)
        struct,o_prop = self.BuildModel(name,**props)
        obj = classobj(c_atts['obj_name'],(object,),o_meth)
        setattr(obj,'struct',struct)
        for i in o_prop.items():
            setattr(obj,i[0],i[1])
        return obj
    
    def InitObject(self,obj,**values):
        dct_keys = obj.__dict__.keys()
        new_obj = obj()
        for i in values.items():
            if i[0] in dct_keys:
                try:
                    new_obj.__dict__[i[0]] = m_const.CAST_TYPES[new_obj.struct[i[0]]](i[1])
                except:
                    new_obj.__dict__[i[0]] = None
        return new_obj
    
    def InitCollection(self,name,**props):
        o = self.CreateClass(name)
        coll_meth = self.GetCollectMethods(name)
        collection = classobj(name,(object,),coll_meth)()
        if props:
            collection.__dict__.update(props)
        setattr(collection,'items',[])
        dctRes = collection.Select()
        if dctRes['res']>= 0:
            collection.items = [self.InitObject(o,**i) for i in dctRes['inf']]
        return collection    

Класс ModelManager на начальном этапе определяет структуру выстраиваемого объекта (BuildModel). При желании, атрибуты модели можно дополнить своими. Методы GetInlineMethods и GetCollectMethods привязывают существующие функции обращения к базе данных к будущим объектам. Объекты-атомы будут иметь встроенные методы Update и Delete, а объекты-контейнеры смогут пополнять и обновлять свои коллекции при помощи методов Insert и Select. Метод CreateClass ответственен за создание класса, экземпляры которого будут нами впоследствии созданы. Здесь мы воспользуемся функцией classobj модуля new, которая и вернет нам новый класс.

Теперь можно указать тип используемой базы данных, создать экземпляр класса ModelManager, и вызвать метод InitCollection, который вернет объект, содержащий в атрибуте items список объектов из таблицы Points.
В принципе, объекты-контейнеры тоже можно использовать в качестве объектов-атомов. В реализации связей «один-ко-многим» объекты-родители как раз и будут выступать в качестве контейнеров для своих потомков.