Python3学习笔记 本篇为学习廖老师的Python3教程 所做的摘要笔记。有一些自己写归纳的东西，一直在完善中。
安装篇
mac上自带的python是2.x的，先装上3.x再说。
brew install python3
brew link python
* 此时再去link一下
基础语法 字符串 1
2
3
4
5
chr(66)输出66的字符形式
ord与上面相反
格式化
>>> "%s,%s" % ('aaa','bbb')
'aaa,bbb'
list，tuple list
tuple注意
条件判断 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
第一种：
if <条件判断>:
...
else:
...
第二种：
if <条件判断1>:
<执行1>
elif <条件判断2>:
<执行2>
elif <条件判断3>:
<执行3>
else:
<执行4>
字符串不能和int型比较，需要用int()函数转换
循环 for…in… 和while
1
2
3
4
5
6
7
8
9
10
11
12
# for...in..
names = ['Michael', 'Bob', 'Tracy']
for name in names:
print(name)
# while
sum = 0
n = 99
while n > 0:
sum = sum + n
n = n - 2
print(sum)
dict和set 1
2
3
4
5
6
7
8
# dict（其实就是map，键值对）
d = {'Michael': 95, 'Bob': 75, 'Tracy': 85}
d['Tracy'] = 67
# 使用'Thomas' in d判断是否存在key
# set定义
s = set([1, 2, 3])
函数 官方文档
使用help(函数名) 查询函数用法
定义 1
2
3
4
abstest.py
def test(x):
return xxx;
使用from abstest import my_abs引入函数调用
关键字pass 什么也不做
检查类型
1
2
3
4
5
6
7
def test(x):
if not isinstance(x, (int, float)):
raise TypeError('bad operand type')
if x >= 0:
return x
else:
return -x
返回多个值
1
2
3
4
5
6
7
def test(x):
return x1,x2
>>> value = test(2)
这个value是一个tuple。
也可以这么写：
>>> x1,x2 = test(2)
参数默认值
可变参
1
2
3
4
5
6
7
def calc(*test):
相当于java中的...
调用时如果参数是个list，可以直接使用*展开
calc(*list)
关键字参数
1
2
3
4
5
def test(test1, test2, **kw)
kw接受一个dict键值对
比如
test(x,x,key=value, key=value)
关键字参数命名检查
1
2
3
4
5
6
7
8
9
# 自我检查
def person(name, age, **kw):
if 'city' in kw:
# 有city参数
pass
if 'job' in kw:
# 有job参数
pass
print('name:', name, 'age:', age, 'other:', kw)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
如果要限制关键字参数的名字，就可以用命名关键字参数，例如，只接收city和job作为关键字参数。这种方式定义的函数如下：
def person(name, age, *, city, job):
print(name, age, city, job)
使用*隔开。
如果中间有一个可变参，就不要写额外的*了 例如
def person(name, age, *args, city, job):
print(name, age, city, job)
调用时一定要写参数名，
# 调用
person('Jack', 24, city='Beijing', job='Engineer')
参数组合
参数定义的顺序必须是：必选参数、默认参数、可变参数、命名关键字参数和关键字参数。
1
2
def f1(a, b, c=0, *args, **kw):
print('a =', a, 'b =', b, 'c =', c, 'args =', args, 'kw =', kw)
切面取值
有一个list值为[0, 1, 2, 3, …, 99]
迭代
默认情况下，dict迭代的是key，for key in d:。如果要迭代value，可以用for value in d.values()，如果要同时迭代key和value，可以用for k, v in d.items()。
迭代中的下标
1
2
3
4
5
6
>>> for i, value in enumerate(['A', 'B', 'C']):
... print(i, value)
...
0 A
1 B
2 C
上面的for循环里，同时引用了两个变量，在Python里是很常见的，比如下面的代码：
1
2
3
4
5
6
>>> for x, y in [(1, 1), (2, 4), (3, 9)]:
... print(x, y)
...
1 1
2 4
3 9
列表生成器
1
2
3
4
5
[x * x for x in range(1, 11) if x % 2 == 0]
x * x 是结果。中间for...in是迭代。后面的if x % 2 == 0是条件。相当于java的list.filter((x) => x %2 == 0).map((x) => x * x);
[m + n for m in 'ABC' for n in 'XYZ']
多维度组合，这个还是比较有用的，使用了两个迭代生成的新的list
上面的列表生成器直接生成一个list对象。使用()可以产生一个generator，动态的计算下一个元素，而不是全部塞在内存里。
1
2
3
g = (x * x for x in range(10))
next(g) //使用next读取元素
for n in g: //使用迭代读取元素
一个函数中出现yield关键字，那么这个函数的返回值是一个generator，每一次的next都执行到yield返回的地方。只到抛出异常。for…in方式不抛出异常。所以读不到最后的return的值
迭代器
凡是可作用于for循环的对象都是Iterable类型；
凡是可作用于next()函数的对象都是Iterator类型，它们表示一个惰性计算的序列；
集合数据类型如list、dict、str等是Iterable但不是Iterator，不过可以通过iter()函数获得一个Iterator对象。
Python的for循环本质上就是通过不断调用next()函数实现的
1
2
isinstance([], Iterable) 判断是不是Iterable
isinstance([], Iterator) 判断是不是Iterator
函数式编程 map，reduce，filter，sorted 闭包 返回闭包时牢记一点：返回函数不要引用任何循环变量，或者后续会发生变化的变量。
匿名函数 lambda x: x * x，冒号前面的x是参数，后面是表达式。
装饰器 函数也是一个对象，所以，函数也有自己的属性，例如name 就是函数名
1
2
3
4
5
def log(func):
def wrapper(*args, **kw):
print('call %s():' % func.__name__)
return func(*args, **kw)
return wrapper
这个例子，log函数接收一个函数参数func。返回一个函数引用wrapper，在wrapper内先打印了func名字，后执行了func。
1
2
3
@log
def now():
print('2015-3-25')
相当于
下面是一个完整的例子，带参数，带copy函数属性
1
2
3
4
5
6
7
8
9
10
import functools
def log(text):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kw):
print('%s %s():' % (text, func.__name__))
return func(*args, **kw)
return wrapper
return decorator
1
2
3
4
5
6
>>> import functools
>>> int2 = functools.partial(int, base=2)
>>> int2('1000000')
64
>>> int2('1010101')
85
functools.partial的作用就是，把一个函数的某些参数给固定住（也就是设置默认值），返回一个新的函数，调用这个新函数会更简单。
重构时会非常有用。
使用模块 非常简单 import
例子:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
' a test module '
__author__ = 'kahn'
import sys
def test():
args = sys.argv
if len(args)==1:
print('Hello,World!')
elif len(args)==2:
print('Hello, %s!' % args[1])
else:
print('Too many arguments!')
if __name__=='__main__':
test()
安装第三方库 官方插件库
先查找好库名，然后使用pip install xxx 安装
综合插件库anaconda
面向对象编程 语法 class Test(object):
声明变量 test = Test()
可以给对象附加属性，例如，test.param1 = ‘testxxx’
定义类的初始变量 1
2
3
4
5
class Test(object):
def __init__(self, param1, param2):
self.param1 = param1
self.param1 = param2
生成对象时，第一个参数self忽略。
定义类的方法 1
2
3
4
5
6
7
8
class Test(object):
def __init__(self, param1, param2):
self.param1 = param1
self.param1 = param2
def testMethod(self):
print(self.param1);
注意：定义类的方法，第一次参数一定要是self，self为当前对象的实例，调用时无需传入
定义私有变量 1
2
3
4
5
class Test(object):
def __init__(self, param1, param2):
self.__param1 = param1
self.__param1 = param2
在成员变量前加两个下划线就是私有成员变量，外部不可访问
继承，多态 class Test(object)就是继承了。
在子类中可以覆盖父类的方法。
和java不同的一点。如果一个方法接收一个对象，它只使用了这个对象的test方法。那么，传入对象的类型可以是不固定的，只要有test方法就可以了。
isinstance(b, Test) 判断对象类型
对象的信息 type方法
1
2
3
4
5
6
7
8
9
10
11
12
获取对象信息
>>> type(123)
<class 'int'>
做比较
>>> import types
>>> type(123)==type(456)
True
>>> type(123)==int
True
>>> type(fn)==types.FunctionType
True
isinstances方法
1
2
3
类型还可以传多个
>>> isinstance([1, 2, 3], (list, tuple))
True
dir()方法，获取对象的属性及方法，此外还有getattr()、setattr()以及hasattr()，都是进行属性方法操作的。
类属性 1
2
class Test(object):
name = 'Test'
在类外给类附加方法 附加在对象上的方法，只本对象有效
1
2
3
4
5
6
7
8
>>> def set_age(self, age): # 定义一个函数作为实例方法
... self.age = age
...
>>> from types import MethodType
>>> s.set_age = MethodType(set_age, s) # 给实例绑定一个方法
>>> s.set_age(25) # 调用实例方法
>>> s.age # 测试结果
25
附加类方法
1
2
3
4
>>> def set_score(self, score):
... self.score = score
...
>>> Student.set_score = set_score
限制附加属性
1
2
class Student(object):
__slots__ = ('name', 'age') # 用tuple定义允许绑定的属性名称
使用@property使外部可以方便访问变量
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class Person(object):
def __init__(self, name, height):
self.__name = name
self.__height = height
@property
def name(self):
return self.__name
@name.setter
def name(self, name):
self.__name = name
@property
def height(self):
return self.__height
def test():
person = Person("kahn", 180)
print(person.name)
print(person.height)
person.name = 'xxx'
print(person.name)
if __name__=='__main__':
test()
多继承 1
class Dog(Mammal, Runnable):
类的特殊字段方法 __slots__ 限制__str__ 相当于toString __repr__ 开发者模式的toString。 可以这样__repr__ = __str___iter__和__next__ ，第一个返回一个迭代对象，使用迭代对象会调用该对象的__next____getitem__ 像数组一样获取数据__getattr__ 如果找不到属性，该方法将会触发__call__ 把对象变为直接可调用 callable()来判断该对象是不是可调用的
枚举 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from enum import Enum
Month = Enum('Month', ('Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec'))
//这样我们就获得了Month类型的枚举类，可以直接使用Month.Jan来引用一个常量，或者枚举它的所有成员：
//遍历
for name, member in Month.__members__.items():
print(name, '=>', member, ',', member.value)
//扩展，自定义value。（默认枚举的值为从1开始赋值）
//@unique装饰器可以帮助我们检查保证没有重复值。
from enum import Enum, unique
@unique
class Weekday(Enum):
Sun = 0 # Sun的value被设定为0
Mon = 1
Tue = 2
Wed = 3
Thu = 4
Fri = 5
Sat = 6
元类 使用type的方式动态创建类
1
2
3
4
>>> def fn(self, name='world'): # 先定义函数
... print('Hello, %s.' % name)
...
>>> Hello = type('Hello', (object,), dict(hello=fn)) # 创建Hello class
metaclass
相当于使用代码动态构建类，具体还请看这边
错误、调试和测试 异常 python的try catch语法
1
2
3
4
5
6
7
8
9
10
11
12
13
try:
print('try...')
r = 10 / int('2')
print('result:', r)
except ValueError as e:
print('ValueError:', e)
except ZeroDivisionError as e:
print('ZeroDivisionError:', e)
else:
print('no error!')
finally:
print('finally...')
print('END')
Python所有的错误都是从BaseException类派生的 常见异常类
记录异常
使用 import logging logging.exception(e)
抛出异常
使用raise FooError('invalid value: %s' % s)
调试 断言语法
assert n != 0, 'n is zero!'
启动Python解释器时可以用-O参数来关闭assert
logging
import logging
pdb 调试器
启动调试器python -m pdb err.py
单元测试 单测写法
测试方法必须以test开头
1
2
3
4
5
6
7
8
9
10
11
12
//引入单测
import unittest
//引入要测试的类
from mydict import Dict
class TestDict(unittest.TestCase):
def test_init(self):
d = Dict(a=1, b='test')
self.assertEqual(d.a, 1)
self.assertEqual(d.b, 'test')
self.assertTrue(isinstance(d, dict))
执行命令
$ python -m unittest mydict_test
setUp与tearDown
和java一样，同样有这两个方法
写法
1
2
3
4
5
6
7
class TestDict(unittest.TestCase):
def setUp(self):
print('setUp...')
def tearDown(self):
print('tearDown...')
文档测试 在文档中也可按固定格式写出测试代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
def abs(n):
'''
Function to get absolute value of number.
Example:
>>> abs(1)
1
>>> abs(-1)
1
>>> abs(0)
0
'''
return n if n >= 0 else (-n)
IO 文件读写 打开文件
第二个参数是模式，r表示读一个文本，rb表示读一个二进制
1
>>> f = open('/Users/michael/gbk.txt', 'r', encoding='gbk', errors='ignore')
关闭流
第一种，try finally
1
2
3
4
5
6
7
try:
f = open('/path/to/file', 'r')
print(f.read())
finally:
if f:
f.close()
第二种 with语法
1
2
3
with open('/path/to/file', 'r') as f:
print(f.read())
读
read() 读取全部
写
write() 默认‘w’是覆盖，‘a’追加
StringIO和BytesIO StringIO 在内存中读写字符串
1
2
3
4
5
6
7
8
9
10
>>> from io import StringIO
>>> f = StringIO()
>>> f.write('hello')
5
>>> f.write(' ')
1
>>> f.write('world!')
6
>>> print(f.getvalue())
hello world!
BytesIO 在内存中读写byte字节
1
2
3
4
5
6
>>> from io import BytesIO
>>> f = BytesIO()
>>> f.write('中文'.encode('utf-8'))
6
>>> print(f.getvalue())
b'\xe4\xb8\xad\xe6\x96\x87'
OS扩展包（操作文件和目录） 引入OS
1
2
3
>>> import os
>>> os.name # 操作系统类型
'posix'
os.environ获取所有环境变量，os.environ.get(‘key’)获取具体某个环境变量
操作文件和目录
基本操作
1
2
3
4
5
6
7
8
9
10
# 查看当前目录的绝对路径:
>>> os.path.abspath('.')
'/Users/michael'
# 在某个目录下创建一个新目录，首先把新目录的完整路径表示出来:
>>> os.path.join('/Users/michael', 'testdir')
'/Users/michael/testdir'
# 然后创建一个目录:
>>> os.mkdir('/Users/michael/testdir')
# 删掉一个目录:
>>> os.rmdir('/Users/michael/testdir')
split
1
2
3
4
5
>>> os.path.split('/Users/michael/testdir/file.txt')
('/Users/michael/testdir', 'file.txt')
>>> os.path.splitext('/path/to/file.txt')
('/path/to/file', '.txt')
shutil模块提供了copyfile()的函数，你还可以在shutil模块中找到很多实用函数，它们可以看做是os模块的补充
实例：
1
2
3
4
5
6
7
列出当前目录下的所有目录
>>> [x for x in os.listdir('.') if os.path.isdir(x)]
['.lein', '.local', '.m2', '.npm', '.ssh', '.Trash', '.vim', 'Applications', 'Desktop', ...]
列出所有的.py文件
>>> [x for x in os.listdir('.') if os.path.isfile(x) and os.path.splitext(x)[1]=='.py']
['apis.py', 'config.py', 'models.py', 'pymonitor.py', 'test_db.py', 'urls.py', 'wsgiapp.py']
序列化 pickle模块的序列化
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
dumps 用法
>>> import pickle
>>> d = dict(name='Bob', age=20, score=88)
>>> pickle.dumps(d)
dump用法
>>> f = open('dump.txt', 'wb')
>>> pickle.dump(d, f)
>>> f.close()
反序列化
>>> f = open('dump.txt', 'rb')
>>> d = pickle.load(f)
>>> f.close()
>>> d
{'age': 20, 'score': 88, 'name': 'Bob'}
json模块的序列化
1
2
3
4
5
6
7
8
9
10
序列化
>>> import json
>>> d = dict(name='Bob', age=20, score=88)
>>> json.dumps(d)
'{"age": 20, "score": 88, "name": "Bob"}'
反序列化
>>> json_str = '{"age": 20, "score": 88, "name": "Bob"}'
>>> json.loads(json_str)
{'age': 20, 'score': 88, 'name': 'Bob'}
定制json序列化
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def student2dict(std):
return {
'name': std.name,
'age': std.age,
'score': std.score
}
>>> print(json.dumps(s, default=student2dict))
{"age": 20, "name": "Bob", "score": 88}
def dict2student(d):
return Student(d['name'], d['age'], d['score'])
>>> json_str = '{"age": 20, "score": 88, "name": "Bob"}'
>>> print(json.loads(json_str, object_hook=dict2student))
<__main__.Student object at 0x10cd3c190>
进程和线程 进程 os.fork() 创建一个子进程，该方法在父进程和子进程中都会返回值
multiprocessing模块Process类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from multiprocessing import Process
import os
# 子进程要执行的代码
def run_proc(name):
print('Run child process %s (%s)...' % (name, os.getpid()))
if __name__=='__main__':
print('Parent process %s.' % os.getpid())
p = Process(target=run_proc, args=('test',))
print('Child process will start.')
p.start()
p.join()
print('Child process end.')
multiprocessing模块Pool类 进程池
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from multiprocessing import Pool
import os, time, random
def long_time_task(name):
print('Run task %s (%s)...' % (name, os.getpid()))
start = time.time()
time.sleep(random.random() * 3)
end = time.time()
print('Task %s runs %0.2f seconds.' % (name, (end - start)))
if __name__=='__main__':
print('Parent process %s.' % os.getpid())
p = Pool(4)
for i in range(5):
p.apply_async(long_time_task, args=(i,))
print('Waiting for all subprocesses done...')
p.close()
p.join()
print('All subprocesses done.')
对Pool对象调用join()方法会等待所有子进程执行完毕，调用join()之前必须先调用close()，调用close()之后就不能继续添加新的Process了
subprocess模块
1
2
3
4
5
6
import subprocess
print('$ nslookup www.python.org')
r = subprocess.call(['nslookup', 'www.python.org'])
print('Exit code:', r)
进程间通信
Python的multiprocessing模块包装了底层的机制，提供了Queue、Pipes等多种方式来交换数据。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from multiprocessing import Process, Queue
import os, time, random
# 写数据进程执行的代码:
def write(q):
print('Process to write: %s' % os.getpid())
for value in ['A', 'B', 'C']:
print('Put %s to queue...' % value)
q.put(value)
time.sleep(random.random())
# 读数据进程执行的代码:
def read(q):
print('Process to read: %s' % os.getpid())
while True:
value = q.get(True)
print('Get %s from queue.' % value)
if __name__=='__main__':
# 父进程创建Queue，并传给各个子进程：
q = Queue()
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
# 启动子进程pw，写入:
pw.start()
# 启动子进程pr，读取:
pr.start()
# 等待pw结束:
pw.join()
# pr进程里是死循环，无法等待其结束，只能强行终止:
pr.terminate()
线程 ython的标准库提供了两个模块：_thread和threading，_thread是低级模块，threading是高级模块，对_thread进行了封装
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import time, threading
# 新线程执行的代码:
def loop():
print('thread %s is running...' % threading.current_thread().name)
n = 0
while n < 5:
n = n + 1
print('thread %s >>> %s' % (threading.current_thread().name, n))
time.sleep(1)
print('thread %s ended.' % threading.current_thread().name)
print('thread %s is running...' % threading.current_thread().name)
t = threading.Thread(target=loop, name='LoopThread')
t.start()
t.join()
print('thread %s ended.' % threading.current_thread().name)
加锁
1
2
3
4
5
6
7
8
9
10
11
lock = threading.Lock()
def run_thread(n):
for i in range(100000):
# 先要获取锁:
lock.acquire()
try:
# 放心地改吧:
change_it(n)
finally:
# 改完了一定要释放锁:
lock.release()
ThreadLocal
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import threading
# 创建全局ThreadLocal对象:
local_school = threading.local()
def process_student():
# 获取当前线程关联的student:
std = local_school.student
print('Hello, %s (in %s)' % (std, threading.current_thread().name))
def process_thread(name):
# 绑定ThreadLocal的student:
local_school.student = name
process_student()
t1 = threading.Thread(target= process_thread, args=('Alice',), name='Thread-A')
t2 = threading.Thread(target= process_thread, args=('Bob',), name='Thread-B')
分布式进程 1
2
import random, time, queue
from multiprocessing.managers import BaseManager
具体看教程
正则表达式 re模块中含有正则表达式所有的方法
1
2
3
s = 'ABC\\-001' # Python的字符串
s = r'ABC\-001' # Python的字符串 前面+r可以省去转译\
match()方法判断是否匹配，如果匹配成功，返回一个Match对象，否则返回None。常见的判断方法就是：
1
2
3
4
5
test = '用户输入的字符串'
if re.match(r'正则表达式', test):
print('ok')
else:
print('failed')
编译正则，以便重复使用
1
2
3
4
5
6
7
8
>>> import re
# 编译:
>>> re_telephone = re.compile(r'^(\d{3})-(\d{3,8})$')
# 使用：
>>> re_telephone.match('010-12345').groups()
('010', '12345')
>>> re_telephone.match('010-8086').groups()
('010', '8086')
常用内置api datetime
1
from datetime import datetime
collections
1
2
3
4
5
namedtuple
deque
defaultdict
OrderedDict
Counter
struct
用来处理基本数据类型和byte之间的转换
1
2
>>> import struct
>>> struct.pack('>I', 10240099)
hashlib
一些hash算法
contextlib
对with语法的解析
urllib
提供一些访问url的方法
get
1
2
3
4
5
6
7
8
9
from urllib import request
req = request.Request('http://www.douban.com/')
req.add_header('User-Agent', 'Mozilla/6.0 (iPhone; CPU iPhone OS 8_0 like Mac OS X) AppleWebKit/536.26 (KHTML, like Gecko) Version/8.0 Mobile/10A5376e Safari/8536.25')
with request.urlopen(req) as f:
print('Status:', f.status, f.reason)
for k, v in f.getheaders():
print('%s: %s' % (k, v))
print('Data:', f.read().decode('utf-8'))
post
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from urllib import request, parse
print('Login to weibo.cn...')
email = input('Email: ')
passwd = input('Password: ')
login_data = parse.urlencode([
('username', email),
('password', passwd),
('entry', 'mweibo'),
('client_id', ''),
('savestate', '1'),
('ec', ''),
('pagerefer', 'https://passport.weibo.cn/signin/welcome?entry=mweibo&r=http%3A%2F%2Fm.weibo.cn%2F')
])
req = request.Request('https://passport.weibo.cn/sso/login')
req.add_header('Origin', 'https://passport.weibo.cn')
req.add_header('User-Agent', 'Mozilla/6.0 (iPhone; CPU iPhone OS 8_0 like Mac OS X) AppleWebKit/536.26 (KHTML, like Gecko) Version/8.0 Mobile/10A5376e Safari/8536.25')
req.add_header('Referer', 'https://passport.weibo.cn/signin/login?entry=mweibo&res=wel&wm=3349&r=http%3A%2F%2Fm.weibo.cn%2F')
with request.urlopen(req, data=login_data.encode('utf-8')) as f:
print('Status:', f.status, f.reason)
for k, v in f.getheaders():
print('%s: %s' % (k, v))
print('Data:', f.read().decode('utf-8'))
Web开发 1
2
3
4
5
6
7
8
9
10
11
12
13
#application.py
def application(environ, start_response):
start_response('200 ok', [('Content-Type', 'text/html')])
return [b'<h1>hello web</h1>']
#server.py
from wsgiref.simple_server import make_server
from web.application import application
httpd = make_server('', 8080, application)
print('Server Http on port 8080...')
httpd.serve_forever()
异步io 1
2
3
4
5
6
7
8
9
10
11
12
13
import threading
import asyncio
@asyncio.coroutine
def hello():
print('Hello world! (%s)' % threading.currentThread())
yield from asyncio.sleep(1)
print('Hello again! (%s)' % threading.currentThread())
loop = asyncio.get_event_loop()
tasks = [hello(), hello()]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
新老用法
1
2
3
4
5
6
7
8
9
10
11
12
@asyncio.coroutine
def hello():
print("Hello world!")
r = yield from asyncio.sleep(1)
print("Hello again!")
async def hello():
print("Hello world!")
r = await asyncio.sleep(1)
print("Hello again!")
注意新语法只能用在Python 3.5以及后续版本