Python高级编程阅读笔记（四）

2to3 xxx.py #仅仅提供建议，不会更改文件
2to3 -w xxx.py #更改文件

def calculate_age_at_wedding(person_id):
    """计算一个人结婚时的年龄，传递一个此人在数据库中的ID"""
    # 从数据库中查询一个人
    # 并且anniversary 是 datetime.date 对象.
    person = get_person_from_db(person_id)
    anniversary = person['anniversary']
    birthday = person['birthday']
    # 计算结婚时的年龄
    age = anniversary.year – birthday.year
    # 如果生日晚于周年纪念日，那么从年龄减去一个
    if birthday.replace(year=anniversary.year) > anniversary:
        age -= 1
    # 完成返回年龄
    return age

def calculate_age_at_wedding(person):
    """计算一个人结婚时的年龄，传递一个此人的的数据，类型为dict-like"""
    # 从数据库中查询一个人
    # 并且anniversary 是 datetime.date 对象.
    anniversary = person['anniversary']
    birthday = person['birthday']
    # 计算结婚时的年龄
    age = anniversary.year - birthday.year
    # 如果生日晚于周年纪念日，那么从年龄减去一个
    if birthday.replace(year=anniversary.year) > anniversary:
        age -= 1
    # 完成返回年龄
    return age

from datetime import date

person = {
    'anniversary' : date(2012, 4, 21),
    'birthday' : date(1986, 6, 15)
}

age = calculate_age_at_wedding(person)
age

25

def test_calculate_age_at_wedding():
    """calculate_age_at_wedding的测试函数"""
    # 如果结婚月日在一个生日月日之前
    person = {'anniversary': date(2012, 4, 21),
            'birthday': date(1986, 6, 15)}
    age = calculate_age_at_wedding(person)
    assert age == 25, 'Expected age 25, got %d.' % age
    # 如果结婚月日在一个生日月日之后
    person = {'anniversary': date(1969, 8, 11),
    'birthday': date(1945, 2, 15)}
    age = calculate_age_at_wedding(person)
    assert age == 24, 'Expected age 24, got %d.' % age

test_calculate_age_at_wedding()

python -m unittest wedding

.
----------------------------------------------------------------------
Ran 1 test in 0.000s
OK

def test_failure_case(self):
    """Assert a wrong age, and fail."""
    person = {'anniversary': date(2012, 4, 21),
    'birthday': date(1986, 6, 15)}
    age = calculate_age_at_wedding(person)
    self.assertEqual(age, 99)

.F
======================================================================
FAIL: test_failure_case (wedding.Tests)
Assert a wrong age, and fail.
----------------------------------------------------------------------
Traceback (most recent call last):
File "wedding.py", line 50, in test_failure_case
self.assertEqual(age, 99)
AssertionError: 25 != 99

def test_error_case(self):
    """Attempt to send an empty dict to the function."""
    person = {}
    age = calculate_age_at_wedding(person)
    self.assertEqual(age, 25)

.EF
======================================================================
ERROR: test_error_case (wedding.Tests)
Attempt to send an empty dict to the function.
----------------------------------------------------------------------
Traceback (most recent call last):
File "wedding.py", line 55, in test_error_case
age = calculate_age_at_wedding(person)
File "wedding.py", line 10, in calculate_age_at_wedding
anniversary = person['anniversary']
KeyError: 'anniversary'
======================================================================
FAIL: test_failure_case (wedding.Tests)
Assert a wrong age, and fail.
----------------------------------------------------------------------
Traceback (most recent call last):
File "wedding.py", line 50, in test_failure_case
self.assertEqual(age, 99)
AssertionError: 25 != 99
----------------------------------------------------------------------
Ran 3 tests in 0.000s
FAILED (failures=1, errors=1)

@unittest.skipIf(True, 'This test was skipped.')
def test_skipped_case(self):
    """Skip this test."""
    pass

