Messense

在 Django 中使用 Peewee Model 作为 User Model

Why

  1. Django 自带的 ORM 很好用,但是过分依赖于 Django 框架,在多个 Django 项目可能共享一些 models 的时候容易出问题;
  2. Peewee 这个小巧的 ORM 也挺好用的,稍微在上层处理一下可以很容易地支持数据库分库和分表;
  3. 在 Django 中只用 Peewee 作为 ORM 的话,Django 的 auth 功能强依赖于 Django 自带的 ORM,不希望项目即使用 Django 的 ORM 又使用 Peewee 的 ORM;
  4. 虽然 Peewee 的 Playhouse 扩展中支持直接使用 Peewee Model 的方式调用 Django Model,但这样还是会存在问题 1 的情况。

How

以下假设项目名为 django_peewee,且包含一个 app accounts

定义一个 Peewee 格式的 User Model

这里简单给个示例的 User Model,假设文件名为 models.py

from peewee import Model  
from peewee import CharField, BooleanField, BigIntegerField  
from django.contrib.auth.hashers import (  
    make_password,  
    check_password,  
    is_password_usable,  
)  


class User(Model):  
    username = CharField(max_length=32, unique=True, verbose_name=u'用户名')  
    password = CharField(max_length=128, verbose_name=u'密码')  
    role = CharField(max_length=32, index=True, verbose_name=u'用户角色类型')  
    is_superuser = BooleanField(default=False, verbose_name=u'超级用户')  
    is_staff = BooleanField(default=False, verbose_name=u'员工账户')  
    is_active = BooleanField(default=True, index=True, verbose_name=u'是否激活')  
    email = CharField(max_length=75, null=True, index=True, verbose_name=u'邮箱')  
    create_time = BigIntegerField(verbose_name=u'创建时间')  
    last_login = BigIntegerField(verbose_name=u'上次登录时间')  

    USERNAME_FIELD = 'id'  
    REQUIRED_FIELDS = ['username', 'password']  

    def __unicode__(self):  
        return self.username  

    def is_anonymous(self):  
        return False  

    def is_authenticated(self):  
        return True  

    def get_full_name(self):  
        return self.real_name  

    def get_short_name(self):  
        return self.real_name  

    def get_username(self):  
        return self.username  

    def set_password(self, raw_password):  
        self.password = make_password(raw_password)  

    def check_password(self, raw_password):  
        """  
        Returns a boolean of whether the raw_password was correct. Handles  
        hashing formats behind the scenes.  
        """  
        def setter(raw_password):  
            self.set_password(raw_password)  
            self.save()  
        return check_password(raw_password, self.password, setter)  

    def set_unusable_password(self):  
        # Sets a value that will never be a valid hash  
        self.password = make_password(None)  

    def has_usable_password(self):  
        return is_password_usable(self.password)

编写一个 Auth backend

Django Auth 功能支持自定义 auth backend,正好是我们需要的,这里给出 auth backend 代码示例,假设文件名为 backends.py

from peewee import DoesNotExist  

from accounts.models import User  


class PeeweeModelBackend(object):  
    """  
    Peewee model auth backend  
    """  
    supports_object_permissions = False  

    def authenticate(self, username=None, password=None):  
        try:  
            user = User.get(User.username == username)  
        except DoesNotExist:  
            return None  
        if password and user.check_password(password):  
            user.backend = 'accounts.backends.PeeweeModelBackend'  
            return user  
        return None  

    def get_user(self, user_id):  
        try:  
            return User.get(User.id == user_id)  
        except DoesNotExist:  
            return None

然后在 Django 的 settings 中加入 auth backend 的配置:

AUTHENTICATION_BACKENDS = (  
    'accounts.backends.PeeweeModelBackend',  
)

未完待续

更新时间:2015-08-09 21:40

Review of 2014

好久没有打理博客,看着 RSS 阅读器里订阅的博客的更新,想起来也是时候写一写年终总结了。

年初的时候,我还是个大三学生,现在,已然是大四狗了。不上课了,找了一份实习工作,做着 Python 相关的开发工作,从 Web 后端开发工程师向着系统工程师和架构师的方向发展ing。(虽然还是个渣……)

年度美剧

  • 大西洋帝国 —— You can't be half a gangster
  • Friends —— 看了第二遍,准备再看第三遍、第四遍……

