• 35.2k
• 13
• 134
• 238

How can I do this the most pythonic way Model for web applications

• 215
• 1
• 2
• 9

I'm developing model for web application. The database that I use is Redis.

I created special DbField objects like HashDbField, SortedSetDbField, ScalarDbField and so on, for each data type in Redis. This objects provides a convenient way to work with redis keys. Each of it takes target key name in constructor

The second type of objects are DbObjects that represents objects in database. DbObjects (UserDbObject, PostDbObject) consist from DbFields, moreover there is class fields, for some global registers and instance fields for values of certain objects.

Each dbObject has DbObjectName class field with contains current object name ('user', 'post', 'comment', and so on...)

DbObject class looks like:

class DbObjectType(type):
 def __init__(cls, name, bases, dct):
 super(DbObjectType, cls).__init__(name, bases, dct)
 cls._TypeInit()
class DbObject(metaclass=DbObjectType):
 DbObjectName = DB_OBJECT_NAME
 @classmethod
 def _TypeInit(cls):
 cls._SelfDbObjectSerializator = DbObjectSerializator(cls)
 cls.LastId = ScalarDbField(cls.DbObjectName, None, R_LAST_ID)
 cls.Reg = SortedSetDbField(cls.DbObjectName, None, R_REG, cls._SelfDbObjectSerializator, DateTimeSerializator)
 cls._Reg_raw = SortedSetDbField(cls.DbObjectName, None, R_REG)
 @classmethod
 def Create(cls):
 newDbObjectId = cls.LastId.increment()
 dbObject = cls(newDbObjectId)
 dbObject.createDate.set(datetime.now())
 return dbObject
 @classmethod
 def Get(cls, dbObjectId):
 assert dbObjectId
 if not cls._Reg_raw.contains(dbObjectId): return None
 return cls(dbObjectId)
 def delete(self):
 for value in self.__dict__.values():
 if isinstance(value, DbField):
 dbField = value
 dbField.destroy()
 def __init__(self, dbObjectId):
 assert dbObjectId
 self._dbObjectId = str(dbObjectId)
 self.createDate = ScoreDbField(self.DbObjectName, self._dbObjectId, F_CREATE_DATE, self.Reg, self, DateTimeSerializator)
 self.tag = ScalarDbField(self.DbObjectName, self._dbObjectId, F_TAG)
 def __hash__(self):
 return hash(self._dbObjectId)
 def __str__(self):
 return self._dbObjectId
 def __eq__(self, other):
 if type(self) != type(other): return False
 return (self.DbObjectName == other.DbObjectName) and (self._dbObjectId == other._dbObjectId)
 def __ne__(self, other):
 return not self.__eq__(other)
 id = property(lambda self: self._dbObjectId)

As you can see, class fields defining in TypeInit, which is called right after class creates. This trick provides inheritance support, so DbObject's derivative classes will contain LastId and Reg but for its own DbObjectName

Is there way to make this more pythonic? I think this can be implemented whith decorators, but I'm not sure it will be better.

Ok, here are some DbFields:

class DbField():
 _Redis = None
 @classmethod
 def Connect(cls, host=None, port=None):
 cls._Redis = Redis(host or 'localhost', port or 6379, decode_responses =True)
 def __init__(self, dbObjectName, dbObjectId, fieldName, valueSerializator=None):
 assert dbObjectName
 assert fieldName
 self._dbObjectName = dbObjectName
 self._dbObjectId = dbObjectId
 self._fieldName = fieldName
 if self._dbObjectId:
 self._key = "%s:%s:%s" % (self._dbObjectName, self._dbObjectId, self._fieldName)
 else:
 self._key = "%s:%s" % (self._dbObjectName, self._fieldName)
 self._valueSerializator = valueSerializator or Serializator
 def exists(self):
 return self._Redis.exists(self._key)
 def destroy(self):
 return self._Redis.delete(self._key)
class HashDbField(DbField):
 def __init__(self, dbObjectName, dbObjectId, fieldName, valueSerializator=None, nameSerializator=None):
 super(HashDbField, self).__init__(dbObjectName, dbObjectId, fieldName, valueSerializator)
 self._nameSerializator = nameSerializator or Serializator
 def set(self, name, value, overwrite=True):
 return (self._Redis.hset if overwrite else self._Redis.hsetnx)(
 self._key, self._nameSerializator.serialize(name), self._valueSerializator.serialize(value)
 )
 def get(self, name):
 return self._valueSerializator.restore(
 self._Redis.hget(self._key, self._nameSerializator.serialize(name))
 )
 def delete(self, name):
 return self._Redis.hdel(self._key, self._nameSerializator.serialize(name))
