Ryan Red HatOpenstack Heat项目组工作,已经用Python写了许多大大小小的应用,包括web、编排计划、后端应用等大小应用。

Python是一种被广泛使用的强大语言,让我们深入这种语言,并且学习一些控制语句的技巧,标准库的窍门和一些常见的陷阱。

1 简介

Python(和它的各种库)非常庞大。它被用于系统自动化、web应用、大数据、数据分析及安全软件。这篇文件旨在展示一些知之甚少的技巧,这些技巧将带领你走上一条开发速度更快、调试更容易并且充满趣味的道路。

学习Python和学习所有其他语言一样,真正有用的资源不是各个语言繁琐的超大官方文档,而是使用常用语法、库和Python社区共享知识的能力。

探索标准数据类型

谦逊的enumerate

遍历在Python中非常简单,使用“for foo in bar:”就可以。

drinks = ["coffee", "tea", "milk", "water"]

for drink in drinks:

print("thirsty for", drink)

#thirsty for coffee

#thirsty for tea

#thirsty for milk

#thirsty for water

drinks = ["coffee", "tea", "milk", "water"]for drink in
drinks:print("thirsty for", drink)#thirsty for coffee#thirsty for
tea#thirsty for milk#thirsty for water

但是同时使用元素的序号和元素本身也是常见的需求。我们经常看到一些程序员使用len()和range()来通过下标迭代列表,但是有一种更简单的方式。

drinks = ["coffee", "tea", "milk", "water"]

for index, drink in enumerate(drinks):

print("Item {} is {}".format(index, drink))

#Item 0 is coffee

#Item 1 is tea

#Item 2 is milk

#Item 3 is water

drinks = ["coffee", "tea", "milk", "water"]for index, drink in
enumerate(drinks):print("Item {} is {}".format(index, drink))#Item 0 is
coffee#Item 1 is tea#Item 2 is milk#Item 3 is water

enumerate 函数可以同时遍历元素及其序号。

Set类型

许多概念都可以归结到对集合(set)的操作。例如:确认一个列表没有重复的元素;查看两个列表共同的元素等等。Python提供了set数据类型以使类似这样的操作更快捷更具可读性。

# deduplicate a list *fast*

print(set(["ham", "eggs", "bacon", "ham"]))

# {'bacon', 'eggs', 'ham'}

# deduplicate a list *fast*print(set(["ham", "eggs", "bacon", "ham"]))# {'bacon', 'eggs', 'ham'}

# compare lists to find differences/similarities

# {} without "key":"value" pairs makes a set

menu = {"pancakes", "ham", "eggs", "bacon"}

new_menu = {"coffee", "ham", "eggs", "bacon", "bagels"}new_items = new_menu.difference(menu)

print("Try our new", ", ".join(new_items))

# Try our new bagels, coffeediscontinued_items = menu.difference(new_menu)

print("Sorry, we no longer have", ", ".join(discontinued_items))

# Sorry, we no longer have pancakes

