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()

展开全文 >>

python多线程教程:并发与并行

各种激烈的讨论都会经常的提到Python的多线程工作是多么的困难，指向被称为全局解释器锁阻碍了Python代码不能多线程同时执行。由于这个，如果你不是一个Python开发者以及你来自其他语言例如C++活着Java,线程模块表现出来的行为可能不是你所期望的那样。但是必须指出的是你仍可以写出并发或者并行的代码并使其表现出色，只要考虑全面。如果你还没有读到，我建议你看一下 Eqbal Quran’s 在Toptal博客上的article on concurrency and parallelism in Ruby。

在Python并发教程中，我们将会写出一个简单的Python脚本去下载来自Imgur排行靠前的图片，我们将从一个顺序下载的版本开始，或一次一个，作为先决条件，你将不得不注册一个应用在Imgur,如果你还没有创建一个Imgur账户，请先创建一个。

这个教程的脚本已经在Python3.4.2上测试通过，稍微作修改，就可以运行在Python2，urllib改动最大在这两个Python版本上。

开始在Python上使用多线程

让我们开始创建一个Python模块，并将其命名成”download.py”, 这个文件将会包含所有需要的函数去获取图片列表以及下载它们。我们将会分开这些函数为三个分离的函数。

  1. get_links
  2. download_link
  3. setup_download_dir

第三个函数, “setup_download_dir”, 将会被用于创建下载保存目录如果不存在。

Imgur’s ApI要求HTTP请求头上携带客户端ID的”Authorization”标头，你可以从已经注册的Imgur的应用面板上查看客户端ID，请求返回结果会是JSON格式。我们可以使用Python的标准JSON库去解码。下载这些图片是一个非常简单的任务，只要我们获取到这些图片的URL以及将它们写入文件。

这个脚本看起来是这样的:

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
import json
import logging
import os
from pathlib import Path
from urllib.request import urlopen, Request

logger = logging.getLogger(__name__)

def get_links(client_id):
   headers = {'Authorization': 'Client-ID {}'.format(client_id)}
   req = Request('https://api.imgur.com/3/gallery/', headers=headers, method='GET')
   with urlopen(req) as resp:
       data = json.loads(resp.readall().decode('utf-8'))
   return map(lambda item: item['link'], data['data'])

def download_link(directory, link):
   logger.info('Downloading %s', link)
   download_path = directory / os.path.basename(link)
   with urlopen(link) as image, download_path.open('wb') as f:
       f.write(image.readall())

def setup_download_dir():
   download_dir = Path('images')
   if not download_dir.exists():
       download_dir.mkdir()
   return download_dir

接下来，我们将会写一个模块并使用这些函数去下载这些图片，一个接着一个，我们将会将其命名成”simple.py”,这个首先将会包含主函数，最初的Imgur图片下载器。这个模块将会从环境变量”IMGUR_CLIENT_ID”中接收Imgur的客户端ID，它将会在”setup_download_dir”中被调用去创建下载目标目录，最终，它会获取一个图片列表使用”get_links”函数，过滤所有的GIF和唱片URLs，然后使用”download_link”区下载以及保存每一张图片到硬盘中，这个”single.py”看起来如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import logging
import os
from time import time

from download import setup_download_dir, get_links, download_link

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logging.getLogger('requests').setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)

def main():
   ts = time()
   client_id = os.getenv('IMGUR_CLIENT_ID')
   if not client_id:
       raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
   download_dir = setup_download_dir()
   links = [l for l in get_links(client_id) if l.endswith('.jpg')]
   for link in links:
       download_link(download_dir, link)
   print('Took {}s'.format(time() - ts))

if __name__ == '__main__':
   main()

在我的笔记本上，这个脚本下载91张图片使用了19.4秒，请注意这些数字将因你的网络环境而异。但是我们我们想要下载更多的图片呢？或者900张，而不是90张，每一张图片平均使用0.2秒，900张图片将会使用大约3分钟，那么9000张图片将会使用掉30分钟。好消息是通过并发或并行介绍，我们可以显著的加速。

所有后续的代码示例将只会导入新的和特定的导入语句，为了方便，这些Python脚本可以在这个GitHub仓库找到。