class SortedSetDbField(DbField):
 def __init__(self, dbObjectName, dbObjectId, fieldName, valueSerializator=None, scoreSerializator=None):
 super(SortedSetDbField, self).__init__(dbObjectName, dbObjectId, fieldName, valueSerializator)
 self._scoreSerializator = scoreSerializator or Serializator
 def add(self, value, score):
 self._Redis.zadd(self._key, self._valueSerializator.serialize(value), self._scoreSerializator.serialize(score))
 def getRange(self, start=0, end=-1, desc=False):
 return map(self._valueSerializator.restore, self._Redis.zrange(self._key, start=start, end=end, desc=desc))
 def getScore(self, value):
 return self._scoreSerializator.restore(self._Redis.zscore(self._key, self._valueSerializator.serialize(value)))
 def setScore(self, value, score):
 self.add(value, score)
 def indexOf(self, value):
 return self._Redis.zrank(self._key, self._valueSerializator.serialize(value))
 def contains(self, value):
 return self.indexOf(value) is not None
 def delete(self, value):
 self._Redis.zrem(self._key, self._valueSerializator.serialize(value))
 def count(self):
 return self._Redis.zcard(self._key)
 def union(self, sortedSetDbField):
 assert isinstance(sortedSetDbField, SortedSetDbField)
 self._Redis.zunionstore(self._key, (self._key, sortedSetDbField._key), aggregate='max')
 def subtract(self, sortedSetDbField):
 assert isinstance(sortedSetDbField, SortedSetDbField)
 self._Redis.zunionstore(self._key, {self._key: 1, sortedSetDbField._key: -1}, aggregate='min')
 self._Redis.zremrangebyscore(self._key, '-inf', 0)
class ScalarDbField(DbField):
 def set(self, value, overwrite=True):
 return (self._Redis.set if overwrite else self._Redis.setnx)(self._key, self._valueSerializator.serialize(value))
 def get(self):
 return self._valueSerializator.restore(self._Redis.get(self._key))
 def increment(self, amount=1):
 return self._Redis.incr(self._key, amount)
class RefValueDbField(ScalarDbField):
 def __init__(self, dbObjectName, dbObjectId, fieldName, targetHashDbField, targetHashDbFieldValue, valueSerializator=None):
 super(RefValueDbField, self).__init__(dbObjectName, dbObjectId, fieldName, valueSerializator)
 self._targetHashDbField = targetHashDbField
 self._targetHashDbFieldValue = targetHashDbFieldValue
 def set(self, name, overwrite=False):
 if not self._targetHashDbField.set(name, self._targetHashDbFieldValue, overwrite):
 return False
 oldName = self.get()
 super(RefValueDbField, self).set(name)
 if oldName is not None: self._targetHashDbField.delete(oldName)
 return True
 def destroy(self):
 name = self.get()
 if name is not None: self._targetHashDbField.delete(name)
 super(RefValueDbField, self).destroy()
class ScoreDbField(ScalarDbField):
 def __init__(self, dbObjectName, dbObjectId, fieldName, targetSortedSetDbField, targetSortedSetDbFieldValue, valueSerializator=None):
 assert isinstance(targetSortedSetDbField, SortedSetDbField)
 super(ScoreDbField, self).__init__(dbObjectName, dbObjectId, fieldName, valueSerializator)
 self._targetSortedSetDbField = targetSortedSetDbField
 self._targetSortedSetDbFieldValue = targetSortedSetDbFieldValue
 def set(self, score):
 self._targetSortedSetDbField.setScore(self._targetSortedSetDbFieldValue, score)
 super(ScoreDbField, self).set(score)
 def exists(self):
 return self._targetSortedSetDbField.contains(self._targetSortedSetDbFieldValue)
 def destroy(self):
 self._targetSortedSetDbField.delete(self._targetSortedSetDbFieldValue)
 super(ScoreDbField, self).destroy()

I'm in trouble with ScalarDbField, because I have to call get/set methods to get/set values, and I can't use it like descriptor to get/set values like in regular valible, because it mostly used like instance field, and I don't like to use properties with lambdas:

nick = property(lambda self: self._nick.get(), lambda self, value: self._nick.set(value))

Here is UserDbObject

