Object Representation

Object presentation of data

Good day. In this blog, I would like to talk about presenting data as a set of objects of the same type. The result of a database query can be represented by a list of tuples, either in the form of a list of named sequences (dictionaries), which are subsequently used in the application, and access to the elements of one sequence occurs by index, or by the attribute name. Let's try as a result of the sample to get a list of objects that have:

• Attributes that have the same name as the table fields (column name in the query)
• Simple data processing methods

For complex queries, the use of this mechanism is likely to be not justified, but it is quite suitable for working with reference data. Let's take python as the programming language.

So let's get started.

Initial conditions

In our database we have a table of settlements. For each entry, a primary key, name, time zone correction, and line activity indicator are defined.

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) ); 

')

Application code

Database Connection Module

We will use the pg canon module to work with 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 () 

Let's sort some methods of the presented class. The SendQueryReturn and SendQueryNoreturn methods are highlighted for convenience, although, in principle, you can dwell on the use of something one. The first method is for queries that return the result of a sample.

It returns, in case of successful execution of the query, the number of rows. The result of the query is stored in the attribute class query_collector.

SendQueryNoreturn, respectively, is designed to execute queries that do not return datasets. If the request is successful, the method will return 0.

In the case of an error, both methods return -1. Error description can be returned to the program using GetError .

The methods GetTupleResult and GetDictResult return the result of the resulting selection as a list of tuples or as a list of named sequences. And one more method to focus on is GetObjectStruct . This method returns a dictionary consisting of the pairs "field name" - "data type".

Query level

I singled out requests to the database in a separate module. The classes used do not know anything about the type of DBMS or database connection library. The use of the pg module was described above, although nothing prevents you from using something else. The main condition is that the database connection level classes should have an identical set of methods.

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

The set_connection function determines the type of connection to the database and imports the corresponding module under the pseudonym connection .
Then there can be one or several classes that hide the query processing mechanisms:

 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':[]} 

In the CustomGet and CustomSet methods, they are a unified add-in on query execution processes, returning results. The return value is the dictionary. The first element of the dictionary is the result of the query. The second is a list of exceptions encountered. The third is the result of the query. The CustomGet method takes the form of the returned data as additional parameters and the processing flag of the empty sample (in some cases this can be perceived as an error).

Requests and their execution

Now we define the query templates for our table.

 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" 

You can use queries explicitly, but I prefer to work with views and stored procedures.

Now you can imagine the mechanism for executing requests at a higher level:

 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) 

Data models

To use the models, we will have to define some constants.

 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 define the base attribute type. It is necessary at the stage of describing the class of the future data model. CAST_TYPES is an attempt to cast the data to the required type during the creation of objects.
The POINTS dictionary is the metadata of our object. The class name is stored here, as well as the names of the methods that will be called when determining the structure of an object, executing the Select, Insert, Update, etc.

Creating objects

At first, I described each type of object obtained from the database as a separate class, but soon I came to a more elegant solution — the use of a class factory. In our example, two types of objects will be created: a container object that will contain a collection of atomic objects, each of which will describe a settlement received from the base.

 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 

The ModelManager class at the initial stage determines the structure of the object being built (BuildModel). If desired, the attributes of the model can be added to your own. The GetInlineMethods and GetCollectMethods methods bind the existing database access functions to future objects. Atom objects will have built-in methods Update and Delete, and container objects will be able to replenish and update their collections using the Insert and Select methods. The CreateClass method is responsible for creating a class , instances of which will be subsequently created by us. Here we will use the classobj function of the module new, which will return us a new class.

Now you can specify the type of database being used, create an instance of the ModelManager class, and call the InitCollection method, which will return an object containing a list of objects from the Points table in the items attribute.
In principle, container objects can also be used as atom objects. In the implementation of one-to-many connections, parent objects will act as containers for their descendants.