BeginMan blog

Python 可控配置类

python-kafka KafkaProducer类写的真好,一个可控的配置类。抽象如下:

import copy
import uuid


class Simple(object):
    DEFAULT_CONFIG = {
        'host': 'localhost',
        'client_id': None,
        'api_version': '1.0.0'
    }

    def __init__(self, **configs):
        self.config = copy.copy(self.DEFAULT_CONFIG)
        for key in self.config:
            if key in configs:
                self.config[key] = configs.pop(key)
        
        # Only check for extra config keys in top-level class
        assert not configs, 'Unrecognized configs: %s' % configs

        if self.config['client_id'] is None:
            self.config['client_id'] = str(uuid.uuid4())

if __name__ == '__main__':
    sim = Simple()
    print(sim.config)
    
    sim = Simple(host='192.168.1.129', client_id='demo')
    print(sim.config)

    # sim = Simple(api_version="1.1.1", other="other value")
    # AssertionError: Unrecognized configs: {'other': 'other value'} 

用起来很爽。

Bash下操作mysql

在看《Python web开发实战》中看到如下的命令,觉得这个技巧很好。

$ (echo "user r;" cat database/schema.sql) | mysql --user='web' --password='web'

直接在命令行中把它导入mysql数据库中。这里用到的主要是管道这一技术。

其他可行的操作如下:

$ echo "select 1;" | mysql -u web -p'fangpeng'
mysql: [Warning] Using a password on the command line interface can be insecure.
1
1

这里有Warnning提示,命令行密码安全问题。

$ mysql -u web -p'fangpeng' -e "select 1;"
mysql: [Warning] Using a password on the command line interface can be insecure.
+---+
| 1 |
+---+
| 1 |
+---+

还可以

$ mysql -u web -p'fangpeng' << END
heredoc> use r;
heredoc> select * from users;
heredoc> END
mysql: [Warning] Using a password on the command line interface can be insecure.
id	name	email
1	zhangsan	zhangsan@qq.com
2	lisan	lisan@qq.com
3	wangsan	wangsan@qq.com

推荐:linux shell 管道命令(pipe)使用及与shell重定向区别

LRU算法及Python实现

面试的过程中被问到”最近最少使用算法”,然后就懵逼了,最后面试官说这就是LRU(Least Recently Used)我才恍然大悟,哦,听说过这个算法,但是..自己不会…。所以这篇文章主要就是以攻破LRU以及Python实现为主。

一.概述

这个算法出现在操作系统内存管理上,因为内存是有限且珍贵的,如何更大更高效利用内存呢?就出现了一种虚拟内存,就是在内存有限的情况下,扩展部分外存作为虚拟内存。虽然虚拟页式内存管理增大了进程所需的内存空间,但是不可避免的是内存外存的交换,由于外存低速,所消耗的时间也较长。于是人们想有没有一种算法减少读取外存的次数

因为内外存信息替换是以页面为单位,当需要外存的页面是就得把它调到内存中,那么,把哪个页面调出去可以达到调动尽量少的目的?LRU算法就因此而生了,意思就是使用频繁的xx在后面也可能会使用频繁,反之已经很久没有使用的则很可能以后也不会使用。

二.基本原理

LRU算法其实就是按照近期最少使用的条件淘汰数据。既然是近期最少使用,那么肯定有时间这个筛选条件,也就是淘汰截止目前缓存区中最久未被访问过的元素。

if data in cacheList:
    cacheList.get(data)
else:
    query data
    cacheList.append(data)

上面的操作就是不断的像内存中添加数据,久而久之就会内存爆满,如何淘汰旧数据呢,依照上面说的LRU算法,记录下每个元素最后一次被访问的时间戳,并按照此时间戳早晚进行排序,那么,距今最久未用的被淘汰掉,那么可实现一个简单版本就是:

sorted_cache_list = sorted(cacheList, key=lambda i: i.vist_time)
saved_cache_list = sorted_cache_list[n:]

图示:

这种操作的时间复杂度O(N), 且非常低效的。

LRU算法不是绝对公平的,比如某个元素在淘汰前才被访问了一次,但是之前很长一段时间都未被访问,那么按照LRU算法它也会被保存下来,它无法准确描述“是否常用”这一特性。

下面是根据上面简单实现原理来实现的一个k-v cache.

LRU常用在缓存置换,实现一个K-V cache则需要下面基本的几个API和因素:

  • capacity: cache容量,超出则使用LRU淘汰
  • get(key): 获取缓存元素
  • set(key, value): 添加缓存
  • del(key): 删除缓存元素

一种比较简单的实现方法如下:

In [151]: class LRUCache(object):
     ...:     def __init__(self, capacity):
     ...:         self.capacity = capacity
     ...:         self.tm = 0
     ...:         self.cache = {}    # cache data
     ...:         self.lru = {}      # track the access history with tm
     ...:
     ...:     def get(self, key):
     ...:         if key in self.cache:
     ...:             self.lru[key] = self.tm
     ...:             self.tm += 1   # automatic increment
     ...:             return self.cache[key]
     ...:
     ...:     def set(self, key, value):
     ...:         if len(self.cache) >= self.capacity:
     ...:             old_key = min(self.lru.keys(), key=lambda k: self.lru[k])
     ...:             self.cache.pop(old_key)
     ...:             self.lru.pop(old_key)
     ...:         self.cache[key] = value
     ...:         self.lru[key] = self.tm
     ...:         self.tm += 1
     ...:

In [152]: l = LRUCache(4)

In [153]: l.set('a', 1)

In [154]: l.set('b', 2)

In [155]: l.set('c', 3)

In [156]: l.get('b')
Out[156]: 2

In [157]: l.set('d', 4)

In [158]: l.set('e', 5)

In [159]: l.cache
Out[159]: {'b': 2, 'c': 3, 'd': 4, 'e': 5}

In [160]: l.get('c')
Out[160]: 3

In [161]: l.get('c')
Out[161]: 3

In [162]: l.get('c')
Out[162]: 3

In [163]: l.lru
Out[163]: {'b': 3, 'c': 8, 'd': 4, 'e': 5}

In [164]: l.set('f', 100)

In [165]: l.lru
Out[165]: {'c': 8, 'd': 4, 'e': 5, 'f': 9}

In [166]: l.cache
Out[166]: {'c': 3, 'd': 4, 'e': 5, 'f': 100}

上面的例子,存在瓶颈的地方是min操作,会处理整个cache,时间复杂度O(n), 如果这个dict cache有序的化我们就不用min操作了,直接根据时间戳已经排好序了,超出阈值直接删除cache旧数据。这里需要用到collections模块的OrderedDict类了。

在Python中 dict由于Hash特性,是无序的,但是collections模块的OrderedDict则是有序的字典对象。

那么上面的例子可以通过OrderDict改写。改写规则如下:

  • 对于get操作,先pop然后再插入,既把它摘出来放在表头上
  • 对于set操作,如果缓存在列表里,把它摘出来放在表头上,否则直接放在表头
import collections

class LRUCache:
    def __init__(self, capacity):
        self.capacity = capacity
        self.cache = collections.OrderedDict()

    def get(self, key):
        try:
            value = self.cache.pop(key)
            self.cache[key] = value
            return value
        except KeyError:
            return -1

    def set(self, key, value):
        try:
            self.cache.pop(key)
        except KeyError:
            if len(self.cache) >= self.capacity:
                self.cache.popitem(last=False)
        self.cache[key] = value

这个算法比上面一个高效多了,主要利用OrderedDict的有序性,对于旧元素的删除,则通过popitem方法,该方法源码如下:

class OrderedDict(dict):
    # .....
    
    def popitem(self, last=True):
        '''od.popitem() -> (k, v), return and remove a (key, value) pair.
        Pairs are returned in LIFO order if last is true or FIFO order if false.

        '''
        if not self:
            raise KeyError('dictionary is empty')
        key = next(reversed(self) if last else iter(self))
        value = self.pop(key)
        return key, value

popitem默认采用LIFO(后进先出)原则,如果把last参数改成false则变成FIFO(先进先出)。

注意,OrderedDict的Key会按照插入的顺序排列,不是Key本身排序:

In [194]: o['e'] = 1

In [195]: o['a'] = 2

In [196]: o['c'] = 3

In [197]: o.keys()   # 按照插入key的顺序返回
Out[197]: ['e', 'a', 'c']

如下实现FIFO dict, 超出容量则删除最早插入的key:

from collections import OrderedDict

class FIFOdict(OrderedDict):
     ...:     def __init__(self, capacity):
     ...:         super(FIFOdict, self).__init__()
     ...:         self.capacity = capacity
     ...:
     ...:     def __setitem__(self, k, v):
     ...:         if len(self) >= self.capacity:
     ...:             last = self.popitem(last=False)
     ...:             print "remove last: ", last
     ...:
     ...:         if k in self:
     ...:             del self[key]
     ...:         OrderedDict.__setitem__(self, k, v)

In [230]: f = FIFOdict(3)

In [231]: f['d'] = 1

In [232]: f['a'] = 2

In [233]: f['e'] = 3

In [234]: f['b'] = 4
remove last:  ('d', 1)

In [235]: f['a'] = 6
remove last:  ('a', 2)

In [236]: f
Out[236]: FIFOdict([('e', 3), ('b', 4), ('a', 6)])

如果使用循环链表则,元素被命中时,移到链表头。执行淘汰时,从链表尾开始淘汰。

三.应用

常应用于缓存中的数据淘汰。在Py 3.2版本,functools模板添加了lru_cache装饰器