.s
----------------------------------------------------------------------
Ran 2 tests in 0.000s
OK (skipped=1)

##########将要被测试的代码##########
def get_person_from_db(person_id):
    raise RuntimeError('The real `get_person_from_db` function was called.')

def calculate_age_at_wedding(person_id):
    """计算一个人结婚时的年龄，传递一个此人在数据库中的ID"""
    # 从数据库中查询一个人
    # 并且anniversary 是 datetime.date 对象.
    person = get_person_from_db(person_id)
    anniversary = person['anniversary']
    birthday = person['birthday']
    # 计算结婚时的年龄
    age = anniversary.year - birthday.year
    # 如果生日晚于周年纪念日，那么从年龄减去一个
    if birthday.replace(year=anniversary.year) > anniversary:
        age -= 1
    # 完成返回年龄
    return ag

##########测试代码##########
import unittest
import sys
from datetime import date
# Import mock regardless of whether it is from the standard library
# or from the PyPI package.
try:
    from unittest import mock
except ImportError:
    import mock

class Tests(unittest.TestCase):
    def test_calculate_age_at_wedding(self):
        """测试一个函数`calculate_age_at_wedding`该函数需要一个参数person_id
        """

        # 在这里测试和逻辑在同一个模块所以要导入使用`__name__`
        # 在一般情况下，测试和代码分离
        # 这个模块发送到'mock.patch.object`
        module = sys.modules[__name__]
        with mock.patch.object(module, 'get_person_from_db') as m:
            # 确保get_person_from_db函数返回一个有效的字典。
            m.return_value = {'anniversary': date(2012, 4, 21),
                                'birthday': date(1986, 6, 15)}
            # 执行函数，并断言测试
            age = calculate_age_at_wedding(person_id=42)
            self.assertEqual(age, 25)

F
======================================================================
FAIL: test_calculate_age_at_wedding (wedding.Tests)
Establish that the 'calculate_age_at_wedding' function seems
----------------------------------------------------------------------
Traceback (most recent call last):
File "/Users/luke/Desktop/wiley/wedding.py", line 58, in
test_calculate_age_at_wedding
m.assert_called_once_with(84)
File
"/Library/Frameworks/Python.framework/Versions/3.4/lib/python3.4/unittest
/mock.py", line 771, in assert_called_once_with
return self.assert_called_with(*args, **kwargs)
File
"/Library/Frameworks/Python.framework/Versions/3.4/lib/python3.4/unittest
/mock.py", line 760, in assert_called_with
raise AssertionError(_error_message()) from cause
AssertionError: Expected call: get_person_from_db(84)
Actual call: get_person_from_db(42)
----------------------------------------------------------------------
Ran 1 test in 0.001s

# 检查是否被调用过
from unittest import mock
m = mock.MagicMock()
print(m.called)
m(foo='bar')
print(m.called)

False
True

# 检查调用次数
from unittest import mock
m = mock.MagicMock()
print(m.call_count)
m(foo='bar')
print(m.call_count)
m(spam='eggs')
print(m.call_count)

0
1
2

from unittest.mock import call
a = call(42)
b = call(42)
c = call('foo')
print(a is b)
print(a == b)
print(a == c)

False
True
False

#检查是否调用了这个签名
from unittest.mock import MagicMock, call
m = MagicMock()
m.call('a')
m.call('b')
m.call('c')
m.call('d')
m.assert_has_calls([call.call('b'), call.call('c')])

from unittest.mock import call
c = call('foo', 'bar', spam='eggs')
print(c[0])
print(c[1])
print(c[2])

('foo', 'bar')
{'spam': 'eggs'}

assert c[2]['spam'] == 'eggs'
assert 'baz' in c[1]

---------------------------------------------------------------------------

AssertionError                            Traceback (most recent call last)

<ipython-input-19-a138eabaca5b> in <module>()
      1 assert c[2]['spam'] == 'eggs'
----> 2 assert 'baz' in c[1]

