Coursera机器学习吴恩达-Week1

机器学习定义

Arthur Samuel (1959). Machine Learning: Field of study that gives computers the ability to learn without being explicitly programmed.

使机器在没有明确编程上的研究领域具备学习的能力.

Tom Mitchell (1998) Well-posed Learning Problem: A computer program is said to learn from experience E with respect to some task T and some performance measure P, if its performance on T, as measured by P, improves with experience E.

对于某类任务T和性能度量P，如果一个计算机程序在T上以P衡量的性能随着经验E而自我完善，那么我们称这个计算机程序在从经验E学习.
完全为了押韵…

机器学习算法

  • 监督式学习
  • 非监督式学习
  • 强化学习
  • 推荐系统

监督式学习介绍

如图片所示，监督式学习为right answers given.．此处为regression(回归模型).

Regression: Predict continuous valued output(price)
回归: 预测一系列的值输出(价钱)

此处为Classification(分类模型),

Classification: Discrete valued output(0 or 1)
分类: 分离值输出(0或者1)

非监督式学习介绍

模型与代价函数

代价函数: 与真实的所表现出来的误差函数.

  • 线性回归: liner regression
  • 损失函数: cost function
  • 梯度下降: gradient descent

梯度下降

强烈建议阅读:深入浅出-梯度下降法及其实现

矩阵

略.

展开全文 >>

关于async并发的疑惑

之前一直很好奇像sanic或者aiohttp是如何实现并发的，其实从网上看无非就是使用了asyncio,更底层就是事件驱动，libevent之类的，但是纯从应用层来讲，这个会是怎么一回事呢，举个例子:

1
2
3
4
5
6
7
8
9
10
11

async def worker():
    # 模拟消耗时间
    await asyncio.sleep(3)
    return "ok"

async def concurrent():
    # 模拟并发
    tasks = [worker() for _ in range(10)]
    await asyncio.gather(*tasks)

从上面代码来看，整个执行周期为3s,但是有个问题是：如果请求一个一个来，如何不阻塞呢？？

1
2
3
4
5
6
7
8
9
10
11
12
13

async def worker(message):
    # deal message
    await asyncio.sleep(3)
    return "result"

async def consumer():
    # 此处一下过来7条消息
    # 这里如何实现并行执行这7条消息呢？？
    # await worker(message) # ?? 这样会阻塞，并不是一旦发生阻塞就调度出去.
    # 那如果这样呢？
    asyncio.create_task(worker(message))
    # 来了消息创建一个任务，调度出去，不阻塞，这样就可以接收大量消息喽？？

如果按照例如eventlet的做法:

1
2
3
4
5
6
7
8
9
procpool = GreenPool(size=poolsize)

def worker(message):
    pass

def consumer():
    gt = procpool.spawn(worker, message)
    gt.link(handle_message_processed, message) # 此处省略

消息过来，直接抛给pool，并不会进行阻塞,并且有poolsize控制池大小，

额，这里是sanic源码中的一部分…

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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
# -*- coding: utf-8 -*-

"""
-------------------------------------------------
   File Name：     serve
   Description :
   Author :       yu.zhang
   date：          18-7-13
-------------------------------------------------
"""
import asyncio
from functools import partial


class EchoServer(asyncio.Protocol):
    def __init__(self, loop, connections=set()):
        self.connections = connections
        self.transport = None
        self.loop = loop

    def connection_made(self, transport):
        self.connections.add(self)
        self.transport = transport

    def connection_lost(self, exc):
        self.connections.discard(self)
        print('closed:', len(self.connections))

    def data_received(self, data):
        print(data)
        self.transport.write(data)


def serve(host, port, connections=None, protocol=EchoServer, loop=None, reuse_port=False):
    connections = connections if connections is not None else set()
    server = partial(protocol,
                     loop=loop,
                     connections=connections)

    server_coroutine = loop.create_server(server, host, port, reuse_port=reuse_port)
    try:
        echo_server = loop.run_until_complete(server_coroutine)
    except BaseException:
        return
    try:
        loop.run_forever()
    finally:
        for connection in connections:
            print(dir(connection))
        loop.run_until_complete(echo_server.wait_closed())
        loop.close()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    serve('127.0.0.1', 8080, loop=loop)

展开全文 >>

异步mq的窘境

目前在开发的内部爬虫框架中，对于从消息队列中取任务部分，如下图所示，遇到了一些问题。