使用多线程进行并发和并行

线程时最为熟知的方法实现Python并发与并行。线程通常由操作系统提供的功能。线程比进程更为轻量级，并且共用同一内存空间。

在我们的Python线程教程中，我们将会写一个新的模块代替”single.py”,这个模块将会创建一个8个线程的池，总共包含主线程一共9个线程。我选择8个线程，因为我的电脑有8个CPU以及每个工作线程运行在每个CPU核心上同时运行看起来会是一个好的选择，在练习上，这个数字是基于其他因素选择得应更为仔细，例如其他应用以及服务运行在同一台机器上。

这个和之前的一个非常相似，除了我们拥有了一个新类，DownloadWorker, 是Thread类下的一个子类，运行方法被重写了。它运行一个无限循环，在每一次的迭代上，它会调用”self.queue.get”去试着从线程安全的队列中获取一个URL，它会阻塞直到队列中有其他worker处理。一旦worker从队列中接收到一条，它接下来将会调用和之前脚本相同的”download_link”方法去下载图片以及保存到图片目录。在下载完成后，worker通知队列这个任务已经完成，这是非常重要的，因为队列跟踪有多少任务入队。然后调用”queue.join()”将会阻塞主线程如果workers没有通知它们已经完成了任务。

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
from queue import Queue
from threading import Thread

class DownloadWorker(Thread):
   def __init__(self, queue):
       Thread.__init__(self)
       self.queue = queue

   def run(self):
       while True:
           # Get the work from the queue and expand the tuple
           directory, link = self.queue.get()
           download_link(directory, link)
           self.queue.task_done()

def main():
   ts = time()
   client_id = os.getenv('IMGUR_CLIENT_ID')
   if not client_id:
       raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
   download_dir = setup_download_dir()
   links = [l for l in get_links(client_id) if l.endswith('.jpg')]
   # Create a queue to communicate with the worker threads
   queue = Queue()
   # Create 8 worker threads
   for x in range(8):
       worker = DownloadWorker(queue)
       # Setting daemon to True will let the main thread exit even though the workers are blocking
       worker.daemon = True
       worker.start()
   # Put the tasks into the queue as a tuple
   for link in links:
       logger.info('Queueing {}'.format(link))
       queue.put((download_dir, link))
   # Causes the main thread to wait for the queue to finish processing all the tasks
   queue.join()
   print('Took {}'.format(time() - ts))

运行这个脚本在同一台机器上会更快的获取结果，只用了4.1秒!和之前的相比快了4.7倍。虽然这要快很多，但是值得提及的是由于GIL的原因同一时刻只有一个线程被执行，因此，这个代码是并发的但是不是并行的，相比之前快的原因在于这是一个IO绑定任务。处理器在下载这些图片时并不需要太耗费力气。大部分时间使用在等待网络上。这也是为什么线程可以提供大幅度的速度提升。处理器可以切换上下文在这些线程无论其中的哪一个准备去做一些任务。在Python或者任务其他带有GIL解释性语言使用线程模块实际上可能导致性能下降，如果你的代码是CPU绑定任务，例如解压文件，使用多线程模块将会使结果更慢，对于CPU绑定的任务来讲和真正的并行执行，我们将会使用multiprocessing模块。

事实上参考Python的实现，CPython，拥有GIL，不是所有的Python解释器都是这样。例如，IronPython,一个是使用.NET 框架实现的Python解释器，并不会有GIL，诸如Jython,基于Java实现的，你可以在这找到一系列Python解释器。

使用多进程

多进程模块比多线程模块更容易使用，因为我们不需要添加类似于线程示例的类。唯一改变的是主函数。

使用多进程我们可以创建一个进程池，使用提供的map方法，我们可以将列表中的URLs放入池中。将会产生8个新进程并使用每个进程并行下载这些图片，这是真正的并行。但它也带来的成本。脚本的整个内存被复制到各个子进程中。在这个简单的例子中，这不是什么大不了的事情，但是它可能很容易产生严重开销。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from functools import partial
from multiprocessing.pool import Pool

