Объектное предсталение данных
Добрый день. В этом блоге я хотел бы повести речь о представлении данных в виде набора однотипных объектов. Результат запроса к базе данных может быть представлен списком кортежей, либо в виде списка именованных последовательностей (словарей), которые, впоследствии, используются в приложении, а доступ к элементам одной последовательности происходит по индексу, либо по имени атрибута. Давайте попробуем в качестве результата выборки получить список объектов, обладающих:
- Атрибутами, имена которых совпадают с наименованием полей таблицы (именем колонки в запросе)
- Простыми методами обработки данных
Для сложных запросов применение этого механизма, скорее всего, будет не оправдано, но для работы со справочными данными вполне подойдет. В качестве языва программирования возьмем 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.
В принципе, объекты-контейнеры тоже можно использовать в качестве объектов-атомов. В реализации связей «один-ко-многим» объекты-родители как раз и будут выступать в качестве контейнеров для своих потомков.