还有一些新剧诸如绿箭侠、闪电侠、神盾局特工等,纯粹打发时间。

年度书籍

书倒是读了一些,不过还有一堆买了没看的书,真是那啥“买书如山倒,读书如抽丝”哎~

年度数码产品

  • MacBook Air Early 2014 —— Apple 大法好!(不买 Retina MacBook Pro 就是因为没钱!!!Orz)
  • iPhone 5s —— 终于把 iPhone 4s 给换了,慢得受不了了
  • Kindle Paperwhite 2 —— 把旧的 Kindle 4 送给了表弟

年度技术相关

  • Django —— 工作系统用的 Django 开发的,很好用
  • pandas —— Python 数据分析神器
  • Celery —— Python 异步任务队列框架

主要用的 Python,也在关注一个小众的语言 Rust,2015 年快要出 1.0.0 正式版了,比较期待。

感情相关

年初的时候谈了个女朋友,在一起了几个月然后分手了,也许和有些人就是不适合吧,分手了反而觉得解脱。

2015 展望

  • 多看点书
  • 再刷一遍美剧 Friends
  • 技术上再深入一点,复习巩固下数据结构和算法的知识
  • 也许再找个女朋友?

OpenCV 与 PIL 图像转换

OpenCV 的 Python binding cv2 库使用 numpy 数组保存图片数据,当需要使用 PIL/Pillow 来处理图片时,需要将 cv2 读取的图片格式转换成 PIL/Pillow 的 Image:

import cv2  
from PIL import Image  

cv2_img = cv2.imread('test.jpg')  
pil_img = Image.fromarray(cv2_img)

处理好之后转回 cv2 格式(numpy array):

import cv2  
import numpy  
from PIL import Image  

cv2_img = cv2.imread('test.jpg')  
pil_img = Image.fromarray(cv2_img)  

# process image with PIL/Pillow as you like...  

img = numpy.array(pil_img, dtype=numpy.uint8)

很简单对吧。

Django 常用测试方法

辅助测试工具

# -*- coding: utf-8 -*-  
# filename: objectdict.py  
class ObjectDict(dict):  

    def __getattr__(self, key):  
        if key in self:  
            return self[key]  
        return None  

    def __setattr__(self, key, value):  
        self[key] = value

View decorators 测试

View decorators 即视图函数装饰器,常用来进行权限控制等等。如:

# -*- coding: utf-8 -*-  
# filename: decorators.py  
from functools import wraps  
from django.http import HttpResponseForbidden  


def active_required(func):  
    @wraps(func)  
    def wrapper(request, *args, **kwargs):  
        if not request.user.is_active:  
            return HttpResponseForbidden("账户被禁言!")  
        return func(request, *args, **kwargs)  
    return wrapper

上面的装饰器对当前登录用户的 active 状态进行了限制,如何写 test case 呢?方法如下:

# -*- coding: utf-8 -*-  
from django.test import TestCase  
from django.test.client import RequestFactory  
from django.http import HttpResponse  
from objectdict import ObjectDict  
from decorators import active_required  


def view(request):  
    return HttpResponse('Hello World!')  


class DecoratorsTest(TestCase):  
    def setUp(self):  
        self.factory = RequestFactory()  
        self.request = self.factory.get('/')  

    def test_active_required(self):  
        # fake active user  
        self.request.user = ObjectDict(is_active=True)  
        decorated = active_required(view)  
        response = decorated(self.request)  
        self.assertEqual(response.status_code, 200)  
        # fake inactive user  
        self.request.user = ObjectDict(is_active=False)  
        response = decorated(self.request)  
        self.assertEqual(response.status_code, 403)

如上代码,使用 django.test.client.RequestFactory 构造了一个 factory,调用 factory 的 get 方法获得了一个模拟的 request 对象。前面定义了一个测试视图函数 view ,通过调用 view(self.request) 即可模拟视图函数的调用过程,而通过 active_required(view)(request) 即模拟了加了装饰器之后的视图函数的调用过程。调用后返回模拟的 HttpResponse 对象,可以通过 response 的 status_code、content 等属性测试结果是否和预期一致。

Middleware 测试

假设我们有如下 middleware 定义:

# -*- coding: utf-8 -*-  
# filename: middlewares.py  
from django.http import HttpResponseRedirect  