def main():
   ts = time()
   client_id = os.getenv('IMGUR_CLIENT_ID')
   if not client_id:
       raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
   download_dir = setup_download_dir()
   links = [l for l in get_links(client_id) if l.endswith('.jpg')]
   download = partial(download_link, download_dir)
   with Pool(8) as p:
       p.map(download, links)
   print('Took {}s'.format(time() - ts))

分配给多个worker

虽然线程和进程脚本都非常适合运行在个人电脑上，如果你想运行在不同的机器上你应该怎么做呢？或者向上扩展增加更多的CPU。 对于长期运行的web项目来讲会是一个很好的用例。如果你有长时间运行任务，你不想在同一台机器上启动一堆需要运行的应用程序在多进程或多线程上。这将会降低你的应用性能对所有的你的用户，最好能够在另外一台或多台上运行这些任务。

对于这种任务一个出色的Python库是RQ,一个非常简单但是又非常功能强大的库，你首先需要确定函数和它的参数,pickles将会被调用，并将其加入到Redis列表中，声明工作排队是第一步，但是他并不会做任何事情，我们至少需要一个worker去监听这个工作队列。

第一步我们需要安装和运行Redis服务在你的个人电脑上，或者拥有访问Redis服务的权限，然后，只会在原有的代码基础上稍作改动，我们首先创建一个RQ队列实例并且连接到Redis服务上通过redis-py library,然后，代替刚才调用的”download_link”方法，我们使用“q.enqueue(download_link, download_dir, link)”。这个队列将函数作为它的第一个参数，然后其他参数将会带入到这个函数直到这个任务被执行。

最后一步我们需要启动这些worker,RQ提供了一系列的脚本去运行这些workers在默认的队列上，只需要在终端是那个执行”rqworker”然后它会启动一个worker监听这个默认的队列，请确保你目前的工作目录和这些脚本是在同一目录中，如果你想监听不同的队列，你可以运行 “rqworker queue_name”然后它就会监听倍命名的队列。关于RQ最出色的地方在于只要你连接到Redis,你可以运行任意多个worker在不同的机器上。这是非常容易扩展的。这个是RQ的版本:

1
2
3
4
5
6
7
8
9
10
11
12
from redis import Redis
from rq import Queue

def main():
   client_id = os.getenv('IMGUR_CLIENT_ID')
   if not client_id:
       raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
   download_dir = setup_download_dir()
   links = [l for l in get_links(client_id) if l.endswith('.jpg')]
   q = Queue(connection=Redis(host='localhost', port=6379))
   for link in links:
       q.enqueue(download_link, download_dir, link)

然而，RQ不是唯一的Python工作队列解决方式，RQ是简单使用并且非常好的覆盖简单用例，如果需要其他高级的特性，其他工作队列，如Celery可以被使用。

结论

如果你的代码是IO绑定的，multiprocessing和multithreading都适合你，多进程比多线程更容易使用，但是会占用更多的内存，如果你的代码是CPU绑定的，那么multiprocessing可能是更优的选择。尤其目标机器拥有多个CPU，对于web应用，如果想要扩展worker,RQ会是更好的选择。

此片文章翻译自:beginners-guide-to-concurrency-and-parallelism-in-python.

展开全文 >>

不同用户的多条操作时间差值平均值

例如: 一站表

id uid login_time
1 1 2018-04-10 18:19:14
2 1 2019-04-10 18:19:14
3 1 2020-04-10 18:19:14
4 2 2019-04-10 18:19:14

那么，最终结果类似于:

uid average
1 (2020-04-10 18:19:14 - 2018-04-10 18:19:14) / 2 
1
2
3

select uid, timestampdiff(day, min(login_time), max(login_time)) / (count(*) - 1) from table_name group by uid having count(*) > 1;

需要买本高级查询的书了-_-!

展开全文 >>

SQLAlchemy字段类型

基本类型

字段类型是在定义模型时, 对每个Column的类型约定. 不同类型的字段类型在输入输出上, 及支持的操作方面, 有所区别.
这里只介绍sqlalchemy.types.*中的类型, SQL 标准类型方面, 是写什么最后生成的DDL语句就是什么, 比如 BIGINT, BLOG 这些, 但是这些类型并不一定在所有数据库中都有支持. 除此而外, SQLAlchemy 也支持一些特定数据库的特定类型, 这些需要从具体的 dialects 实现里导入.

