Python SQLite的使用经验

SQLite是一款轻量级的数据库,很适合用着移动设备上,或者是客户端程序。SQLite的优点有:1. 不需要为数据库起一个单独的进程 2. 整个数据库可以随时拷贝走 3. 不需要任何配置。从Python 2.5开始,SQLite就在标准库了,所以用起来比较方便。下面是我使用过程中的一些使用经验。

连接到数据库?很简单。

import sqlite3
conn = sqlite3.connect('/tmp/sqlite_db')
cur = conn.cursor()

接下来干嘛呢?建一张表吧。这里需要注意的是,SQLite不支持在创建表的同时创建索引,所以要分两步走,先创建表然后再创建索引

create_table_stmt = '''
    CREATE TABLE IF NOT EXISTS test_table (
    id TEXT,
    duration INTEGER,
    event_date TEXT,
    parameter TEXT
);'''
create_index = 'CREATE INDEX IF NOT EXISTS idx_id ON test_table (id);'
cur.execute(create_table_stmt)
cur.execute(create_index)
conn.commit()

然后往里面插一点数据吧,SQLite只支持5种基本的数据类型


NULL. The value is a NULL value.
INTEGER. The value is a signed integer, stored in 1, 2, 3, 4, 6, or 8 bytes depending on the magnitude of the value.
REAL. The value is a floating point value, stored as an 8-byte IEEE floating point number.
TEXT. The value is a text string, stored using the database encoding (UTF-8, UTF-16BE or UTF-16LE).
BLOB. The value is a blob of data, stored exactly as it was input.

问题来了,SQLite的时间和日期类型在哪里?原来SQLite可以把时间日期保存在一下几种数据类型里面


TEXT as ISO8601 strings (“YYYY-MM-DD HH:MM:SS.SSS”).
REAL as Julian day numbers, the number of days since noon in Greenwich on November 24, 4714 B.C. according to the proleptic Gregorian calendar.
INTEGER as Unix Time, the number of seconds since 1970-01-01 00:00:00 UTC.

insert_stmt = 'insert into test_table values (?, ?, ?, ?)'
record = ('test', 123, '2011-11-30 12:34:56', 'hello')
cur.execute(insert_stmt, record)
conn.commit()

把日期保存为字符串以后,不能直接拿出来直接当日期用,在用之前要调用SQLite的date函数

例如找前一天存进去的数据:

SELECT
    id,
    duration,
    event_date,
    parameter
FROM test_table
WHERE
    date(event_date) = date('now', '-1 day', 'localtime')
ORDER BY id, event_date

SQLite没有show tables,怎么查询数据库里面有什么表?

SELECT name FROM sqlite_master WHERE type = "table"

在web.py中处理表单中的多选下拉框

web.py是一个轻量级的web framework,源代码只有7000多行(我自己用wc -l计算,肯定偏多了),用它来做个简单的站点很方便,上手容易。我在公司就用web.py做了我们自己的自动化测试管理工具,主要是管理自动化用例,展示测试结果。最近遇到一个问题,我想做一个多选下拉框,但是等我做完以后发现结果跟我预想的不太一样。

我用的是web.py的web.form.Dropdown类来生成一个多选下拉框。只需在初始化dropdown的时候传一个multiple=”参数就OK了

form.Dropdown(‘adserver_id’, description=’Adserver: ‘, args=ADSERVERS, multiple=”)

这样生成的html就是



不过当我选中多个选项提交表单的时候,我在服务端得到的结果却是最后一个选中的选项,例如我选了Coretest2 – 1和Coretest3 – 1,那么我在服务端拿到的数据就只有Coretest3 – 1。用Firebug抓了一下请求看,原来POST的数据是长这个样子的:

adserver_id: 1
adserver_id: 2