class HelloMiddleware(object):  
    def process_request(self, request):  
        msg = request.GET.get('msg', None)  
        if msg:  
            print(msg)  
        else:  
            return HttpResponseRedirect(request.get_full_path() + '?msg=Hello')

则 Test case 写法:

# -*- coding: utf-8 -*-  
from django.test import TestCase  
from django.test.client import RequestFactory  
from django.http import HttpResponseRedirect  
from middlewares import HelloMiddleware  


class HelloMiddlewareTest(TestCase):  
    def setUp(self):  
        self.middleware = HelloMiddleware()  
        self.factory = RequestFactory()  

    def test_with_msg(self):  
        request = self.factory.get('/?msg=World')  
        self.assertEqual(self.middleware.process_request(request), None)  

    def test_without_msg(self):  
        request = self.factory.get('/')  
        self.assertIsInstance(self.middleware.process_request(request), HttpResponseRedirect)

通过调用 middleware 的 process_* 方法即可模拟 middleware 的处理过程,然后根据返回值进行测试。

Template tags 测试

假设有如下 template tag 定义:

# -*- coding: utf-8 -*-  
# filename: helloworld.py  
from django.template import Library  

register = Library()  

@register.simple_tag()  
def hello():  
    return 'hello world!'  


@register.simple_tag(takes_context=True)  
def world(context, msg):  
    return 'hello %s' % msg

则测试方法为:

# -*- coding: utf-8 -*-  
from django.test import TestCase  
from django.template import Template, Context  


class TemplateTagsTest(TestCase):  
    def test_hello(self):  
        out = Template(  
            '{% load helloword %}'  
            '{% hello %}'  
        ).render(Context({}))  
        self.assertEqual('hello world!', out)  

    def test_world(self):  
        out = Template(  
            '{% load helloword %}'  
            '{% world msg %}'  
        ).render(Context({  
            'msg': 'world!',  
        }))  
        self.assertEqual('hello world!', out)

如上,通过构造 Template 对象并调用 render 方法来获得渲染结果并予以测试。如果 template tag 函数中需要用到 request 对象,可将上面的 Context 换成 RequestContext(先用 RequestFactory 构造 request)。

Template filter 测试

Template filter 其实就是个普通的 Python 函数,可以直接通过调用测试返回值即可。当然,如果你不嫌麻烦,也可以用和测试 template tags 一样的方法构造 Template 对象后 render 测试渲染结果和预期是否一致。

Management Commands 测试

Management commands 是提供给 manage.py 进行命令行调用的,在代码中可通过 django.core.management.call_command 函数调用相应的命令,然后测试预期的产生的影响。

这里的问题主要在于 management commands 中可能使用到了 input/raw_input/getpass.getpass 等需要用户交互输入的函数导致不好写 testcase.

解决方法也简单,利用和 gevent.monkey 类似的 Python 模块的 monkey patch 方法即可。在代码中不直接调用 input/raw_input/getpass.getpass 而是先在其所在模块中定义它们的 wrapper 后通过调用这些 wrapper 实现功能,此后在 test case 中就可以先将这些 wrapper monkey patch 后再调用 call_command.

View Function 测试

直接通过 TestCase 内置的 client 模拟请求测试即可,不多说。

MongoDB 作为数据库的测试

如果使用 MongoDB 作为 Django 的后端数据库,在 TestCase 中要自动建立测试数据库的话,可继承 TestCase 写个自定义的新 TestCase 解决:

# -*- coding: utf-8 -*-  
from django.conf import settings  
from django.test import TestCase as DjangoTestCase  
from pymongo import MongoClient  


class TestCase(DjangoTestCase):  
    __dbname__ = 'test'  

    def _pre_setup(self):  
        super(TestCase, self)._pre_setup()  
        client = MongoClient(settings.MONGODB_HOST, settings.MONGODB_PORT)  
        self.db = client[self.__dbname__]  

    def _post_teardown(self):  
        super(TestCase, self)._post_teardown()  
        for collection in self.db.collection_names():  
            if collection == 'system.indexes':  
                continue  
            self.db.drop_collection(collection)

然后需要用到测试数据库的 test case 类从这个自定义的 TestCase 继承即可。

« Older posts

Copyright © 2015 Messense

Theme by Anders NorenUp ↑