Integer/BigInteger/SmallInteger

整形.

Boolean

布尔类型. Python 中表现为 True/False , 数据库根据支持情况, 表现为 BOOLEAN 或SMALLINT . 实例化时可以指定是否创建约束(默认创建).

Date/DateTime/Time (timezone=False)

日期类型, Time 和 DateTime 实例化时可以指定是否带时区信息.

Interval

时间偏差类型. 在 Python 中表现为 datetime.timedelta() , 数据库不支持此类型则存为日期.

Enum (*enums, **kw)

枚举类型, 根据数据库支持情况, SQLAlchemy 会使用原生支持或者使用 VARCHAR 类型附加约束的方式实现. 原生支持中涉及新类型创建, 细节在实例化时控制.

Float

浮点小数.

Numeric (precision=None, scale=None, decimal_return_scale=None, …)

定点小数, Python 中表现为 Decimal .

LargeBinary (length=None)

字节数据. 根据数据库实现, 在实例化时可能需要指定大小.

PickleType

Python 对象的序列化类型.

String (length=None, collation=None, …)

字符串类型, Python 中表现为 Unicode , 数据库表现为 VARCHAR , 通常都需要指定长度.

Unicode

类似与字符串类型, 在某些数据库实现下, 会明确表示支持非 ASCII 字符. 同时输入输出也强制是 Unicode 类型.

Text

长文本类型, Python 表现为 Unicode , 数据库表现为 TEXT .

UnicodeText

参考 Unicode .

展开全文 >>

SQLAlchemy会话与事务控制

基本使用

SQLAlchemy 的session是用于管理数据库操作的一个像容器一样的东西. 模型实例对象本身独立存在, 而要让其修改(创建)生效, 则需要把它们加入某个session. 同时你也可以把模型实例对象从session中去除. 被session管理的实例对象, 在session.commit()时被提交到数据库. 同时session.rollback()是回滚变更.

session.flush()的作用是在事务管理内与数据库发生交互, 对应的实例状态被反映到数据库. 比如自增ID被填充上值.

1
2
3
4
5
6
7
8
9
user = User(name=u'名字')
session.add(user)
session.commit()
try:
    user = session.Query(User).first()
    user.name = u'改名字'
    session.commit()
except:
    session.rollback()

for update

SQLAlchemy 的 Query 支持 select ... for update / share .

1
2
session.Query(User).with_for_update().first()
session.Query(User).with_for_update(read=True).first()

完整形式是:

1
with_for_update(read=False, nowait=False, of=None)

read

是标识加互斥锁还是共享锁. 当为True时, 即for share的语句, 是共享锁. 多个事务可以获取共享锁, 互斥锁只能一个事务获取. 有”多个地方”都希望是”这段时间我获取的数据不能被修改, 我也不会改”, 那么只能使用共享锁.

nowait

其它事务碰到锁, 是否不等待直接”报错”.

of

指明上锁的表, 如果不指明, 则查询中涉及的所有表(行)都会加锁.

事务嵌套

SQLAlchemy 中的事务嵌套有两种情况. 一是在session中管理的事务, 本身有层次性. 二是session和原始的connection之间, 是一种层次关系, 在这session,connection两个概念之中的事务同样具有这样的层次.
session中的事务, 可能通过begin_nested()方法做savepoint:

1
2
3
4
5
6
7
8
session.add(u1)
session.add(u2)

session.begin_nested()
session.add(u3)
session.rollback() # rolls back u3, keeps u1 and u2

session.commit()

或者使用上下文对象:

1
2
3
4
5
6
7
for record in records:
    try:
        with session.begin_nested():
            session.merge(record)
    except:
        print "Skipped record %s" % record
session.commit()

嵌套的事务的一个效果, 是最外层事务提交整个变更才会生效.

1
2
3
4
5
6
7
user = User(name='2')

session.begin_nested()
session.add(user)
session.commit()

session.rollback()