AssertionError:

import optparse

if __name__ == '__main__':
    parser = optparse.OptionParser()
    options, args = parser.parse_args()

    print(' '.join(args).upper())

import optparse
if __name__ == '__main__':
    parser = optparse.OptionParser()
    parser.add_option('-q', '--quiet',
        action='store_true',
        dest='quiet',
        help='Suppress output.',
    )

import optparse

if __name__ == '__main__':
    parser = optparse.OptionParser()
    parser.add_option('-H', '--host',
        default='localhost',
        dest='host',
        help='The host to connect to. Defaults to localhost.',
        type=str,
    )
    options, args = parser.parse_args()

parser.add_option('-p', '--port',
    default=5432,
    dest='port',
    help='The port to connect to. Defaults to 5432.',
    type=int,
)

import optparse

if __name__ == '__main__':
    parser = optparse.OptionParser()

    parser.add_option('-H', '--host',
        default='localhost',
        dest='host',
        help='The host to connect to. Defaults to localhost.',
        type=str,
    )
    parser.add_option('-p', '--port',
        default=5432,
        dest='port',
        help='The port to connect to. Defaults to 5432.',
        type=int,
    )

    options, args = parser.parse_args()

    print('The host is %s, and the port is %d.' %
        (options.host, options.port))

python optparse_host_and_port.py -H localhost
python optparse_host_and_port.py -H "localhost"
python optparse_host_and_port.py -Hlocalhost
python optparse_host_and_port.py -H"localhost"

python optparse_host_and_port.py -H=localhost
The host is =localhost, and the port is 5432.

python optparse_host_and_port.py -H="localhost"
The host is =localhost, and the port is 5432.

python cli_script.py --host localhost
python cli_script.py --host "localhost"
python cli_script.py --host=localhost
python cli_script.py --host="localhost"

import optparse

if __name__ == '__main__':
    parser = optparse.OptionParser()
    options, args = parser.parse_args()

    print('The sum of the numbers sent is: %d' %
        sum([int(i) for i in args]))

python optparse_sum.py 1 2 5
The sum of the numbers sent is: 8

if __name__ == '__main__':
    parser = optparse.OptionParser()
    parser.add_option('-v',
        action='count',
        default=0,
        dest='verbosity',
        help='Be more verbose. This flag may be repeated.',
    )

    options, args = parser.parse_args()
    print('The verbosity level is %d, ah ah ah.' % options.verbosity)

python count_script.py
The verbosity level is 0, ah ah ah.
python count_script.py -v
The verbosity level is 1, ah ah ah.
python count_script.py -v -v
The verbosity level is 2, ah ah ah.
python count_script.py -vvvvvvvvvvv
The verbosity level is 11, ah ah ah.

import optparse

if __name__ == '__main__':
    parser = optparse.OptionParser()
    parser.add_option('-u', '--user',
        action='append',
        default=[],
        dest='users',
        help='The username to be printed. Provide this multiple times to '
        'print the username for multiple users.',
    )
    options, args = parser.parse_args()

    for user in options.users:
        print('Username: %s.' % user)

python echo_usernames.py -u me
Username: me.
python echo_usernames.py -u me -u myself
Username: me.
Username: myself.

import argparse

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    args = parser.parse_args()
    print('The script ran successfully and did nothing.')

parser.add_argument('-q', '--quiet',
    action='store_true',
    dest='quiet',
    help='Suppress output.',
)

import argparse

if __name__ == '__main__':
    parser = argparse.ArgumentParser(prefix_chars='/')

    parser.add_argument('/q', '//quiet',
        action='store_true',
        dest='quiet',
        help='Suppress output.',
    )
    args = parser.parse_args()
    print('Quiet mode is %r.' % args.quiet)

import argparse

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
        parser.add_argument('-H', '--host',
        default='localhost',
        dest='host',
        help='The host to connect to. Defaults to localhost.',
        type=str,
    )
    args = parser.parse_args()
    print('The host is %s.' % args.host)