class UserDbObject(DbObject):
 DbObjectName = DB_OBJECT_NAME
 @classmethod
 def _TypeInit(cls):
 super(cls, cls)._TypeInit()
 cls.Nicks = HashDbField(cls.DbObjectName, None, R_NICKS, cls._SelfDbObjectSerializator, nameSerializator=IgnoreCaseSerializator)
 cls.Emails = HashDbField(cls.DbObjectName, None, R_EMAILS, cls._SelfDbObjectSerializator, nameSerializator=IgnoreCaseSerializator)
 cls.AuthKeys = HashDbField(cls.DbObjectName, None, R_AUTH_KEYS, cls._SelfDbObjectSerializator)
 cls.LastLoginDates = SortedSetDbField(cls.DbObjectName, None, R_LAST_LOGIN_DATES, cls._SelfDbObjectSerializator, DateTimeSerializator)
 cls.LastActiveDates = SortedSetDbField(cls.DbObjectName, None, R_LAST_ACTIVE_DATES, cls._SelfDbObjectSerializator, DateTimeSerializator)
 @classmethod
 def Create(cls, nick, email, password, name, surname):
 assert nick
 assert email
 assert password
 userDbObject = super(UserDbObject, cls).Create()
 def setRefFields():
 if not userDbObject.nick.set(nick): return False
 if not userDbObject.email.set(email): return False
 return True
 if not setRefFields():
 userDbObject.delete()
 return None
 userDbObject.generateNewAuthKey()
 userDbObject.setPassword(password)
 if name: userDbObject.name.set(name)
 if surname: userDbObject.surname.set(surname)
 return userDbObject
 @classmethod
 def Get(cls, dbObjectId=None, nick=None, email=None, authKey=None):
 if dbObjectId:
 return super(UserDbObject, cls).Get(dbObjectId)
 if nick:
 return cls.Nicks.get(nick)
 if email:
 return cls.Emails.get(email)
 if authKey:
 return cls.AuthKeys.get(authKey)
 return None
 def setPassword(self, password):
 salt = generateRandomKey(SALT_LENGTH)
 self._passwordHash.set(countCoolHash(password, salt))
 self._salt.set(salt)
 def checkPassword(self, password):
 return self._passwordHash.get() == countCoolHash(password, self._salt.get())
 def generateNewAuthKey(self):
 while True:
 if self.authKey.set(generateRandomKey(AUTH_KEY_LENGTH)): break
 def __init__(self, dbObjectId):
 super(UserDbObject, self).__init__(dbObjectId)
 self.nick = RefValueDbField(self.DbObjectName, self._dbObjectId, F_NICK, self.Nicks, self)
 self.email = RefValueDbField(self.DbObjectName, self._dbObjectId, F_EMAIL, self.Emails, self)
 self.authKey = RefValueDbField(self.DbObjectName, self._dbObjectId, F_AUTH_KEY, self.AuthKeys, self)
 self._passwordHash = ScalarDbField(self.DbObjectName, self._dbObjectId, F_PASSWORD_HASH)
 self._salt = ScalarDbField(self.DbObjectName, self._dbObjectId, F_SALT)
 self.name = ScalarDbField(self.DbObjectName, self._dbObjectId, F_NAME)
 self.surname = ScalarDbField(self.DbObjectName, self._dbObjectId, F_SURNAME)
 self.lastLoginDate = ScoreDbField(self.DbObjectName, self._dbObjectId, F_LAST_LOGIN_DATE, self.LastLoginDates, self, DateTimeSerializator)
 self.lastActiveDate = ScoreDbField(self.DbObjectName, self._dbObjectId, F_LAST_ACTIVE_DATE, self.LastActiveDates, self, DateTimeSerializator)
 self.posts = SortedSetDbField(self.DbObjectName, self._dbObjectId, F_POSTS, self._PostDbObjectSerializator, DateTimeSerializator)
 self.followers = SortedSetDbField(self.DbObjectName, self._dbObjectId, F_FOLLOWERS, self._UserDbObjectSerializator, DateTimeSerializator)
 self.followingUsers = SortedSetDbField(self.DbObjectName, self._dbObjectId, F_FOLLOWING_USERS, self._UserDbObjectSerializator, DateTimeSerializator)
 self.feed = SortedSetDbField(self.DbObjectName, self._dbObjectId, F_FEED, self._PostDbObjectSerializator, DateTimeSerializator)

How can you help me to improve this code?