因为我们整个模型为async的，底层使用的asyncio，对于消息队列的客户端来讲，可选择有的pika和kombu，对于kombu，的确是一个不错的选择，相比pika callback的写法，的确封装的更为高级，写起来比较方便，另外官方支持与eventlet工作，例如openstack的nova底层使用的就是eventlet与kombu，但是kombu目前不支持asyncio,官方将于5.0版本提供支持,我很看好，题外话。

那么目前能够选择貌似只有pika了，官方示例,奈何实力不高，只能在callback的基础上加代码，此处分成两个版本，如下所示。

推入线程池/进程池

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
32
33
34
35
36
37
38
39
40
41
42

class Async PikaConsumer(object):
    def on_message(self, unused_channel, basic_deliver, properties, body):
        """Invoked by pika when a message is delivered from RabbitMQ. The
        channel is passed for your convenience. The basic_deliver object that
        is passed in carries the exchange, routing key, delivery tag and
        a redelivered flag for the message. The properties passed in is an
        instance of BasicProperties with the message properties and the body
        is the message that was sent.

        :param pika.channel.Channel unused_channel: The channel object
        :param pika.Spec.Basic.Deliver: basic_deliver method
        :param pika.Spec.BasicProperties: properties
        :param str|unicode body: The message body

        """
        LOGGER.info('Received message # %s from %s: %s',
                    basic_deliver.delivery_tag, properties.app_id, body)

        # 将获取到的消息推入线程池/进程池中进行处理
        self._connection.ioloop.loop.run_in_executor(crawl.dispatch)  #此处传入dispatch...
        # 直接确认消息，不拒绝消息，不会重新入队列
        self.acknowledge_message(basic_deliver.delivery_tag)

class Crawl(object):
    def dispatch(unused_channel, basic_deliver, properties, body):
        try:
            if body["type"] == "list":
                return self.get_list(body)
            return self.get_data(body)
        except Exception as e:
            # todo something.

    def get_list(self,task):
        pass

    def get_data(self,task):
        pass




异步取来的任务，将其抛给线程池/进程池处理，那么上层应用者直接在get_list和get_data中进行处理，那么此处就有非常大的问题了，因为一个好好的异步模型活生生的给改成了有点同步的感觉了，如果我想在业务层执行异步任务，会发现在当前线程中无法get eventloop,哈哈，好尴尬。这点实际在tornado官方文档中描述如何执行同步代码中有提示，是不是和这个很类似.

1
2
3
4
5
6
7
8
9
10

executor = concurrent.futures.ThreadPoolExecutor(8)

class ThreadPoolHandler(RequestHandler):
    @gen.coroutine
    def get(self):
        for i in range(5):
            print(i)
            yield executor.submit(time.sleep, 1)

异步处理

上面的实在受不了，太烂了，所以此处还是要想办法给异步了，此处斜眼。
此前在看nameko中消费者处理时，他是使用eventlet.spawn方法开启一个新的协程进行处理，从而不阻塞当前loop,那么在asyncio中也一定有相应的方法，下面出场asyncio.async和asyncio.ensure_future方法，其实也是一个方法，asyncio.async将被放弃啦，所以所以介绍ensure_future方法。

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
32
33
34
35
36
37
38
39
40
41
42
43

class Async PikaConsumer(object):
    def on_message(self, unused_channel, basic_deliver, properties, body):
        """Invoked by pika when a message is delivered from RabbitMQ. The
        channel is passed for your convenience. The basic_deliver object that
        is passed in carries the exchange, routing key, delivery tag and
        a redelivered flag for the message. The properties passed in is an
        instance of BasicProperties with the message properties and the body
        is the message that was sent.

        :param pika.channel.Channel unused_channel: The channel object
        :param pika.Spec.Basic.Deliver: basic_deliver method
        :param pika.Spec.BasicProperties: properties
        :param str|unicode body: The message body

        """
        LOGGER.info('Received message # %s from %s: %s',
                    basic_deliver.delivery_tag, properties.app_id, body)

        # 此处改变了哦
        asyncio.ensure_future(self.deal_message(unused_channel, basic_deliver, properties, body))
        # 直接确认消息，不拒绝消息，不会重新入队列
        self.acknowledge_message(basic_deliver.delivery_tag)

 
class Crawl(object):
    config = {"CONCURRENT":2}
    async def dispatch(self,u,b,p, body):
        # 简化写了，，，明白流程就行啦...
        await self.get_list(json.loads(body.decode()))

    async def get_list(self,body):
        #time.sleep(3)

        try:
            with aiohttp.ClientSession() as session:
                resp = await session.get(body["url"], timeout=10)
                body = await resp.text()
                print("异步处理body:", resp.status)
        except:
            print("超时")