python argparse_args.py -Hlocalhost
python argparse_args.py -H"localhost"
python argparse_args.py -H=localhost
python argparse_args.py -H="localhost"
python argparse_args.py -H localhost
python argparse_args.py -H "localhost"
python argparse_args.py --host=localhost
python argparse_args.py --host="localhost"
python argparse_args.py --host localhost
python argparse_args.py --host "localhost"

import argparse

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--cheese',
        choices=('american', 'cheddar', 'provolone', 'swiss'),
        default='swiss',
        dest='cheese',
        help='The kind of cheese to use',
    )
    args = parser.parse_args()
    print('You have chosen %s cheese.' % args.cheese)

import argparse

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--madlib',
        default=['fox', 'dogs'],
        dest='madlib',
        help='Two words to place in the madlib.',
        nargs=2,
    )
    args = parser.parse_args()
    print('The quick brown {0} jumped over the '
        'lazy {1}.'.format(*args.madlib))

python argparse_multiargs.py
    The quick brown fox jumped over the lazy dogs.
python argparse_multiargs.py --madlib pirate ninjas
    The quick brown pirate jumped over the lazy ninjas.

import argparse

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('addends',
        help='Integers to provide a sum of',
        nargs='+',
        type=int,
    )

    args = parser.parse_args()
    print('%s = %d' % (
        ' + '.join([str(i) for i in args.addends]),
        sum(args.addends),
    ))

python cli_script.py --help
usage: cli_script.py [-h] addends [addends…]
positional arguments:
addends Integers to provide a sum of
optional arguments:
-h, --help show this help message and exit


python cli_script.py 1 2 5
1 + 2 + 5 = 8

import argparse

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('-c', '--config-file',
        default='/etc/cli_script',
        dest='config',
        help='The configuration file to use.',
        type=argparse.FileType('r')
    )
    args = parser.parse_args()
    print(args.config.read())

echo "This is my config file." > foo.txt
python cli_script.py --config-file foo.txt
    This is my config file.

python cli_script.py --config-file bar.txt
usage: cli_script.py [-h] [-c CONFIG]
cli_script.py: error: argument -c/--config-file: can't open 'bar.txt':
[Errno 2] No such file or directory: 'bar.txt'

import asyncio
loop = asyncio.get_event_loop()
loop.is_running()

False

loop.run_forever()

import functools
def hello_world():
    print('hello world!')

def stop_loop(loop):
    print('Stopping loop.')
    loop.stop()

loop.call_soon(hello_world)
loop.call_soon(functools.partial(stop_loop, loop))
loop.run_forever()

hello world!
Stopping loop.

loop = asyncio.get_event_loop()
loop.call_later(1, hello_world)
loop.call_later(2, functools.partial(stop_loop, loop))
loop.run_forever()

hello world!
Stopping loop.

# 以下和hello_world函数等价
partial = functools.partial(print, 'hello world!')
partial.func

<function print>

loop = asyncio.get_event_loop()
@asyncio.coroutine
def trivial():
    return '在结束前执行'

loop.run_until_complete(trivial())

'在结束前执行'

# 引入线程模块
import asyncio
import threading

def run_loop_in_background(loop):
    def thread_func(l):
        asyncio.set_event_loop(l)
        l.run_forever()
    thread = threading.Thread(target=thread_func, args=(loop,))
    thread.start()
    return thread

loop = asyncio.get_event_loop()
t = run_loop_in_background(loop)
loop.is_running()

loop.call_soon_threadsafe(functools.partial(print,'hello, world!'))

<Handle print('hello, world!')()>

import asyncio

@asyncio.coroutine
def coro_sum(*args):
    answer = 0
    for i in args:
        answer += i
    return answer

loop = asyncio.get_event_loop()
loop.run_until_complete(coro_sum(1, 2, 3, 4, 5))