# compare lists to find differences/similarities# {} without
"key":"value" pairs makes a setmenu = {"pancakes", "ham", "eggs",
"bacon"}new_menu = {"coffee", "ham", "eggs", "bacon", "bagels"}new_items
= new_menu.difference(menu)print("Try our new", ", ".join(new_items))#
Try our new bagels, coffeediscontinued_items =
menu.difference(new_menu)print("Sorry, we no longer have", ",
".join(discontinued_items))# Sorry, we no longer have pancakes

old_items = new_menu.intersection(menu)

print("Or get the same old", ", ".join(old_items))

# Or get the same old eggs, bacon, hamfull_menu = new_menu.union(menu)

print("At one time or another, we've served:", ", ".join(full_menu))

# At one time or another, we've served: coffee, ham, pancakes, bagels, bacon, eggs

old_items = new_menu.intersection(menu)print("Or get the same old",
", ".join(old_items))# Or get the same old eggs, bacon, hamfull_menu =
new_menu.union(menu)print("At one time or another, we've served:", ",
".join(full_menu))# At one time or another, we've served: coffee, ham,
pancakes, bagels, bacon, eggs

intersection 函数比较列表中所有元素,返回两个集合的交集。在我们的例子中,早餐的主食为bacon、eggs和ham。

collections.namedtuple

如果你不想给一个类添加方法,但又想使用foo.prop的调用方式,那么你需要的就是namedtuple。你提前定义好类属性,然后就可以实例化一个轻量级的类,这样的方式会比完整的对象占用更少的内存。

LightObject = namedtuple('LightObject', ['shortname', 'otherprop'])

m = LightObject()

m.shortname = 'athing'

> Traceback (most recent call last):

> AttributeError: can't set attribute

LightObject = namedtuple('LightObject', ['shortname', 'otherprop'])m
= LightObject()m.shortname = 'athing'> Traceback (most recent call
last):> AttributeError: can't set attribute

用这种方式你无法设置namedtuple的属性,正如你不能修改元组(tuple)中元素的值。你需要在实例化namedtuple的时候设置属性的值。

LightObject = namedtuple('LightObject', ['shortname', 'otherprop'])

n = LightObject(shortname='something', otherprop='something else')

n.shortname # something

LightObject = namedtuple('LightObject', ['shortname', 'otherprop'])n
= LightObject(shortname='something', otherprop='something
else')n.shortname # something

collections.defaultdict

在写Python应用使用字典时,很多时候有些关键字一开始并不存在,例如下面的例子。

login_times = {}

for t in logins:

if login_times.get(t.username, None):

login_times[t.username].append(t.datetime)

else:

login_times[t.username] = [t.datetime]

login_times = {}for t in logins:if login_times.get(t.username,
None):login_times[t.username].append(t.datetime)else:login_times[t.username]
= [t.datetime]

使用defaultdict 我们可以跳过检查关键字是否存在的逻辑,对某个未定义key的任意访问,都会返回一个空列表(或者其他数据类型)。

login_times = collections.defaultdict(list)

for t in logins:

login_times[t.username].append(t.datetime)

login_times = collections.defaultdict(list)for t in logins:login_times[t.username].append(t.datetime)

你甚至可以使用自定义的类,这样调用的时候实例化一个类。

from datetime import datetime

class Event(object):

def __init__(self, t=None):

if t is None:

self.time = datetime.now()

else:

self.time = tevents = collections.defaultdict(Event)for e in user_events:

print(events[e.name].time)

from datetime import datetimeclass Event(object):def __init__(self,
t=None):if t is None:self.time = datetime.now()else:self.time = tevents =
collections.defaultdict(Event)for e in
user_events:print(events[e.name].time)

如果既想具有defaultdict的特性,同时还想用访问属性的方式来处理嵌套的key,那么可以了解一下 addict。

normal_dict = {

'a': {

'b': {

'c': {

'd': {

'e': 'really really nested dict'

}

}

}

}

}from addict import Dict

addicted = Dict()

addicted.a.b.c.d.e = 'really really nested'

print(addicted)

# {'a': {'b': {'c': {'d': {'e': 'really really nested'}}}}}

normal_dict = {'a': {'b': {'c': {'d': {'e': 'really really nested
dict'}}}}}from addict import Dictaddicted = Dict()addicted.a.b.c.d.e =
'really really nested'print(addicted)# {'a': {'b': {'c': {'d': {'e':
'really really nested'}}}}}

这段小程序比标准的dict要容易写的多。那么为什么不用defaultdict呢? 它看起来也够简单了。

from collections import defaultdict

default = defaultdict(dict)

default['a']['b']['c']['d']['e'] = 'really really nested dict' # fails

from collections import defaultdictdefault =
defaultdict(dict)default['a']['b']['c']['d']['e'] = 'really really
nested dict' # fails

这段代码看起来没什么问题,但是它最终抛出了KeyError异常。这是因为default[‘a’]是dict,不是defaultdict.让我们构造一个value是defaulted
dictionaries类型的defaultdict,这样也只能解决两级嵌套。

如果你只是需要一个默认计数器,你可以使用collection.Counter,这个类提供了许多方便的函数,例如 most_common.

控制流

当学习Python中的控制结构时,通常要认真学习 for, while,if-elif-else,
和 try-except。只要正确使用,这几个控制结构能够处理绝大多数的情况。也是基于这个原因,几乎你所遇到的所有语言都提供类似的控制结构语句。在基本的控制结构以外,Python也额外提供一些不常用的控制结构,这些结构会使你的代码更具可读性和可维护性。

Great Exceptations

Exceptions作为一种控制结构,在处理数据库、sockets、文件或者任何可能失败的资源时非常常用。使用标准的 try 、except 结构写数据库操作时通常是类型这样的方式。

try:

# get API data

data = db.find(id='foo') # may raise exception

# manipulate the data

db.add(data)

# save it again

db.commit() # may raise exception

except Exception:

# log the failure

db.rollback()db.close()

try:# get API datadata = db.find(id='foo') # may raise exception#
manipulate the datadb.add(data)# save it againdb.commit() # may raise
exceptionexcept Exception:# log the failuredb.rollback()db.close()

你能发现这里的问题吗?这里有两种可能的异常会触发相同的except模块。这意味着查找数据失败(或者为查询数据建立连接失败)会引发回退操作。这绝对不是我们想要的,因为在这个时间点上事务并没有开始。同样回退也不应该是数据库连接失败的正确响应,因此让我们将不同的情况分开处理。

首先,我们将处理查询数据。

try:

# get API data

data = db.find(id='foo') # may raise exception

except Exception:

# log the failure and bail out

log.warn("Could not retrieve FOO")

return# manipulate the data

db.add(data)

try:# get API datadata = db.find(id='foo') # may raise
exceptionexcept Exception:# log the failure and bail outlog.warn("Could
not retrieve FOO")return# manipulate the datadb.add(data)

现在数据检索拥有自己的try-except,这样当我们没有取得数据时,我们可以采取任何处理方式。没有数据我们的代码不大可能再做有用的事,因此我们将仅仅退出函数。除了退出你也可以构造一个默认对象,重新进行检索或者结束整个程序。

现在让我们将commit的代码也单独包起来,这样它也能更优雅的进行错误处理。

try:

db.commit() # may raise exception

except Exception:

log.warn("Failure committing transaction, rolling back")

db.rollback()

else:

log.info("Saved the new FOO")

finally:

db.close()

try:db.commit() # may raise exceptionexcept
Exception:log.warn("Failure committing transaction, rolling
back")db.rollback()else:log.info("Saved the new FOO")finally:db.close()

实际上,我们已经增加了两端代码。首先,让我们看看else,当没有异常发生时会执行这里的代码。在我们的例子中,这里只是将事务成功的信息写入日志,但是你可以按照需要进行更多有趣的操作。一种可能的应用是启动后台任务或者通知。

很明显finally 子句在这里的作用是保证db.close() 总是能够运行。回顾一下,我们可以看到所有和数据存储相关的代码最终都在相同的缩进级别中形成了漂亮的逻辑分组。以后需要进行代码维护时,将很直观的看出这几行代码都是用于完成 commit操作的。

Context and Control

之前,我们已经看到使用异常来进行处理控制流。通常,基本步骤如下:

尝试获取资源(文件、网络连接等)

如果失败,清除留下的所有东西

成功获得资源则进行相应操作

写日志

程序结束

考虑到这一点,让我们再看一下上一章数据库的例子。我们使用try-except-finally来保证任何我们开始的事务要么提交要么回退。

try:

# attempt to acquire a resource

db.commit()

except Exception:

# If it fails, clean up anything left behind

log.warn("Failure committing transaction, rolling back")

db.rollback()

else:

# If it works, perform actions

# In this case, we just log success

log.info("Saved the new FOO")

finally:

# Clean up

db.close()

# Program complete

try:# attempt to acquire a resourcedb.commit()except Exception:# If
it fails, clean up anything left behindlog.warn("Failure committing
transaction, rolling back")db.rollback()else:# If it works, perform
actions# In this case, we just log successlog.info("Saved the new
FOO")finally:# Clean updb.close()# Program complete

我们前面的例子几乎精确的映射到刚刚提到的步骤。这个逻辑变化的多吗?并不多。

差不多每次存储数据,我们都将做相同的步骤。我们可以将这些逻辑写入一个方法中,或者我们可以使用上下文管理器(context manager)

db = db_library.connect("fakesql://")

# as a function

commit_or_rollback(db)# context manager

with transaction("fakesql://") as db:

# retrieve data here

# modify data here

db = db_library.connect("fakesql://")# as a
functioncommit_or_rollback(db)# context managerwith
transaction("fakesql://") as db:# retrieve data here# modify data here

上下文管理器通过设置代码段运行时需要的资源(上下文环境)来保护代码段。在我们的例子中,我们需要处理一个数据库事务,那么过程将是这样的:

阿里云-推广AD

连接数据库

在代码段的开头开始操作

在代码段的结尾提交或者回滚

在代码段的结尾清除资源

让我们建立一个上下文管理器,使用上下文管理器为我们隐藏数据库的设置工作。contextmanager 的接口非常简单。上下文管理器的对象需要具有一个__enter__()方法用来设置所需的上下文环境,还需要一个__exit__(exc_type,
exc_val, exc_tb) 方法在离开代码段之后调用。如果没有异常,那么三个 exc_* 参数将都是None。

此处的__enter__方法非常简单,我们先从这个函数开始。

class DatabaseTransaction(object):

def __init__(self, connection_info):

self.conn = db_library.connect(connection_info)def __enter__(self):

return self.conn

class DatabaseTransaction(object):def __init__(self,
connection_info):self.conn = db_library.connect(connection_info)def
__enter__(self):return self.conn

__enter__方法只是返回数据库连接,在代码段内我们使用这个数据库连接来存取数据。数据库连接实际上是在__init__ 方法中建立的,因此如果数据库建立连接失败,那么代码段将不会执行。

现在让我们定义事务将如何在 __exit__ 方法中完成。这里面要做的工作就比较多了,因为这里要处理代码段中所有的异常并且还要完成事务的关闭工作。

def __exit__(self, exc_type, exc_val, exc_tb):

if exc_type is not None:

self.conn.rollback()try:

self.conn.commit()

except Exception:

self.conn.rollback()

finally:

self.conn.close()

def __exit__(self, exc_type, exc_val, exc_tb):if exc_type is not
None:self.conn.rollback()try:self.conn.commit()except
Exception:self.conn.rollback()finally:self.conn.close()

现在我们就可以使用 DatabaseTransaction 类作为我们例子中的上下文管理器了。在类内部, __enter__ 和 __exit__ 方法将开始和设置数据连接并且处理善后工作。

# context manager

with DatabaseTransaction("fakesql://") as db:

# retrieve data here

# modify data here

# context managerwith DatabaseTransaction("fakesql://") as db:# retrieve data here# modify data here

为了改进我们的(简单)事务管理器,我们可以添加各种异常处理。即使是现在的样子,这个事务管理器已经为我们隐藏了许多复杂的处理,这样你不用每次从数据库拉取数据时都要担心与数据库相关的细节。

生成器

Python 2中引入的生成器(generators)是一种实现迭代的简单方式,这种方式不会一次产生所有的值。Python中典型的函数行为是开始执行,然后进行一些操作,最后返回结果(或者不返回)。

生成器的行为却不是这样的。

def my_generator(v):

yield 'first ' + v

yield 'second ' + v

yield 'third ' + vprint(my_generator('thing'))

# <generator object my_generator at 0x….>

def my_generator(v):yield 'first ' + vyield 'second ' + vyield
'third ' + vprint(my_generator('thing'))# <generator object
my_generator at 0x….>

使用 yield 关键字代替 return ,这就是生成器的独特之处。当我们调用 my_generator('thing') 时,我得到的不是函数的结果而是一个生成器对象,这个生成器对象可以在任何我们使用列表或其他可迭代对象的地方使用。

更常见的用法是像下面例子那样将生成器作为循环的一部分。循环会一直进行,直到生成器停止 yield值。

for value in my_generator('thing'):

print value# first thing

# second thing

# third thinggen = my_generator('thing')

next(gen)

# 'first thing'

next(gen)

# 'second thing'

next(gen)

# 'third thing'

next(gen)

# raises StopIteration exception

for value in my_generator('thing'):print value# first thing# second
thing# third thinggen = my_generator('thing')next(gen)# 'first
thing'next(gen)# 'second thing'next(gen)# 'third thing'next(gen)# raises
StopIteration exception

生成器实例化之后不做任何事直到被要求产生数值,这时它将一直执行到遇到第一个 yield 并且将这个值返回给调用者,然后生成器保存上下文环境后挂起一直到调用者需要下一个值。

现在我们来写一个比刚才返回三个硬编码的值更有用的生成器。经典的生成器例子是一个无穷的斐波纳契数列生成器,我们来试一试。数列从1开始,依次返回前两个数之和。

def fib_generator():

a = 0

b = 1

while True:

yield a

a, b = b, a + b

def fib_generator():a = 0b = 1while True:yield aa, b = b, a + b

函数中的 while True 循环通常情况下应该避免使用,因为这会导致函数无法返回,但是对于生成器却无所谓,只要保证循环中有 yield 。我们在使用这种生成器的时候要注意添加结束条件,因该生成器可以持续不断的返回数值。

现在,使用我们的生成器来计算第一个大于10000的斐波纳契数列值。

min = 10000

for number in fib_generator():

if number > min:

print(number, "is the first fibonacci number over", min)

break

min = 10000for number in fib_generator():if number > min:print(number, "is the first fibonacci number over", min)break

这非常简单,我们可以把数值定的任意大,代码最终都会产生斐波纳契数列中第一个大于X的值。

让我们看一个更实际的例子。翻页接口是应对应用限制和避免向移动设备发送大于50兆JSON数据包的一种常见方法。首先,我们定义需要的API,然后我们为它写一个生成器在我们的代码中隐藏翻页逻辑。

我们使用的API来自Scream,这是一个用户讨论他们吃过的或想吃的餐厅的地方。他们的搜索API非常简单,基本是下面这样。

GET http://scream-about-food.com/search?q=coffee

{

"results": [

{"name": "Coffee Spot",

"screams": 99

},

{"name": "Corner Coffee",

"screams": 403

},

{"name": "Coffee Moose",

"screams": 31

},

{…}

]

"more": true,

"_next": "http://scream-about-food.com/search?q=coffee?p=2"

}

GET http://scream-about-food.com/search?q=coffee{"results":
[{"name": "Coffee Spot", "screams": 99},{"name": "Corner Coffee",
"screams": 403},{"name": "Coffee Moose", "screams": 31},{…}]"more":
true,"_next": "http://scream-about-food.com/search?q=coffee?p=2"}

他们将下一页的链接嵌入到API应答中,这样当需要获得下一页时就非常简单了。我们能够不考虑页码,只是获取第一页。为了获得数据,我们将使用常见的 requests 库,并且用生成器将其封装以展示我们的搜索结果。

这个生成器将处理分页并且限制重试逻辑,它将按照下述逻辑工作:

收到要搜索的内容

查询scream-about-food接口

如果接口失败进行重试

一次yield一个结果

如果有的话,获取下一页

当没有更多结果时,退出

非常简单。我来实现这个生成器,为了简化代码我们暂时不考虑重试逻辑。

import requestsapi_url = "http://scream-about-food.com/search?q={term}"def infinite_search(term):

url = api_url.format(term)

while True:

data = requests.get(url).json()for place in data['results']:

yield place# end if we've gone through all the results

if not data['more']: breakurl = data['_next']

import requestsapi_url =
"http://scream-about-food.com/search?q={term}"def
infinite_search(term):url = api_url.format(term)while True:data =
requests.get(url).json()for place in data['results']:yield place# end if
we've gone through all the resultsif not data['more']: breakurl =
data['_next']

当我们创建了生成器,你只需要传入搜索的内容,然后生成器将会生成请求,如果结果存在则获取结果。当然这里有些未处理的边界问题。异常没有处理,当API失败或者返回了无法识别的JSON,生成器将抛出异常。

尽管存在这些未处理完善的地方,我们仍然能使用这些代码获得我们的餐厅在关键字“coffee”搜索结果中的排序。

# pass a number to start at as the second argument if you don't want

# zero-indexing

for number, result in enumerate(infinite_search("coffee"), 1):

if result['name'] == "The Coffee Stain":

print("Our restaurant, The Coffee Stain is number ", number)

return

print("Our restaurant, The Coffee Stain didnt't show up at all! :(")

# pass a number to start at as the second argument if you don't
want# zero-indexingfor number, result in
enumerate(infinite_search("coffee"), 1):if result['name'] == "The Coffee
Stain":print("Our restaurant, The Coffee Stain is number ",
number)returnprint("Our restaurant, The Coffee Stain didnt't show up at
all! :(")

如果使用Python 3,当你使用标准库时你也能使用生成器。调用类似 dict.items() 这样的函数时,不返回列表而是返回生成器。在Python 2中为了获得这种行为,Python 2中添加了 dict.iteritems() 函数,但是用的比较少。

Python 2 and 3 compatibility

从Python 2 迁移到Python
3对任何代码库(或者开发人员)都是一项艰巨的任务,但是写出两个版本都能运行的代码也是可能的。Python
2.7将被支持到2020年,但是许多新的特性将不支持向后兼容。目前,如果你还不能完全放弃Python 2, 那最好使用Python 2.7 和
3+兼容的特性。

对于两个版本支持特性的全面指引,可以在python.org上看 Porting Python 2 Code 。

让我们查看一下在打算写兼容代码时,你将遇到的最常见的情况,以及如何使用 __future__ 作为变通方案。

print or print()

几乎每一个从Python 2 切换到Python 3的开发者都会写出错误的print 表达式。幸运的是,你能够通过导入 print_function 模块,将print作为一个函数而不是一个关键字来写出可兼容的print.

for result in infinite_search("coffee"):

if result['name'] == "The Coffee Stain":

print("Our restaurant, The Coffee Stain is number ", result['number'])

return

print("Our restaurant, The Coffee Stain didn't show up at all! :(")

for result in infinite_search("coffee"):if result['name'] == "The
Coffee Stain":print("Our restaurant, The Coffee Stain is number ",
result['number'])returnprint("Our restaurant, The Coffee Stain didn't
show up at all! :(")

Divided Over Division

从Python 2 到 Python 3,除法的默认行为也发生了变化。在Python 2中,整数的除法只进行整除,小数部分全部截去。大多数用户不希望这样的行为,因此在Python 3中即使是整数之间的除法也执行浮点除。

print "hello"# Python 2

print("hello") # Python 3from __future__ import print_function

print("hello") # Python 2

print("hello") # Python 3

print "hello"# Python 2print("hello") # Python 3from __future__
import print_functionprint("hello") # Python 2print("hello") # Python 3

这种行为的改变会导致编写同时运行在Python 2 和 Python 3上的代码时,带来一连串的小bug。我们再一次需要 __future__ 模块。导入 division 将使代码在两个版本中产生相同的运行结果。

print(1 / 3) # Python 2

# 0

print(1 / 3) # Python 3

# 0.3333333333333333

print(1 // 3) # Python 3

# 0

print(1 / 3) # Python 2# 0print(1 / 3) # Python 3# 0.3333333333333333print(1 // 3) # Python 3# 0

最后-感谢阅读

感谢阅读,我希望你至少能学到了一点东西。假如你想要添加内容(或者纠正内容,没有哪个作者是完美的)请留言,我会经常来看评论的。如果你喜欢这篇文章,你可能需要阅读 list 和 dict comprehensions 或者关于 Python
2 和 3的更深刻的理解。