由于使用的pika连接器是asyncio的，那么根据pika的官方文档描述，获取到的任务只有在完成的时候才会进行下发新的任务，
如果如上get_list方法下面的使用者写的是同步代码，会导致效率非常地下，所以此处会强制提升业务代码至异步模型，貌似有点激进，所以暂时不更新.

同步处理

额，几天没有更新了，经过几天的思考，目前采用多线程/多进程模型，为什么没有使用异步呢？我觉得可以从一下几方面总结：

  1. 关于消息队列客户端，没有异步支持的客户端，kazoo亦是如此，但是这两者都有gevent、eventlet的支持，为什么不使用呢？因为在看openstack官方论坛以及asyncio的发展趋势，更应该顺应技术发展，如果kombu明年支持了，可以自行再重新实现一遍，整体架构会比目前的更为成熟.
  2. 代理隧道数量有限，其实这个不应该考虑到框架层面上的，因为这个可以通过并发数(目前通过信号量控制)控制，目前没有达到必须使用异步的地步，去加速速度或者减轻资源的消耗.
  3. 满足目前的整体需求，另外使用同步的话整个爬虫团队更方便和他们熟悉的软件工具配合使用，例如chrome headless、selenium、splash等等，如果使用异步的话，还要把他run_in_executor中，还要在同步处理上进一步封装，也比较麻烦.
  4. 真正使用时遇到了其他的问题. 因为我们list和data任务推入同一个消息队列中，导致list任务会非常多导致data任务很难被消费，以及消息堆积导致最新消息不能被及时消费。由于消息队列是FIFO模型，由于生产速度大于消费速度，导致迟迟无法看到data结果，这一方面准备改成两个队列，list和data队列,data队列优先级提高，加速获取结果.
    另外由于list任务生产速度过快，如何加速处理呢，可以使用多个客户端同时处理，由于有去重，那么就会加速消费速度.

先把第一个版本做稳定了，因为目前还是有一些问题的，因为分布式爬虫框架，程序异常退出以及退出迟迟没有在server端看到客户端下线，还要定位原因以及加强处理，等稳定后再加入timer等其他功能。

展开全文 >>

python中不得不说的装饰器

常用以及常看见的写法

1
2
3
4
5
6
7
8
9

from functools import wraps

