基于Torndb的简易ORM
============================================================================
原创作品,允许转载。转载时请务必以超链接形式标明原始出处、以及本声明。
请注明转自:http://yunjianfei.iteye.com/blog/
============================================================================
最近在用tornado写一个基于Rest的WebService服务端,只提供后端服务,其他webserver应用通过URL,Rest的方式来访问。
我们在开发web应用的时候,难免会想到ORM的一些框架,比如java ee中常用的hibernate, ibatis以及python中的SQLAlchemy之类。使用ORM会在一定程度上加快我们的开发效率。
一个简易ORM框架主要实现如下几个功能就足够了:
1.插入: 类对象映射为数据库记录
2.查询:数据库记录映射为类对象
3.修改、删除:可以通过自己写sql语句来搞定。
python中有类,同时也有dict字典类型,如果将字典再包装为类,则显得过渡包装了,反倒很不灵活,因此,提炼一下,python的ORM框架只需要实现如下几点就足够:
1.插入: python的dict映射为数据库记录
2.查询:数据库记录映射为python的dict以及list等
3.修改、删除:可以通过自己写sql语句来搞定。
经过一些测试,技术选型,最终确定了使用tornadb,非常轻量级,查询数据库返回的对象直接映射为python的数据类型dict或者list之类。可以用类似java中“对象.属性”的方式来访问数据。这简直是太爽了~首先,看一个小例子。
import types import time class Row(dict): """A dict that allows for object-like property access syntax.""" def __getattr__(self, name): try: return self[name] except KeyError: raise AttributeError(name) dic = Row() dic.name = 'hello' dic.num = '12334' print type(dic) print "dic.name: " + dic.name print "dic.num: " + dic.num
输出结果为:
dic.name: hello
dic.num: 12334
通过这个例子,我们可以看到,python里面的dict类型,是可以变成类似java中“对象.属性”的方式来访问的。
torndb就是通过这样的方式,查询返回的数据可以通过“.列名”来直接访问。
查询的时候直接返回dict或者list类型,那插入呢?如果可以像java一样,传入一个对象,通过ORM框架直接反射为sql操作,这样多方便啊~
还是dict,如果我们插入的时候,直接将插入的数据格式保存为dict,通过这个dict生成insert语句就可以了,经过查阅各种资料,我提炼出来了如下方法:(使用的时候直接将该方法放入torndb.py中即可)
def insert_by_dict(self, tablename, rowdict, replace=False): cursor = self._cursor() cursor.execute("describe %s" % tablename) allowed_keys = set(row[0] for row in cursor.fetchall()) keys = allowed_keys.intersection(rowdict) if len(rowdict) > len(keys): unknown_keys = set(rowdict) - allowed_keys logging.error("skipping keys: %s", ", ".join(unknown_keys)) columns = ", ".join(keys) values_template = ", ".join(["%s"] * len(keys)) if replace: sql = "REPLACE INTO %s (%s) VALUES (%s)" % ( tablename, columns, values_template) else: sql = "INSERT INTO %s (%s) VALUES (%s)" % ( tablename, columns, values_template) values = tuple(rowdict[key] for key in keys) try: cursor.execute(sql, values) #self._execute(cursor, sql, values, None) return cursor.lastrowid finally: cursor.close()
这样,插入的时候我们就再也不用写繁琐的sql语句了,只需要将我们要插入的对象使用dict封装,比如:
有个host表,里面有hostname,ip两个字段,则我们可以用如下几行代码,就可以插入到数据库:
host = {} host['hostname'] = 'test1' host['ip'] = '10.22.10.90' ret = db.insert_by_dict("Host", host)
是不是很方便呢?:)附件里是我修改过后,完整的torndb源码。欢迎大家多多下载使用。
外带一个小例子,完整版请参照我在github上发布的一个webservice框架:https://github.com/yunfeiflying/tornado-rest-web-service-framwork/
#!/usr/bin/env python2.7 # # -*- coding:utf-8 -*- # # Author : YunJianFei # E-mail : yunjianfei@126.com # Date : 2014/02/25 # Desc : Test db # """ Data Access Object This file impelements DBI for the table 'Host' The Host table's create sql is : CREATE TABLE IF NOT EXISTS `test`.`Host` ( `host_id` INT NOT NULL AUTO_INCREMENT, `host_type` INT NULL, `hostname` VARCHAR(45) NULL, `ip` VARCHAR(45) NULL, `create_time` VARCHAR(45) NULL, `cpu_count` INT NULL, `cpu_pcount` INT NULL, `memory` INT NULL, `os` VARCHAR(200) NULL, `comment` VARCHAR(200) NULL, PRIMARY KEY (`host_id`)) ENGINE = InnoDB; """ from util.dbconst import TableName, TableFields, TableSelectSql import logging class HostDao: def __init__(self, db): mysql_host = "192.168.10.11:3306" db_name = "test" db_user = "root" db_pass = "" self.db = torndb.Connection( host=mysql_host, database=db_name, user=db_user, password=db_pass ) def insert_by_dict(self, host, replace=False): try: id = self.db.insert_by_dict("Host", host, replace) return id except Exception, ex: logging.error("Insert host failed! Exception: %s Host: %s", str(ex), str(host)) return None def if_exist(self, hostname, ip): ret = self.get_by_hostname(hostname) if ret != None: return True ret = self.get_by_ip(ip) if ret != None: return True return False def get_by_ip(self, ip): sql = TableSelectSql.HOST + " where ip='" + str(ip)+"'" return self.db.get(sql) def get_all(self): sql = TableSelectSql.HOST return self.db.query(sql) def get_by_hostname(self, hostname): sql = TableSelectSql.HOST + " where hostname='" + str(hostname)+"'" return self.db.get(sql) def get_by_id(self, host_id): sql = TableSelectSql.HOST + " where host_id=%s" % str(host_id) return self.db.get(sql) def get_id_by_hostname(self, hostname): sql = TableSelectSql.HOST + " where hostname='" + str(hostname)+"'" ret = self.db.get(sql) if ret != None: return ret.host_id return None def update_worker_num_by_hostname(self, hostname, worker_num): try: sql = "UPDATE Host SET worker_num=%s WHERE hostname='%s'" % (worker_num, str(hostname)) ret = self.db.execute(sql) return ret except Exception, ex: logging.error("Update Host failed! Exception: %s hostname: %s , worker_num: %s", str(ex), str(hostname), worker_num) return None def update_worker_num_by_id(self, host_id, worker_num): try: sql = "UPDATE Host SET worker_num=%s WHERE host_id=%s" % (worker_num, host_id) ret = self.db.execute(sql) return ret except Exception, ex: logging.error("Update Host failed! Exception: %s host_id: %s , worker_num: %s", str(ex), host_id, worker_num) return None def del_by_hostname(self, hostname): try: sql = "DELETE FROM Host WHERE hostname='" + str(hostname) + "'" ret = self.db.execute(sql) return ret except Exception, ex: logging.error("Delete host failed! Exception: %s hostname: %s", str(ex), str(hostname)) return None def del_by_id(self, host_id): try: sql = "DELETE FROM Host WHERE host_id=" + str(host_id) ret = self.db.execute(sql) return ret except Exception, ex: logging.error("Delete host failed! Exception: %s host_id: %s", str(ex), host_id) return None
上一篇: 怎样成为优秀的架构师
下一篇: PLSQL中汉字显示乱码问题