hello, world!

15

print(coro_sum(1, 2, 3, 4, 5))
try:
    next(coro_sum(1, 2, 3, 4, 5))
except StopIteration as ex:
    print(ex.value)

<generator object coro_sum at 0x7f87630d2fc0>
15

import asyncio

@asyncio.coroutine
def nested(*args):
    print('The `nested` function ran with args: %r' % (args,))
    return [i + 1 for i in args]

@asyncio.coroutine
def outer(*args):
    print('The `outer` function ran with args: %r' % (args,))
    answer = yield from nested(*[i * 2 for i in args])
    return answer

loop = asyncio.get_event_loop()
loop.run_until_complete(outer(2, 3, 5, 8))

The `outer` function ran with args: (2, 3, 5, 8)
The `nested` function ran with args: (4, 6, 10, 16)

[5, 7, 11, 17]

import asyncio

@asyncio.coroutine
def make_tea(variety):
    print('Now making %s tea.' % variety)
    asyncio.get_event_loop().stop()
    return '%s tea' % variety

task = asyncio.ensure_future(make_tea('chamomile'))
task

<Task pending coro=<make_tea() running at /root/anaconda3/lib/python3.6/asyncio/coroutines.py:208>>

task.done()

False

task.result()

---------------------------------------------------------------------------

InvalidStateError                         Traceback (most recent call last)

<ipython-input-13-753967b63670> in <module>()
----> 1 task.result()

InvalidStateError: Result is not ready.

loop = asyncio.get_event_loop()
loop.run_forever()

Now making chamomile tea.

task.done()

True

task.result()

'chamomile tea'

import asyncio
loop = asyncio.get_event_loop()

@asyncio.coroutine
def make_tea(variety):
    print('Now making %s tea.' % variety)
    return '%s tea' % variety

def confirm_tea(future):
    print('The %s is made.' % future.result())

task = asyncio.ensure_future(make_tea('green')) #注册事件循环
task.add_done_callback(confirm_tea) #添加回调，将会在事件执行结束后执行

loop.run_until_complete(task)

Now making green tea.
The green tea is made.

'green tea'

import asyncio
import functools

loop = asyncio.get_event_loop()

@asyncio.coroutine
def make_tea(variety):
    print('Now making %s tea.' % variety)
    return '%s tea' % variety

def add_ingredient(ingredient, future):
    print('Now adding %s to the %s.' % (ingredient, future.result()))


task = asyncio.ensure_future(make_tea('herbal'))
task.add_done_callback(functools.partial(add_ingredient, 'honey'))

loop.run_until_complete(task)

Now making herbal tea.
Now adding honey to the herbal tea.

'herbal tea'

import asyncio
loop = asyncio.get_event_loop()

@asyncio.coroutine
def make_tea(variety):
    print('Now making %s tea.' % variety)
    return '%s tea' % variety

meta_task = asyncio.gather(
    make_tea('chamomile'),
    make_tea('green'),
    make_tea('herbal')
)

print(meta_task.done())

loop.run_until_complete(meta_task)
print(meta_task.done())
print(meta_task.result())

False
Now making chamomile tea.
Now making herbal tea.
Now making green tea.
True
['chamomile tea', 'green tea', 'herbal tea']

# 为一组任务全部完成后，执行一次的回调
import asyncio
loop = asyncio.get_event_loop()

@asyncio.coroutine
def make_tea(variety):
    print('Now making %s tea.' % variety)
    return '%s tea' % variety

def mix(future):
    print('Mixing the %s together.' % ' and '.join(future.result()) )

meta_task = asyncio.gather(make_tea('herbal'), make_tea('green'))
meta_task.add_done_callback(mix)

loop.run_until_complete(meta_task)

Now making green tea.
Now making herbal tea.
Mixing the herbal tea and green tea together.

['herbal tea', 'green tea']

import asyncio
loop = asyncio.get_event_loop()