在web.py中,一般是通过web.input()来获取表单提交的数据。而web.input()会把所有提交的数据给storify方法处理一下(可以更容易的访问dict的数据,例如原来是mydict[‘name’],可以写成mydict.name),而这个方法默认会把adserver_id处理成唯一的key,所以无论输入是多少个adserver_id,最终拿到的只有1个。解决方法很简单,用这个方法来获取post的数据web.input(adserver_id=[]),这样提交adserver_id的多个数据就会被保存到一个列表里面了。

但是这时候会遇到另一个问题,就是如果想要更改这个case的时候,就没有办法在web.py里面用 value=xxx 来指定哪些options是应该被选中的。value=xxx只能指定一个option,怎么办?我是用了JavaScript来解决。思路很简单,就是页面Load完以后,JavaScript获取一下哪些options应该被选中,然后就选上那些项,OK

用Python对体积较大的CSV文件进行比较的经验

最近的工作总是跟数据打交道,需要经常比较一些CSV文件,这些CSV文件其实都需要被LOAD到数据库里面,所以也就是一堆堆的数据文件需要比较。暂时没有发现有比较好用的现成的CSV比较工具,自己动手用Python做了一个凑合能用的。思想比较简单,就是把CSV文件的内容读取出来,保存为一个list,然后把2个CSV文件所生成的list进行对比。有个特殊的需求,就是对于CSV文件中一些肯定不一样的列,例如process date这样的字段,是需要跳过的。由于本地生成的CSV文件比较小,刚开始没有注意到如果文件太大的话会占用很多的内存。最开始的版本是:

def readcsv2list(filename, rows):
    fileobj = open(filename, 'rb')
    csvreader = csv.reader(fileobj)
    retlist = []
    for row in csvreader:
        clist = []
        selected_rows = [ic for ic in range(len(row)) if ic not in rows]
        for c in selected_rows:
            clist.append(row[c])
        retlist.append(clist)

    fileobj.close()
    return retlist

后来用这个脚本比较生产环境数据的时候就遇到问题了,其中最大的一个数据文件大概是1.5GB,这只是文件大小,把文件转成list以后所占用的内存会翻几倍(这个很容易理解,整数1在文件里面站1个字节,放到list里面就要4个字节了)。一下子把机器的内存用光了。随后找了一下文档,csv.reader是没有一个方法可以指定一次读取若干行数据的。后来就利用file object有一个readline()方法,通过一个参数来控制一次读取多少行的记录,从而达到控制内存使用量的目的。需要的注意的点有:1. 在读完若干行数据以后,需要获取一下当前这个file object的位置,Python提供了.tell()方法来获取这个值;2. 读取文件的时候需要知道上一会读到什么地方了,并且从那里继续往下读,用到了.seek()方法;3. readline()方法在读到文件末尾的时候只会返回一个空字符,所以需要对这个空字符做一点处理。

def readcsv2list(filename, rows, last_position, max_line):
    fileobj = open(filename, 'rb')
    fileobj.seek(last_position)
    datalines = []
    for i in range(max_line):
        line_itme = fileobj.readline()
        if len(line_itme) > 0:
            datalines.append(line_itme)
        else:
            break

    csvreader = csv.reader(datalines)
    retlist = []
    for row in csvreader:
        clist = []
        selected_rows = [ic for ic in range(len(row)) if ic not in rows]
        for c in selected_rows:
            clist.append(row[c])
        retlist.append(clist)

    current_position = fileobj.tell()
    fileobj.close()
    return retlist, current_position

Python,尤其是低版本(例如我们用的2.4.3),对于在程序里面显式地del一些变量(通常是个大list之类),是不会立刻释放内存的,所以对于处理数据量比较大的case的时候就需要特别注意内存的使用。参考文章:
Python Memory Management
Why doesn’t Python release the memory when I delete a large object?

DSL与自动化测试 – 用Python实现简单的DSL