于是, 前面说的第二种情况有一种应用方式, 就是在connection上做一个事务, 最终也在connection上回滚这个事务, 如果session是bind到这个连接上的, 那么 session上所做的更改全部不会生效:

1
2
3
4
5
6
7
8
9
10
11
12
13
conn = Engine.connect()
session = Session(bind=conn)
trans = conn.begin()

user = User(name='2')
session.begin_nested()
session.add(user)
session.commit()

session.commit()

trans.rollback()

在测试中这种方式可能会有用.

二段式提交

二段式提交, Two-Phase, 是为解决分布式环境下多点事务控制的一套协议.
与一般事务控制的不同是, 一般事务是 begin, 之后 commit 结束.
而二段式提交的流程上, begin 之后, 是 prepare transaction transaction_id , 这时相关事务数据已经持久化了. 之后, 再在任何时候(哪怕重启服务), 作 commit prepared transaction_id 或者 rollback prepared transaction_id .
从多点事务的控制来看, 应用层要做的事是, 先把任务分发出去, 然后收集”事务准备”的状态(prepare transaction 的结果). 根据收集的结果决定最后是commit还是rollback.
简单来说, 就是事务先保存, 再说提交的事.
SQLAlchemy 中对这个机制的支持, 是在构建会话类是加入twophase参数:

1
Session = sessionmaker(twophase=True)

然后会话类可以根据一些策略, 绑定多个Engine, 可以是多个数据库连接, 比如:

1
2
Session = sessionmaker(twophase=True)
Session.configure(binds={User: Engine, Blog: Engine2})

这样, 在获取一个会话实例之后, 就处在二段式提交机制的支持之下, SQLAlchemy 自己会作多点的协调了. 完整的流程:

1
2
3
4
5
6
7
8
9
10
11
Engine = create_engine('postgresql://test@localhost:5432/test', echo=True)
Engine2 = create_engine('postgresql://test@localhost:5432/test2', echo=True)

Session = sessionmaker(twophase=True)

Session.configure(binds={User: Engine, Blog: Engine2})
session = Session()

user = User(name=u'名字')
session.add(user)
session.commit()

对应的SQL大概就是:

1
2
3
4
begin;
insert into "user" (name) values (?);
prepare transaction 'xx';
commit prepared 'xx';

使用时, Postgresql数据库需要把 max_prepared_transactions这个配置项的值改成大于 0

Example

如何正确使用事务？

假设有一个简单的银行系统，一共两名用户：

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

    id = Column(Integer, primary_key=True)
    money = Column(DECIMAL(10, 2))

class TanseferLog(BaseModel):
    __tablename__ = 'tansefer_log'

    id = Column(Integer, primary_key=True)
    from_user = Column(Integer, ForeignKey('user.id', ondelete='CASCADE', onupdate='CASCADE'))
    to_user = Column(Integer, ForeignKey('user.id', ondelete='CASCADE', onupdate='CASCADE'))
    amount = Column(DECIMAL(10, 2))

user = User(money=100)
session.add(user)
user = User(money=0)
session.add(user)
session.commit()

然后开两个 session，同时进行两次转账操作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
session1 = DB_Session()
session2 = DB_Session()

user1 = session1.query(User).get(1)
user2 = session1.query(User).get(2)
if user1.money >= 100:
    user1.money -= 100
    user2.money += 100
    session1.add(TanseferLog(from_user=1, to_user=2, amount=100))

user1 = session2.query(User).get(1)
user2 = session2.query(User).get(2)
if user1.money >= 100:
    user1.money -= 100
    user2.money += 100
    session2.add(TanseferLog(from_user=1, to_user=2, amount=100))

session1.commit()
session2.commit()

现在看看结果：

1
2
3
4
5
6
7

>>> user1.money
Decimal('0.00')
>>> user2.money
Decimal('100.00')
>>> session.query(TanseferLog).count()
2L

两次转账都成功了，但是只转走了一笔钱，这明显不科学。

可见 MySQL InnoDB 虽然支持事务，但并不是那么简单的，还需要手动加锁。
首先来试试读锁：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
user1 = session1.query(User).with_lockmode('read').get(1)
user2 = session1.query(User).with_lockmode('read').get(2)
if user1.money >= 100:
    user1.money -= 100
    user2.money += 100
    session1.add(TanseferLog(from_user=1, to_user=2, amount=100))