coro = asyncio.wait([asyncio.sleep(5), asyncio.sleep(1)], timeout=3)
loop.run_until_complete(coro)

({<Task finished coro=<sleep() done, defined at /root/anaconda3/lib/python3.6/asyncio/tasks.py:462> result=None>},
 {<Task pending coro=<sleep() running at /root/anaconda3/lib/python3.6/asyncio/tasks.py:476> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f8763088078>()]>>})

# 当任意一个任务完成后立即停止执行其他任务
import asyncio
loop = asyncio.get_event_loop()

coro = asyncio.wait([
    asyncio.sleep(3),
    asyncio.sleep(2),
    asyncio.sleep(1),
    ], return_when=asyncio.FIRST_COMPLETED)

loop.run_until_complete(coro)

({<Task finished coro=<sleep() done, defined at /root/anaconda3/lib/python3.6/asyncio/tasks.py:462> result=None>},
 {<Task pending coro=<sleep() running at /root/anaconda3/lib/python3.6/asyncio/tasks.py:476> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f8763088768>()]>>,
  <Task pending coro=<sleep() running at /root/anaconda3/lib/python3.6/asyncio/tasks.py:476> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f87630888b8>()]>>})

# 当异常发生时立即停止执行，
# 若没有发生异常，将等待所有任务执行完毕返回
import asyncio
loop = asyncio.get_event_loop()

@asyncio.coroutine
def raise_ex_after(seconds):
    yield from asyncio.sleep(seconds)
    raise RuntimeError('Raising an exception.')

coro = asyncio.wait([
    asyncio.sleep(1),
    raise_ex_after(2),
    asyncio.sleep(3),
    ], return_when=asyncio.FIRST_EXCEPTION)

loop.run_until_complete(coro)

({<Task finished coro=<raise_ex_after() done, defined at <ipython-input-25-fbc575f4c652>:5> exception=RuntimeError('Raising an exception.',)>,
  <Task finished coro=<sleep() done, defined at /root/anaconda3/lib/python3.6/asyncio/tasks.py:462> result=None>},
 {<Task pending coro=<sleep() running at /root/anaconda3/lib/python3.6/asyncio/tasks.py:476> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f8763088678>()]>>})

import asyncio
queue = asyncio.Queue()
queue.put_nowait('foo')
print(queue.qsize())
print(queue.get_nowait())
print(queue.qsize())

1
foo
0

import asyncio
queue = asyncio.Queue(maxsize=5)

import asyncio

class Shutdown(Exception):
    pass

class ServerProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        self.write('Welcome.')

    def data_received(self, data):
        # 检查数据对于空数据不执行任何操作
        if not data:
            return

        # 这个服务器的命令应该是一个单一的单词
        # 空格分隔的参数。
        message = data.decode('ascii')
        command = message.strip().split(' ')[0].lower()
        args = message.strip().split(' ')[1:]
        # 验证命令是否存在
        if not hasattr(self, 'command_%s' % command):
            self.write('Invalid command: %s' % command)
            return
        # 返回一个命令属性
        try:
            return getattr(self, 'command_%s' % command)(*args)
        except Exception as ex:
            self.write('Error: %s\n' % str(ex))

    def write(self, msg_string):
        string += '\n'
        self.transport.write(msg_string.encode('ascii', 'ignore'))

    def command_add(self, *args):
        args = [int(i) for i in args]
        self.write('%d' % sum(args))

    def command_shutdown(self):
        self.write('Okay. Shutting down.')
        raise KeyboardInterrupt

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    coro = loop.create_server(ServerProtocol, '127.0.0.1', 8000)
    asyncio.async(coro)
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass

telnet 127.0.0.1 8000
Trying 127.0.0.1…
Connected to localhost.
Escape character is 'ˆ]'.
Welcome.

add 3 5
8

make_tea
Invalid command: make_tea

shutdown
Okay. Shutting down.
Connection closed by foreign host.