自动化测试,一个在测试领域中被广为熟知,也是被谈论最多的概念之一。DSL (Domain Specific Language),一种高度抽象,用于某个特定领域下编程语言。软件测试在大多数情况下都是对某个特定行业的软件系统进行测试,所以这两者应该可以很好的结合起来,事实上也是这样的,QTP里面的keyword view,其实就是DSL的一个实现。DSL一般可以分为两个大的类型,分别是External DSL 和 Internal DSL (引用自Martin Fowler)。External DSL 一般来说是跟其实现语言不一样的 DSL,常见的External DSL 有:SQL和XML配置文件;而Internal DSL 一般来说就是该DSL使用某个现成的编程语言(就是所谓的host language),然后对host language进行一些改造而成。

我们在测试中会遇到很多问题,其中一些问题,几乎是所有公司所有团队都会遇到的,例如测试覆盖率不够,测试的时间不够等等。面对这些问题,自动化测试自然而然地成为解决这些问题的首选方法。但是自动化测试真的就是银弹麽?不见得!以前曾经在ASP.NET QA 的博客中给他们留言,请教过关于自动化测试的事情,我记得其中有一个回复是说,在某个release中过度地使用自动化测试,一切东西都想实现自动化测试,而忽略了产品本身的功能、特性的关注,结果就是超高的自动化测试覆盖率,但是很差的产品质量。大家都去实现自动化测试了,谁来做功能点的覆盖呢?某些领域的专家(SME),他们可能对测试技术是一无所知的,要把这些领域专家和测试实施结合起来,DSL就是一个比较好的桥梁。

我在工作中遇到的问题是,我需要测试一个类似UV(独立用户访问数)统计的系统,统计UV的方法其实就是根据_uid cookie的值来判断这个用户在某段时间内访问过我们的系统多少次,访问了哪些站点,进行了什么样的行为。其中有2个地方比较麻烦,第一就是在测试过程中要不断地拷贝cookie,这样拷来拷去两三次以后很容易就混乱,出错;第二就是需要记录访问哪些站点,这些站点都只是ID,也是需要不断地修改请求,测试时间长了也是很容易出错。所以我就打算在原来的测试工具基础上,实现一个简单的Internal DSL。先看成品:

@tc
def uniq_inventory_case01():
    test= testTool()
    test.user('a').view('asset55100002').anetwork('55100').onsite('site55100503').snetwork('55100').dnetwork('55100').times(1).go()
    test.user('b').view('asset55100002').anetwork('55100').onsite('site55100503').snetwork('55100').dnetwork('55100').times(2).go()
    test.user('b').view('asset55100002').anetwork('55100').onsite('site55100504_noad').snetwork('55100').dnetwork('55100').times(4).go()

实例化一个testTool对象,然后就是指定哪个用户:user(‘a’)或者user(‘b’),看的视频的ID:view(‘asset55100002’),这个视频属于哪个CRO呢?anetwork(‘55100’);放在哪个网站呢?onsite(‘site55100503’);网站是谁的呢?snetwork(‘55100’);谁是分发者呢?dnetwork(‘55100’);看了多少次呢?times(4);最后一个有点儿丑陋的go()。

像这样子一句话里面N个方法连着用,就叫Method Chaining,Method Chaining通常可以让代码变得更加人性化,读起来更加容易。但是使用Method Chaining通常会遇到一个问题,就是很难判断就是到了哪个方法才是终结呢?是不是有些方法的调用是可选的,有些是必选的呢?其中一个解决方法就是我用到的,放一个.go()方法在最后,作为终结方法。要实现Method Chaining,其实只需要顶一个类,对于需要做连接的方法,最后都返回这个类的实例。例如:

def view(self, assetid):
    if assetid:     self.asset_id = assetid
    return self

def anetwork(self, networkid):
    if networkid:   self.a_network_id = networkid
    return self

def snetwork(self, networkid):
    if networkid:   self.s_network_id = networkid
    return self

def dnetwork(self, networkid):
    if networkid:   self.d_network_id = networkid
    return self

def onsite(self, sectionid):
    if sectionid:   self.site_section_id = sectionid
    return self

def times(self, times):
    if times>0:       self.request_times = times
    return self

最后一个终结方法go(),就做真正的处理