user1 = session2.query(User).with_lockmode('read').get(1)
user2 = session2.query(User).with_lockmode('read').get(2)
if user1.money >= 100:
    user1.money -= 100
    user2.money += 100
    session2.add(TanseferLog(from_user=1, to_user=2, amount=100))
session1.commit()
session2.commit()

现在在执行 session1.commit() 的时候，因为 user1 和 user2 都被 session2 加了读锁，所以会等待锁被释放。超时以后，session1.commit() 会抛出个超时的异常，如果捕捉了的话，或者 session2 在另一个进程，那么 session2.commit() 还是能正常提交的。这种情况下，有一个事务是肯定会提交失败的，所以那些更改等于白做了。

接下来看看写锁，把上段代码中的 ‘read’ 改成 ‘update’ 即可。这次在执行 select 的时候就会被阻塞了：

1
2
user1 = session2.query(User).with_lockmode('update').get(1)

这样只要在超时期间内，session1 完成了提交或回滚，那么 session2 就能正常判断 user1.money >= 100 是否成立了。
由此可见，如果需要更改数据，最好加写锁。

那么什么时候用读锁呢？如果要保证事务运行期间内，被读取的数据不被修改，自己也不去修改，加读锁即可。
举例来说，假设我查询一个用户的开支记录（同时包含余额和转账记录），可以直接把 user 和 tansefer_log 做个内连接。
但如果用户的转账记录特别多，我在查询前想先验证用户的密码（假设在 user 表中），确认相符后才查询转账记录。而这两次查询的期间内，用户可能收到了一笔转账，导致他的 money 字段被修改了，但我在展示给用户时，用户的余额仍然没变，这就不正常了。
而如果我在读取 user 时加了读锁，用户是无法收到转账的（因为无法被另一个事务加写锁来修改 money 字段），这就保证了不会查出额外的 tansefer_log 记录。等我查询完两张表，释放了读锁后，转账就可以继续进行了，不过我显示的数据在当时的确是正确和一致的。

另外要注意的是，如果被查询的字段没有加索引的话，就会变成锁整张表了：

1
2
3
4
5
6
7
8
9
session1.query(User).filter(User.id > 50).with_lockmode('update').all()
session2.query(User).filter(User.id < 40).with_lockmode('update').all() # 不会被锁，因为 id 是主键

session1.rollback()
session2.rollback()

session1.query(User).filter(User.money == 50).with_lockmode('update').all()
session2.query(User).filter(User.money == 40).with_lockmode('update').all() # 会等待解锁，因为 money 上没有索引

要避免的话，可以这样：

1
2
money = Column(DECIMAL(10, 2), index=True)

另一个注意点是子事务。
InnoDB 支持子事务（savepoint 语句），可以简化一些逻辑。
例如有的方法是用于改写数据库的，它执行时可能提交了事务，但在后续的流程中却执行失败了，却没法回滚那个方法中已经提交的事务。这时就可以把那个方法当成子事务来运行了：

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
def step1():
    # ...
    if success:
        session.commit()
        return True
    session.rollback()
    return False

def step2():
    # ...
    if success:
        session.commit()
        return True
    session.rollback()
    return False

session.begin_nested()
if step1():
    session.begin_nested()
    if step2():
        session.commit()
    else:
        session.rollback()
else:
    session.rollback()

此外，rollback 一个子事务，可以释放这个子事务中获得的锁，提高并发性和降低死锁概率。

如何对一个字段进行自增操作？

最简单的办法就是获取时加上写锁：

1
2
3
user = session.query(User).with_lockmode('update').get(1)
user.age += 1
session.commit()

如果不想多一次读的话，这样写也是可以的：

1
2
3
4
5
6
7
8
session.query(User).filter(User.id == 1).update({
    User.age: User.age + 1
})
session.commit()
# 其实字段之间也可以做运算：
session.query(User).filter(User.id == 1).update({
    User.age: User.age + User.id
})

展开全文 >>

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

探索世界美好的存在。