def cal_time(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

给装饰器传参

1
2
3
4
5
6
7
8
9
10
11

from functools import wraps

def cal_time(max_time):
    def wrapper_func(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        return wrapper
    return wrapper_func

但是今天将介绍另外两种不同的写法，而这两种写法也在库中比较常见　

wrapper callback

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class App(object):
    def __init__(self):
        self.exc = None

    def on_server_error(self, callback):
        # 在这个地方可以做一些其他的东西
        self.exc = callback


if __name__ == "__main__":
    app = App()

    @app.on_server_error
    def deal_server_error():
        print("deal server error")

    try:
        raise ValueError("服务出现异常")
    except Exception:
        app.exc()

flask-like decorator

为什么叫flask-like呢？因为我也没有想到好的名字来叫，另外这种写法也是在flask中见，因为他没有包装func(*args, **kwargs)这层，而是仅给func传递参数．

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
32
33
34
35
36
37
38

import sys

class App(object):
    def __init__(self):
        self.exc = []

    def errorhandler(self, code):
        def decorator(f):
            self.exc.append(f)
            return f

        return decorator


if __name__ == '__main__':
    app = App()


    @app.errorhandler(403)
    def deal_error(e):
        print("403 error.")


    def deal_exception(e):
        exc_type, exc_value, tb = sys.exc_info()
        assert exc_value is e
        print(exc_type, exc_value, tb)
        for deco in app.exc:
            deco(exc_type)


    try:
        raise ValueError("dddd")
    except Exception as e:
        deal_exception(e)


展开全文 >>

cached_property的理解与使用

今天在同事的推荐下,看了下扇贝的sea代码,
没细看,突然看到了cached_property的代码,这个让我突然想到了我们内部爬虫框架的cached_property,当时我在写这部分代码的时候主要目的有点类似如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class A(object):
    def __init__(self):
        pass

    def random(self):
        return random.random()


class B(object):

    @cached_property
    def a(self):
        return A()



A为一个类,然后B有点类似A的一个超集,平常使用的使用为实例化B, 然后为了操作A下面的方法,基本流程为:

1
2
3
>>> b = B()
>>> b.a.random()
>>> b.a.other_method()

当时设计的时候，并没有想到通过property这个将方法变成属性的方法，当时想我只需要初始化两次就行，如下:

1
2
3
4
5

>>> b = B()
>>> a = b.a()
>>> a.random()
>>> a.other_method()

但是后来我就意识到我这种设计有点low,因为其他同事调用的时候可能不会这么使用，另外因为B下面还会有其他类似A这种东西，例如:

1
2
3
4
5
6
7
8
9
10
class B(object):

    @cached_property
    def a(self):
        return A()

    @cached_property
    def c(self):
        return C()

难道每次使用的时候都拿到c这个实例吗，有点傻，所以参考了Flask的cached_property实现，改成了如上了流程．那么进入主题，cached_property到底干了什么事情?

先看Flask的实现:

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
32


class _Missing(object):

    def __repr__(self):
        return 'no value'

    def __reduce__(self):
        return '_missing'


_missing = _Missing()


class cached_property(property):
    def __init__(self, func, name=None, doc=None):
        self.__name__ = name or func.__name__
        self.__module__ = func.__module__
        self.__doc__ = doc or func.__doc__
        self.func = func

    def __set__(self, obj, value):
        obj.__dict__[self.__name__] = value

    def __get__(self, obj, type=None):
        if obj is None:
            return self
        value = obj.__dict__.get(self.__name__, _missing)
        if value is _missing:
            value = self.func(obj)
            obj.__dict__[self.__name__] = value
        return value

再看sea的实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

class cached_property:
    """ thread safe cached property """

    def __init__(self, func, name=None):
        self.func = func
        self.__doc__ = getattr(func, '__doc__')
        self.name = name or func.__name__
        self.lock = Lock()

    def __get__(self, instance, cls=None):
        with self.lock:
            if instance is None:
                return self
            try:
                return instance.__dict__[self.name]
            except KeyError:
                res = instance.__dict__[self.name] = self.func(instance)
                return res

唯一不同点应该就在于加了一个锁吧，那么抛去锁的部分，单纯讲cached_propery的实现

1
2
3
4
5

class B(object):
    @cached_property
    def b(self):
        pass

第一步

学过Python的应该蛮清楚关于装饰器这个概念，当将cached_property加在b上时，就已经完成了cached_property类的实例化(看最后一个类版本的计算时间装饰器)，那怎么传进去的呢？

1
2
3
4
5
6
7
8
9
10


class cached_property:
    """ thread safe cached property """

    def __init__(self, func, name=None):
        self.func = func # 这个func对应上面的就为b
        self.__doc__ = getattr(func, '__doc__') # None
        self.name = name or func.__name__ # func.__name__ 为b

第二步

调用的时候怎么个过程?看Python Document

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
32
33
34
35
36
37
class Property(object):
    "Emulate PyProperty_Type() in Objects/descrobject.c"

    def __init__(self, fget=None, fset=None, fdel=None, doc=None):
        self.fget = fget
        self.fset = fset
        self.fdel = fdel
        if doc is None and fget is not None:
            doc = fget.__doc__
        self.__doc__ = doc

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        if self.fget is None:
            raise AttributeError("unreadable attribute")
        return self.fget(obj)

    def __set__(self, obj, value):
        if self.fset is None:
            raise AttributeError("can't set attribute")
        self.fset(obj, value)

    def __delete__(self, obj):
        if self.fdel is None:
            raise AttributeError("can't delete attribute")
        self.fdel(obj)

    def getter(self, fget):
        return type(self)(fget, self.fset, self.fdel, self.__doc__)

    def setter(self, fset):
        return type(self)(self.fget, fset, self.fdel, self.__doc__)

    def deleter(self, fdel):
        return type(self)(self.fget, self.fset, fdel, self.__doc__)

1
2
3
4
5
6
7
8
9
10
11
def __get__(self, instance, cls=None):
        # instance代表实例化B
        with self.lock:
            if instance is None:
                return self
            try:
                return instance.__dict__[self.name]
            except KeyError: 
                # 如果没有在B的__dict__找到self.name(即方法b), 那么self.func(即方法b)则会执行b(B), 并将其保存到实例化b的__dict__下
                res = instance.__dict__[self.name] = self.func(instance)
                return res

总结

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class B(object):

    @cached_property
    def a(self):
        return A()

    @cached_property
    def c(self):
        return C()


>>> b = B()
>>> for i in range(3):
        print(b.__dict__)
        b.a.random()

剩下自行理解吧..

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 类版本的计算时间的装饰器
class cal_time(object):
    def __init__(self, func):
        self.func = func

    def __call__(self, *args, **kwargs):
        s = time.time()
        resp = self.func(*args, **kwargs)
        e = time.time()
        print("使用时长:{}".format(e - s))
        return resp



展开全文 >>

SQLAlchemy示例:AdjacencyList,单向链接列表

这里说的 AdjacencyList , 就是最常用来在关系数据库中表示树结构的,parent方式:

id name parent
1 一 null
2 二 1
3 三 2

上面的数据, 表示的结构就是:
一
|- 二
|- 三
模型定义很好做:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# -*- coding: utf-8 -*-

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, ForeignKey
from sqlalchemy.types import Integer, Unicode
from sqlalchemy.orm import relationship, sessionmaker, joinedload

BaseModel = declarative_base()
Engine = create_engine('sqlite://', echo=True)
Session = sessionmaker(Engine)

class Node(BaseModel):
    __tablename__ = 'node'

    id = Column(Integer, autoincrement=True, primary_key=True)
    name = Column(Unicode(32), nullable=False, server_default='')
    parent = Column(Integer, ForeignKey('node.id'), index=True,
                    nullable=False, server_default='0')

这里不让parent字段有null, 而使用0代替.
这个例子在关系上, 有一个纠结的地方, 因为 node 这个表, 它是自关联的, 所以如果想要children 和 parent_obj 这两个关系时:

1
2
children = relationship('Node')
parent_obj = relationship('Node')

呃, 尴尬了.
如果是两个表, 那么 SQLAlchemy 可以通过外键在哪张表这个信息, 来确定关系的方向:

1
2
3
4
5
6
7
8
class Blog(BaseModel):
    ...
    user = Column(Integer, ForeignKey('user.id'))
    user_obj = relationship('User')

class User(BaseModel):
    ...
    blog_list = relationship('Blog')

因为外键在 Blog 中, 所以 Blog -> User 的 user_obj 是一个 N -> 1关系.
反之, User -> Blog 的 blog_list 则是一个 1 -> N 的关系.
而自相关的 Node 无法直接判断方向, 所以 SQLAlchemy 会按 1 -> N 处理, 那么:

1
2
children = relationship('Node')
parent_obj = relationship('Node')

这两条之中, children 是正确的, 是我们想要的. 要定义 parent_obj 则需要在 relationship 中通过参数明确表示方向:

1
parent_obj = relationship('Node', remote_side=[id])

这种方式就定义了一个, “到 id” 的 N -> 1 关系.
现在完整的模型定义是:

1
2
3
4
5
6
7
8
9
10
class Node(BaseModel):
    __tablename__ = 'node'

    id = Column(Integer, autoincrement=True, primary_key=True)
    name = Column(Unicode(32), nullable=False, server_default='')
    parent = Column(Integer, ForeignKey('node.id'), index=True,
                    nullable=False, server_default='0')

    children = relationship('Node') # 1 -> N
    parent_obj = relationship('Node', remote_side=[id])

查询方面没什么特殊的了, 不过我发现在自相关的模型关系, lazy 选项不起作用:

1
2
children = relationship('Node', lazy="joined")
parent_obj = relationship('Node', remote_side=[id], lazy="joined")

都是无效的, 只有在查询时, 手动使用 options() 定义:

1
2
n = session.query(Node).filter(Node.name==u'一')\
           .options(joinedload('parent_obj')).first()

如果要一次查出多级的子节点:

1
2
3
n = session.query(Node).filter(Node.name==u'一')\
           .options(joinedload('children').joinedload('children')).first()
print n.name, n.children, n.children[0].children

多个 joinedload() 串连的话, 可以使用 joinedload_all() 来整合:

1
2
3
4
from sqlalchemy.orm import joinedload_all

n = session.query(Node).filter(Node.name==u'一')\
           .options(joinedload_all('children', 'children')).first()

在修改方面, 删除的话, 配置了 cascade , 删除父节点, 则子节点也会自动删除:

1
2
3
4
children = relationship('Node', lazy='joined', cascade='all') # 1 -> N
node = session.query(Node).filter(Node.name == u'一').first()
session.delete(node)
session.commit()

如果只删除子节点, 那么 delete-orphan 选项就很好用了:

1
2
3
4
children = relationship('Node', lazy='joined', cascade='all, delete-orphan') # 1 -> N
node = session.query(Node).filter(Node.name == u'一').first()
node.children = []
session.commit()

展开全文 >>

关于Python的mock

示例

每次使用mock都记不住，今天也是这样，但是这里出现了不一样的地方，先举个例子:

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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
from mock import Mock, patch
import requests


class User(object):
    def __init__(self):
        """

        """


class Admin(User):
    table = "admin"

    def __init__(self):
        super(Admin, self).__init__()

    def list(self):
        return []

# 对类进行mock
@patch("__main__.Admin")
def do(cls):
    mock = Mock(table="Admin")
    mock.list.return_value = ["A", ]
    cls.return_value = mock # __init__

    a = Admin()
    return a.list()

# 对方法进行mock
def send_request():
    return requests.request('GET', "http://www.baidu.com/")


@patch("__main__.send_request")
def do2(method):
    mock = Mock(status_code=200)
    mock.json.return_value = {}

    method.return_value = mock

    a = send_request()
    return a.status_code, a.json()


if __name__ == '__main__':
    l = do()
    status_code, json = do2()
    print(l)
    print(status_code, json)

where to patch

上述看也没什么问题，但是此处讲一个Where to patch的问题.

简单描述:

1
2
3
4
5
6
7
8
a.py

    -> Defines SomeClass

b.py

    -> from a import SomeClass
    -> some_function instantiates SomeClass

如上，a.py定义了一个SomeClass,然后b.py引入这个Class,然后在某个地方进行实例化此类，ok，如果要patch SomeClass这个类，要从a.py进行patch呢还是从b.py进行patch呢？
例如:

1
2
3
4
5
6
7
8
9

# b.py

from a import SomeClass

def func():
    s = SomeClass()
    s.do_something()

1
2
3
4
5
6
7
8
9
10
11
12

# test.py

class Test1(TestCase):
    # 这里对？
    @patch("a.SomeClass")
    def do1(self,cls):
        # todo
    # 还是这里对？
    @patch("b.SomeClass")
    def do2(self,cls):
        # todo

上述哪一种方法才能被正常mock??

引用外文的描述:

1
2
3
4
5
6
7
8
Now we want to test some_function but we want to mock out SomeClass using patch. The problem is that when we import module b, which we will have to do then it imports SomeClass from module a. If we use patch to mock out a.SomeClass then it will have no effect on our test; module b already has a reference to the real SomeClass and it looks like our patching had no effect.

The key is to patch out SomeClass where it is used (or where it is looked up ). In this case some_function will actually look up SomeClass in module b, where we have imported it. The patching should look like:

@patch(‘b.SomeClass’)
However, consider the alternative scenario where instead of from a import SomeClass module b does import a and some_function uses a.SomeClass. Both of these import forms are common. In this case the class we want to patch is being looked up on the a module and so we have to patch a.SomeClass instead:

@patch(‘a.SomeClass’)

翻译成中文(翻译不好-_-!)

1
2
3
4
5
6
7
8
9

现在我们想使用patch mock SomeClass测试some_function，但是问题是，当我们导入模块b时，我们将不得不从模块a导入SomeClass,如果我们想patch a.SomeClass发现他对于我们的测试没有任何影响，模块b已经从真正的SomeClass中引用，看起来我们的patch没有被影响。

关键是SomeClass从哪里进行patch,或者从哪里进行查找。在这个示例some_function会从模块b中查找SomeClass,我们也已经导入了，故修改后看起来应该是这样:

@patch(‘b.SomeClass’)

后面就不翻译了...

常见的一些mock

  1. mock class staticmethod
1
2
3
4
5
6
7
8
9
10

class Payroll(object):
    @staticmethod
    def hello():
        pass

@patch("where.Payroll.hello")
def test_hello(self, mock_hello):
    mock_hello.return_value = lambda x: x # example

  1. 如何mock一个异常
1
2
3
4
with patch(
    "addons.hcm.models.payroll.grpc_service.call"
) as mock_ar:
    mock_ar.side_effect = Exception("i wanted.")
  1. mock class method
1
2
3
4
5
6
7
8
9
10

class UserResource(object):
    def _add(self):
        pass


@mock.patch.object(UserResource, "_add")
def test_mock_add(self, mock_add):
    mock_add.return_value = {} # anything u want.

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

探索世界美好的存在。