关于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.

展开全文 >>

bloomfilter

概念

创建一个m位的位数组(bitmap),先将所有的位数组初始化为0。然后选择k个不同的哈希函数。第i个哈希函数对应的字符串str哈希的结果记为h(i, str),且h(i, str)的范围要在0至m-1。如下图所示。

如何判断字符串是否存在呢？
字符串也经过h(i, str),h(2, str),h(3, str)…哈希映射，检查每一个映射到m位的位数组上是否为1，如果不全为1，则表示一定不存在，否则，不能说明完全存在，有误差率在里面，为什么不能说一定存在呢，没看懂，可看：BloomFilters

关于容错率

既然不能完全表示存在，那么如何计算这个误差率呢？一共有三个参数:k,m,n。

参数 表示
k 哈希个数
m 位数组大小
n 字符串个数

下图表示m/n的结果与k个哈希函数的选择出现的错误率表格。

如上，举例简单说明，如果声明一个为数组大小为2，只传入一个字符串，那么m/n为2，如果选择k个哈希，导致的容错率分别是1.39,0.393,0.400。ok，如果要存1亿个字符串，那么大概为多少呢？

简单计算：如果容错率要求为5.73e-06,那么m/n=32,如果n为1亿的话，那么m为32亿，32亿 / 8/ 1024/1024=381.4697265625MB内存。
还是相当可以的。关于更多的容错率，可以看BloomFilters。

此处有个在线计算器

相关的包

Python实现:python-bloomfilter
C实现:pybloomfiltermmap

后续

  1. 选择使用的包
  2. 最终效果

展开全文 >>

SQLAlchemy混合属性机制

直接行为

混合属性, 官方文档中称之为Hybrid Attributes. 这种机制表现为, 一个属性, 在 类 和层面, 和实例 的层面, 其行为是不同的. 之所以需要关注这部分的差异, 原因源于 Python 上下文和 SQL 上下文的差异.
类 层面经常是作为 SQL 查询时的一部分, 它面向的是 SQL 上下文. 而 实例 是已经得到或者创建的结果, 它面向的是 Python 上下文.
定义模型的 Column() 就是一个典型的混合属性. 作为实例属性时, 是具体的对象值访问, 而作为类属性时, 则有构成 SQL 语句表达式的功能.

1
2
3
4
5
6
7
8
9
class Interval(BaseModel):
    __tablename__ = 'interval'

    id = Column(Integer, autoincrement=True, primary_key=True)
    start = Column(Integer)
    end = Column(Integer)

session.add(Interval(start=0, end=100))
session.commit()

实例行为:

1
2
ins = session.query(Interval).first()
print ins.end - ins.start

类行为:

1
ins = session.query(Interval).filter(Interval.end - Interval.start > 10).first()

这种机制其实一直在被使用, 但是可能大家都没有留意一个属性在类和实例上的区别.
如果属性需要被进一步封装, 那么就需要明确声明Hybrid Attributes了:

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 sqlalchemy.ext.hybrid import hybrid_property, hybrid_method

class Interval(BaseModel):
    __tablename__ = 'interval'

    id = Column(Integer, autoincrement=True, primary_key=True)
    start = Column(Integer)
    end = Column(Integer)

    @hybrid_property
    def length(self):
        return self.end - self.start

    @hybrid_method
    def bigger(self, i):
        return self.length > i


session.add(Interval(start=0, end=100))
session.commit()

ins = session.query(Interval).filter(Interval.length > 10).first()
ins = session.query(Interval).filter(Interval.bigger(10)).first()
print ins.bigger(1)

setter的定义同样使用对应的装饰器即可:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class Interval(BaseModel):
    __tablename__ = 'interval'

    id = Column(Integer, autoincrement=True, primary_key=True)
    start = Column(Integer)
    end = Column(Integer)

    @hybrid_property
    def length(self):
        return abs(self.end - self.start)

    @length.setter
    def length(self, l):
        self.end = self.start + l

表达式行为

前面说的属性, 在类和实例上有不同行为, 可以看到, 在类上的行为, 其实就是生成 SQL 表达式时的行为. 上面的例子只是简单的运算, SQLAlchemy 可以自动处理好 Python 函数和 SQL 函数的区别. 但是如果是一些特性更强的 SQL 函数, 就需要手动指定了. 于时, 这时的情况变成, 实例行为是 Python 范畴的调用行为, 而类行为则是生成SQL 函数的相关表达式.
同时是前面的例子, 对于 length 的定义, 更严格上来说, 应该是取绝对值的.

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

    id = Column(Integer, autoincrement=True, primary_key=True)
    start = Column(Integer)
    end = Column(Integer)

    @hybrid_property
    def length(self):
        return abs(self.end - self.start)

但是, 如果使用了 Python 的abs()函数, 在生成 SQL 表达式时显示有无法处理了. 所以, 需要手动定义:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from sqlalchemy import func

class Interval(BaseModel):
    __tablename__ = 'interval'

    id = Column(Integer, autoincrement=True, primary_key=True)
    start = Column(Integer)
    end = Column(Integer)

    @hybrid_property
    def length(self):
        return abs(self.end - self.start)

    @length.expression
    def length(self):
        return func.abs(self.end - self.start)

这样查询时就可以直接使用:

1
ins = session.query(Interval).filter(Interval.length > 1).first()

对应的 SQL :

1
2
3
4
SELECT *
FROM interval
WHERE abs(interval."end" - interval.start) > ?
 LIMIT ? OFFSET ?

应用于关系

总体上没有特别之处:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class Account(BaseModel):
    __tablename__ = 'account'

    id = Column(Integer, autoincrement=True, primary_key=True)
    user = Column(Integer, ForeignKey('user.id'), index=True)
    balance = Column(Integer, server_default='0')


class User(BaseModel):
    __tablename__ = 'user'

    id = Column(Integer, autoincrement=True, primary_key=True)
    name = Column(Unicode(32), nullable=False, server_default='')

    accounts = relationship('Account')
    #balance = association_proxy('accounts', 'balance')

    @hybrid_property
    def balance(self):
        return sum(x.balance for x in self.accounts)

查询时:

1
2
user = session.query(User).first()
print user.balance

这里涉及的东西都是 Python 自己的, 包括那个sum()函数, 和SQL没有关系.
如果想实现的是, 使用SQL的sum()函数, 取出指定用户的总账户金额数, 那么就要考虑把balance 作成表达式的形式:

1
2
3
4
5
6
from sqlalchemy import select

@hybrid_property
def balance(self):
    return select([func.sum(Account.balance)]).where(Account.user == self.id).label('balance_v')
    #return func.sum(Account.balance)

这样的话,User.balance只是单纯的一个表达式了, 查询时指定字段:

1
2
user = session.query(User, User.balance).first()
print user.balance

注意, 如果写成:

1
session.query(User.balance).first()

意义就不再是”获取第一个用户的总金额”, 而变成”获取总金额的第一个”. 这里很坑吧.
像上面这样改, 实例层面就无法使用 balance 属性. 所以, 还是先前介绍的, 表达式可以单独处理:

1
2
3
4
5
6
7
@hybrid_property
def balance(self):
    return sum(x.balance for x in self.accounts)

@balance.expression
def balance(self):
    return select([func.sum(Account.balance)]).where(Account.user == self.id).label('balance_v')

定义了表达式的 balance , 这部分作为查询条件上当然也是可以的:

1
user = session.query(User).filter(User.balance > 1).first()

展开全文 >>

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

探索世界美好的存在。