def go(self):
    if self.asset_id and self.site_section_id and self.times and self.a_network_id and self.s_network_id:
        self.prepareRequest()
        for i in range(self.request_times):
            self.sendRequest()

        self.cleanup()
    else:
        info = 'Required information missing, abort running.'
        logging.debug(info)
        print info

如果是实现一个External DSL 的话,的确难度不小;但是Internal DSL其实并不是很高深,也不是很难实现,在它的帮助下,可以把工作完成的更好,对自己以后维护测试用例也带来了不少方便。

Python多行注释技巧

Python语言本身是没有注释多行的支持的,如果需要注册多行,可以用一个取巧的方法,就是把需要注释的代码块用三个引号括起来,赋值为一个永远都不会使用的字符串变量,例如:

__devilcomment = '''
if bCmpLog == "True":
        self.appendAdsLogToCmpLog("")

if bCmpBinaryLog == "True":
        self.appendBinaryAdsLogToCmpLog(res)

if bCompareResp == "True":
    self.appendResponseToCmpLog(response_strs)

print "move new and debug logs"
self.tools.move (src_db,dst_db)
'''

用PDB库调试Python程序

如果使用过微软技术的朋友应该体会过微软的Visual Studio系列IDE给debug程序带来的方便,换了个工作就没有Visual Studio了,对于我这种从未在非GUI环境下调试过程序的人来说实在有点不爽,今天花了点时间看了一下Python自带的pdb库,发现用pdb来调试程序还是很方便的,当然了,什么远程调试,多线程之类,pdb是搞不定的。

用pdb调试有多种方式可选:

1. 命令行启动目标程序,加上-m参数,这样调用myscript.py的话断点就是程序的执行第一行之前
python -m pdb myscript.py

2. 在Python交互环境中启用调试
>>> import pdb
>>> import mymodule
>>> pdb.run(‘mymodule.test()’)

3. 比较常用的,就是在程序中间插入一段程序,相对于在一般IDE里面打上断点然后启动debug,不过这种方式是hardcode的

if __name__ == "__main__":
    a = 1
    import pdb
    pdb.set_trace()
    b = 2
    c = a + b
    print (c)

然后正常运行脚本,到了pdb.set_trace()那就会定下来,就可以看到调试的提示符(Pdb)了

常用的调试命令

  • h(elp),会打印当前版本Pdb可用的命令,如果要查询某个命令,可以输入 h [command],例如:“h l” — 查看list命令
  • l(ist),可以列出当前将要运行的代码块

(Pdb) l
497 pdb.set_trace()
498 base_data = {}
499 new_data = {}
500 try:
501 execfile(base_file_name,{},base_data)
502 -> execfile(new_file_name,{},new_data)
503 except:
504 logger.writeLog(“error! load result log error!”)
505 print “load cmp logs error!”
506 raise Exception, “load cmp logs error!”
507

  • b(reak), 设置断点,例如 “b 77″,就是在当前脚本的77行打上断点,还能输入函数名作为参数,断点就打到具体的函数入口,如果只敲b,会显示现有的全部断点

(Pdb) b 504
Breakpoint 4 at /home/jchen/regression/regressionLogCMP.py:504

  • condition bpnumber [condition],设置条件断点,下面语句就是对第4个断点加上条件“a==3”

(Pdb) condition 4 a==3
(Pdb) b
Num Type Disp Enb Where
4 breakpoint keep yes at /home/jchen/regression/regressionLogCMP.py:504
stop only if a==3

  • cl(ear),如果后面带有参数,就是清除指定的断点(我在Python2.4上从来没成功过!!!);如果不带参数就是清除所有的断点

(Pdb) cl
Clear all breaks? y

  • disable/enable,禁用/激活断点

(Pdb) disable 3
(Pdb) b
Num Type Disp Enb Where
3 breakpoint keep no at /home/jchen/regression/regressionLogCMP.py:505

  • n(ext),让程序运行下一行,如果当前语句有一个函数调用,用n是不会进入被调用的函数体中的

Continue reading “用PDB库调试Python程序”