>>> from functools import lru_cache
>>> @lru_cache(maxsize=None)
... def fib(n):
...     if n < 2:
...             return n
...     return fib(n-1) + fib(n-2)
...
>>> [fib(n) for n in range(16)]
[0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610]
>>> fib.cache_info()
CacheInfo(hits=28, misses=16, maxsize=None, currsize=16)

>>> fib.cache_clear()
>>> fib.cache_info()
CacheInfo(hits=0, misses=0, maxsize=None, currsize=0)

还有一个 PyLRU库 A least recently used (LRU) cache for Python.也挺不错的。

参考

python RESTful浅知

一.RESTful概念

REST (REpresentational State Transfer, 表现层状态转换) 已经变成了 web services 和 web APIs 的标配。它有下面几个核心:

  1. 资源:一个URI映射一个资源,如图片、音频、视频、文档、文件等。
  2. 表现层:资源是一种实体,实体该如何表示?资源的表现形式就是表现层,如文档资源常用JSON/XML等表示。URI可以确定一个资源,但是如何确定它的表现形式呢?通过HTTP的ACCEPT和Content-Type字段来确定。
  3. 状态转换:因为HTTP协议是无状态的,这里所谓的状态是由HTTP方法和服务器配合完成。

REST 设计不需要特定的数据格式。在请求中数据可以以 JSON 形式, 或者有时候作为 url 中查询参数项。

API 规范:

http://[hostname]/name/api/version/ 
# 如
http://[hostname]/todo/api/v1.0/

实例:gist

  • v1:flask实现简单的rest
  • v2:Flask-RESTful 设计 RESTful API
  • v3:Flask 设计简单的基于RESTful 的认证

二.身份认证

然而 REST 的规则之一就是 “无状态”, 因此我们必须要求客户端在每一次请求中提供认证的信息,目前我总结的有如下方案:

  1. Http Basic Authentication
  2. Access Token
  3. Api Key + Security Key + Sign
  4. JWT
  5. OAuth

2.1 Http Basic Authentication

最简单也是最不安全的一种,就是通过username:password 再用Base64编码后的字符串随HTTP首部发出,再由接收者解析校验。

可运行上面的例子,进行如下带认证的访问:

$ curl -u 用户名:密码 -i http://localhost:5000/todo/api/v1.0/tasks

如果缺少用户名及密码,则会出现401 UNAUTHORIZED状态,

这种方案没啥安全性,一解码立马破解掉了.如果不加HTTPS,则相当于裸奔。

>>>"admin:12345".encode('base64')
'YWRtaW46MTIzNDU=\n'

>>>'YWRtaW46MTIzNDU=\n'.decode('base64')
'admin:12345'

2.2 Access Token

原理就是客户端登录后返回给它一个Token, 这个Token包含认证信息,同时要设置过期时间,然后每次访问都携带这个Token.服务端进行认证。

这里涉及三个重要因素:

  1. Token如何设计才能不易破解和伪造,要包含哪些信息?
  2. 过期时间,过长存在盗用风险,过短则重新生成并获取Token次数增多。
  3. HTTPS加密

基于此,这套方案相当于穿着裤衩裸奔。相关代码在auth-token-api.py

2.3 Api Key + Security Key + Sign

这个用的比较多。原理如下:

客户端构造请求:

Server校验

实例代码在:gist:Api Key + Security Key + Sign

2.4 JWT (JSON Web Token)

JWT 由头部、载荷与签名这三部分组成,如下图:

>>> import jwt
>>> encoded = jwt.encode({'some': 'payload'}, 'secret', algorithm='HS256')
'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzb21lIjoicGF5bG9hZCJ9.4twFt5NiznN84AWoo1d7KO1T_yoc0Z6XOpOVswacPZg'

>>> jwt.decode(encoded, 'secret', algorithms=['HS256'])
{'some': 'payload'}

# ref:https://github.com/jpadilla/pyjwt/

三.参考

2016年个人总结

2014年做了个人总结, 2015年也做了年终总结, 不免俗,2016年的也来了。回顾这两年,成长太缓慢了,我有时候很焦虑,也没挣到钱,年龄越来越大了。至此,感觉这篇总结负面情绪会比较多。

2016年唯一跑了一个长途就是去云南,短途则是天津,其他都在公司或小屋里待着,走的江湖少了,世面就见得狭隘。恰逢买了kindle, 总不能浪费吧,于是就人丑多读点书呗,囫囵吞枣看了近60本。但可笑的是,这些书籍并没有拯救我的贫乏,依然是之前的老毛病,日复一日,年复一年,然后到年底了罪恶感上来了,就再扯出来,调侃两下,并大发感慨,最后痛定思痛,装模作样地列出了自己要怎么改,怎么努力。再过一年回头来看,Shit!

今年收获甚少,不多啰嗦了。

s3

like

列了个2017年技术进阶书籍, 重点打牢底层的知识。希望2017年再年终总结时不会被人嗤之以鼻。

(完)~