「阿里OSS」集成Django的基本配置及使用-2

发表于 2018-06-11 | 分类于 Web开发

前言：

之前曾经写过的「阿里OSS」集成Django的基本配置及使用，里边在使用OSS是通过Django的两个信号量（signals）去实现的，一个post_init获取保存动作前的图片，一个post_save获取保存动作后的图片.通过对比两张图片，如果发生变化的话则将图片进行上传。

实现的思路没什么问题，在后期的使用过程中主要有如下两个问题发生：

post_init信号会在除了保存动作以外的情况发生，这样一直对post_init进行监听，对整个代码执行的效率有影响；
如果项目中大量的ImageField都需要用到OSS的话，这样的信号量复用起来太糟糕。

解决思路：

通过自定义一个ImageField，从而更灵活去对OSS进行配置；
找到ImageField对象在进行save动作的方法，在其后上传图片即可。

实现：

自定义一个ImageField类：

import oss2

from django.db.models import ImageField
from django.conf import settings


class OssImageField(ImageField):

    def pre_save(self, model_instance, add):
        file = getattr(model_instance, self.attname)
        if file and not file._committed:  # 这行代码判断文件是否有改动
            file.save(file.name, file.file, save=False)

            # 以下代码块为自定义部分：上传到oss
            auth = oss2.Auth(settings.ACCESS_KEY_ID, settings.ACCESS_KEY_SECRET)
            bucket = oss2.Bucket(auth, settings.ALI_OSS_ENDPOINT, settings.ALI_BUCKET_NAME)
            bucket.put_object_from_file('media/' + file.name, file.path)

        return file

使用：

由于OssImageField继承的是django的model中的ImageField类，所以使用的时候就跟跟普通的Field一样使用，如下：

...

class Designer(models.Model):
    name = models.CharField(max_length=50, verbose_name="设计师名称")
    country = models.CharField(max_length=50, verbose_name="国籍")
    portrait = OssImageField(upload_to='designer/portrait/%Y/%m', verbose_name="肖像", help_text='尺寸500*500为宜')
    bg_image = OssImageField(upload_to='designer/bg_image/%Y/%m', verbose_name="背景图片", help_text='尺寸1000*1000为宜')
    desc = models.TextField(verbose_name='设计师描述')

    class Meta:
        verbose_name = '设计师'
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.name

...

参考：

「阿里OSS」集成Django的基本配置及使用-1

「Celery」django-celery-beat配置定时任务

发表于 2018-03-21 | 分类于 Web开发

环境信息：

python 3.6.2
Django 1.11.3
celery 4.1.0
django-celery-beat 1.1.1

准备工作：

安装redis；
启动redis；
安装celery；
Django集成Celery；
启动celery；

以上步骤参考：「Celery」集成Django的基本配置及使用，同时接下来django-celery-beat的配置是在上述链接的基础上进行配置。

django-celery-beat配置：

Django中一个‘incubator’项目的目录如下：

- incubator/
  - manage.py
  - incubator/
    - __init__.py
    - settings.py
    - urls.py
  - mails/  # 这是一个app
    - models.py
    - views.py
    - tasks.py

安装：
1
pip install -U django-celery-beat
如果项目中settings.py中设置了USE_TZ = False，那么建议使用源码安装，有一个bug需要修改源文件的方式进行解决。随后在项目根目录，进行migrate：
1
python manage.py migrate

配置 incubator/incubator/settings.py:

...

INSTALLED_APPS = (
      ...,
      'django_celery_beat',
)

...

# 指定任务调度器
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers.DatabaseScheduler'

配置 incubator/incubator/celery.py ：

如果在settings.py中设置了USE_TZ = False，在使用django-celery-beat的过程中可能会报错，所以需要在原本配好的基础上加一行app.now = datime：

from __future__ import absolute_import, unicode_literals
import os
import datetime

import django
from celery import Celery


os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'incubator.settings')
django.setup()  # 如果 tasks 里边涉及对 model 的操作，则需要加上 django.setup()

app = Celery('incubator')
app.now = datetime.datetime.now  # 解决时区的报错
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()　　# 自动加载各个app下边的tasks.py

解决时区报错：

按上述加了app.now = datetime.datetime.now之后会引发另一个报错，需要通过修改django-celery-beat源码解决，其须修改的文件路径如下：django_celery_beat/schedulers.py，其中ModelEntry类中的_default_now方法，如下：

class ModelEntry(ScheduleEntry):
    """Scheduler entry taken from database row."""
	
    ...
	
    def _default_now(self):
        now = self.app.now()
        # The PyTZ datetime must be localised for the Django-Celery-Beat
        # scheduler to work. Keep in mind that timezone arithmatic
        # with a localized timezone may be inaccurate.
    
        # 下边这行是原来的，注释掉
        # return now.tzinfo.localize(now.replace(tzinfo=None))
        return now  # 改成直接返回 now
	
...

配置 incubator/mails/tasks.py:

随便写个样例

from celery import shared_task
	
	
# @shared_task(name='new_task_name')  # 重命名可以方便任务调用
@shared_task
def do_something(*arg, **kwargs):  
    pass  # 后台可以灵活配置arg以及kwargs的参数，在运行的时候通过数据库传参

将django-celery-beat新增的model，配置到后台：

看django_celery_beat.model可以看到一共是有5个model，具体的功能可以看源码，也很容易懂，这里主要用的是IntervalSchedule、CrontabSchedule和PeriodicTask。因为这里用的是xadmin后台的框架，但后台框架不会影响定时任务的功能，就不赘述。

import xadmin
from django_celery_beat.models import IntervalSchedule, CrontabSchedule, PeriodicTask


class PeriodicTaskAdmin(object):
    list_display = ['name', 'task', 'interval', 'crontab']


class IntervalScheduleAdmin(object):
    pass


class CrontabScheduleAdmin(object):
    pass


xadmin.site.register(PeriodicTask, PeriodicTaskAdmin)
xadmin.site.register(IntervalSchedule, IntervalScheduleAdmin)
xadmin.site.register(CrontabSchedule, CrontabScheduleAdmin)

后台配置：

首先按需配置好interval、crontab：

随后再新建periodic task：

需要注意Task name字段，为任务的路径或者名字，如果按照上述步骤配置的话，Task name应为mails.tasks.do_something。如果使用了@shared_task(name='new_task_name')进行重命名，那么Task name应为new_task_name

另外，如果是带参的定时任务，可以通过Arguments、Keyword arguments进行传参。

启动Celery

在 incubator/ 下运行：

1	celery -A incubator beat -l info

注意：一定要在worker运行后，beat运行的任务才会执行

参考：

「阿里OSS」集成Django的基本配置及使用-1

发表于 2018-01-30 | 分类于 Web开发

环境信息：

python 3.6.2
Django 2.0.1

依赖库安装：

1	pip install oss2

说明：

主要的实现思路是通过model 的信号量（signals）监测model 的create 以及update 动作，随后把相应的文件上传至阿里服务器；
通过Django 的MEDIA_URL来配置上传文件的访问路径。

signals.py 的配置：

Django中一个app 内的目录如下：

- goods/  # 这是一个app
  - apps.py
  - models.py
  - views.py
  - signals.py

signals.py的编写：

import oss2
from django.db.models.signals import post_save, post_init
from django.dispatch import receiver
from project.settings import ACCESS_KEY_ID, ACCESS_KEY_SECRET, ALI_OSS_ENDPOINT, ALI_BUCKET_NAME

from goods.models import GoodsSeries


# 获取修改前的图片
@receiver(post_init, sender=GoodsSeries)
def get_original_image(instance, **kwargs):
    instance.__original_image = instance.image


# 根据图片更新情况上传图片
@receiver(post_save, sender=GoodsSeries)
def create_user(sender, instance=None, created=False, **kwargs):
    if not created and instance.__original_image == instance.image:
        # 图片没有更新
        pass
    else:
        # 新建对象或更新
        img = instance.image
        auth = oss2.Auth(ACCESS_KEY_ID, ACCESS_KEY_SECRET)
        img_path = './media/' + str(img)
        bucket = oss2.Bucket(auth, ALI_OSS_ENDPOINT, ALI_BUCKET_NAME)
        bucket.put_object_from_file('media/' + img.name, img_path)

在apps.py内导入signals：

from django.apps import AppConfig


class GoodsConfig(AppConfig):
    name = 'goods'
    verbose_name = "产品管理"

    def ready(self):
        import goods.signals

settings.py 的配置：

阿里OSS秘钥等其他基本配置（具体看官方文档）：

...
ACCESS_KEY_ID = 'access key id'
ACCESS_KEY_SECRET = 'access key secret'
ALI_OSS_ENDPOINT = 'oss-cn-shenzhen.aliyuncs.com'  # 具体看文档，这个是类似存放服务器地区的东西
ALI_BUCKET_NAME = 'bucket'  # 具体看文档，这个是类似仓库名的东西

配置MEDIA_URL：

...
# 这个URL需要根据 ALI_OSS_ENDPOINT 、 ALI_BUCKET_NAME 以及 signals.py 填的路径进行配置
MEDIA_URL = 'http://bucket.oss-cn-shenzhen.aliyuncs.com/media/'  
MEDIA_ROOT = os.path.join(BASE_DIR, 'media')

后续：

「阿里OSS」集成Django的基本配置及使用-2

参考：

阿里OSS Python-SDK使用说明

「Celery」集成Django的基本配置及使用

发表于 2018-01-19 | 分类于 Web开发

环境信息：

python 3.6.2
Django 1.9.8
celery 4.1.0

Redis安装及启动：

安装redis：

Mac 上通过 brew 进行安装
1
brew install redis
启动redis：
1
sudo /usr/local/bin/redis-server
查看redis进程及端口：

一般默认的端口是6379
1
ps aux|grep redis

Celery安装：

1	pip install -U celery[redis]

Django集成Celery配置：

Django中一个‘incubator’项目的目录如下：

- incubator/
  - manage.py
  - incubator/
    - __init__.py
    - settings.py
    - urls.py
  - mails/  # 这是一个app
    - models.py
    - views.py
    - tasks.py

配置 incubator/incubator/celery.py ：

from __future__ import absolute_import, unicode_literals
import os

import django
from celery import Celery


os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'incubator.settings')
django.setup()  # 如果 tasks 里边涉及对 model 的操作，则需要加上 django.setup()

app = Celery('incubator')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()　　# 自动加载各个app下边的tasks.py

配置 incubator/incubator/__init__.py:

from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ['celery_app']

配置 incubator/incubator/settings.py:

...

# 配置Broker
CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'  # 这里注意redis的端口

配置 incubator/mails/tasks.py：

异步发送邮件，同时保存发送邮件时的报错信息

from django.core.mail import send_mail
from celery import shared_task

from .models import EmailError


@shared_task
def send_emails(sub, content, to_email_list, from_email, is_active=True):
    if is_active:
        for mail in to_email_list:
            try:
                send_mail(sub, content, from_email, [mail])
            except Exception as e:
                EmailError.objects.create(error_msg=str(e), mail=mail, sub=sub, content=content)

    else:
        pass

使用

from mails.task import send_emails

...

send_emails.delay(sub, content, to_mail_list, from_email, is_active)  # 注意delay

启动Celery

在 incubator/ 下运行：

1	celery -A incubator worker -l info

参考：

Using Celery with Django

「Django富文本」django-ckeditor的使用

发表于 2018-01-11 | 分类于 Web开发

之前在副文本上一直使用百度家的UEditor，但是官方在Django以及python3.X上没有做维护，所以换上了ckeditor，使用下来还算方便。

环境信息：

python 3.6.2
Django 2.0.1
django-ckeditor 5.4.0

正文：

用pip安装ckeditor：
1
pip install django-ckeditor

在settings的INSTALLED_APPS注册：

INSTALLED_APPS = [
	...

    'ckeditor',
    'ckeditor_uploader',  # 如果富文本中允许用户上传图片/文件到后台，则需要加上
	
	...
]

配置上传文件路径（如果用到上传功能）：
ckeditor的上传路径是将media的路径后边拼接upload_path，也就是说按照下边的配置，最终的上传路径为`/media/uploads/
1
2
3
MEDIA_URL = '/media/'
MEDIA_ROOT = os.path.join(BASE_DIR, 'media')
CKEDITOR_UPLOAD_PATH = "uploads/"

富文本url配置

from django.urls import include, re_path


urlpatterns = [
    ...

    # 富文本
    re_path(r'^ckeditor/', include('ckeditor_uploader.urls')),  # Django2.0要用're_path'
    
    ...
]

在settings中按需定制自己的富文本面板（非必须）：
可以按照自己的喜好，定制富文本的编辑工具面板，下边这个是我花了点时间选出了一些常用以及实用的功能。其中具体参数含义可以到官方文档查询。

CKEDITOR_CONFIGS = {
    'default': {
        'toolbar_ToolbarConfig': [
            {'name': 'basic', 'items': [
                'Font', 'FontSize', 'BGColor', 'TextColor',
                '-',
                'JustifyLeft', 'JustifyCenter', 'JustifyRight', 'JustifyBlock',
                '-',
                'Bold', 'Italic', 'Underline', 'Strike', 'Undo', 'Redo',
                '-',
                'Strike', 'Subscript', 'Superscript',
                '-',
                'Image', 'Flash', 'Table', 'HorizontalRule', 'SpecialChar',
            ]},
            '/',
            {'name': 'custom', 'items': [
                'Source',
                '-',
                'Preview', 'Maximize',
            ]}
        ],
        'toolbar': 'ToolbarConfig',
        'width': '100%',
    }
}

field的使用方式：
这里的上传功能有个小坑，我开始一直没找着为什么不能开启上传功能

from django.db import models
from ckeditor.fields import RichTextField
from ckeditor_uploader.fields import RichTextUploadingField


class Post(models.Model):
    content_without_upload = RichTextField()  # 不带上传功能的富文本
    content_with_upload = RichTextUploadingField()  # 带上传功能的富文本（需要先在INSTALL_APP中声明）

效果：

后记：
在前后端分离的开发过程中，富文本内所插入的图片需要返回绝对路径，查了许多文档，ckeditor没有像百度的UEditor一样把url_prefix放出来，方便设置。后来我摸索了一下，改动了一下源码就可以实现功能了。文件路径ckeditor_uploader/util.py中的get_media_url(path)，具体如下：

def get_media_url(path):
    """
    Determine system file's media URL.
    """
    url_prefix = hasattr(settings, 'CKEDITOR_URL_PREFIX') and settings.CKEDITOR_URL_PREFIX or ''
    return url_prefix + default_storage.url(path)

随后只需要在settings中加入：

1	CKEDITOR_URL_PREFIX = 'https://actmerce.github.io'

参考：

django-ckeditor 文档

「环境配置笔记」Mac下安装mysqlclient

发表于 2018-01-02 | 分类于 Web开发

Mac在第一次安装mqsql的时候，一般通过pip install mysqlclient的命令去安装，然后会报一个EnvironmentError: mysql_config not found的错误，解决如下：

用brew安装mysql：
1
brew install mysql

安装brew的方法如下：

1	/usr/bin/ruby -e "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/master/install)"

「unittest开发笔记」芝士点备忘

发表于 2018-01-02 | 分类于基础

通过main()去运行测试

...

if __name__ == '__main__':
    unittest.main()

命令行接口方式进行测试

1
2
3

python -m unittest test_module1 test_module2
python -m unittest test_module.TestClass
python -m unittest test_module.TestClass.test_method

测试中主要通过3个方法进行判断是否正常：

assertEqual()去检查结果是否和期望的结果相同；
assertTrue()用于核实最终状态；
assertRaises()去检查是否抛出期望的异常。

还有其他的一些如下表：

Method	Checks that
assertEqual(a, b)	a == b
assertNotEqual(a, b)	a != b
assertTrue(x)	bool(x) is True
assertFalse(x)	bool(x) is False
assertIs(a, b)	a is b
assertIsNot(a, b)	a is not b
assertIsNone(x)	x is None
assertIsNotNone(x)	x is not None
assertIn(a, b)	a in b
assertNotIn(a, b)	a not in b
assertIsInstance(a, b)	isinstance(a, b)
assertNotIsInstance(a, b)	not isinstance(a, b)
assertRaises(exc, fun, args, *kwds)	fun(args, *kwds) raises exc
assertRaisesRegexp(exc, r, fun, args, *kwds)	fun(args, *kwds) raises exc and the message matches regex r

在TestCase的类里，setUp()会先被加载，然后再去运行需要测试的方法

import unittest

class CustomTestCase(unittest.TestCase):
    def setUp(self):
        pass

    ...

如果实在生产环境下运行的测试，在一般情况下，是在setUp()创建或更新一些运行测试必要的数据，整个测试运行完后，tearDown()会被执行，删除或恢复数据的操作可以放在tearDown()里边，如：

class ListApiTests(unittest.TestCase):
    def setUp(self):
        # 先创建一个测试的用户
        user_info = {
            'username': 'test_user_actmerce',
            'password': make_password('yoyoqiekenao'),
            'real_name': '晓测'
        }
        user = UserProfile.objects.create(**user_info)
        self.user = user

    def tearDown(self):
        # 删除新增的数据
        self.user.delete()
        
    def test_one(self):
        pass

参考：

「pandas笔记」对比两个dataframe

发表于 2017-12-18 | 分类于基础

今天撸代码撸的太难受了，pandas强大是强大，但是感觉国内用的不是太多，遇到一些问题不知道怎么去解决好，只能google。

今天遇到问题主要是对比两个不同时间、同个数据库导出的csv文件。因为本人这次主要用pandas去处理csv文件，那么随之而来的就是如何通过对比两个dataframe获取新的数据以及更新了的数据。

import chardet

import pandas as pd


def get_file_encoding(file_path):
    f = open(file_path, 'rb')
    file_encoding = chardet.detect(f.read())['encoding']
    f.close()
    return file_encoding


def get_df(file_path, id_name, field_list):
    # 获取file 的编码格式
    file_encoding = get_file_encoding(file_path)

    # 初始化df，包括把id_name替换成索引，以及替换nan为空，不然会有异常，最后再来一个过滤显示的字段
    df = pd.read_csv(file_path, encoding=file_encoding).set_index([id_name])[field_list].fillna('').sort_index()
    return df


def contrast_df(new_path, old_path, id_name, field_list, save_file_path='./auto_create/'):
    # 把id的列移除list
    field_list.pop(field_list.index(id_name))

    # 获取file_name
    file_name = new_path.split('/')[-1]

    # 获取新旧的df
    new_df = get_df(new_path, id_name, field_list)
    old_df = get_df(old_path, id_name, field_list)

    # 比较数据是否有更新前，须要先看是否有新增的数据，不然size不同会报错
    old_index = old_df.index.values
    add_df = new_df[~new_df.index.isin(old_index)]
    if not add_df.empty:
        print('有{}条数据新插入了'.format(len(add_df)))
        # 把新增的数据从new_df 中删除
        add_index = add_df.index.values
        new_df = new_df[~new_df.index.isin(add_index)]
        
        # 新建csv文件
        add_df.to_csv(save_file_path + 'add_' + file_name)
    else:
        print('没有数据新插入')
        
    ne_stacked = (new_df != old_df).stack()
    changed = ne_stacked[ne_stacked]
    changed.index.names = ['ID_P', 'col']
    changed_index = set([seri[0] for seri in changed.index])
    update_df = new_df[new_df.index.isin(changed_index)]
    
    if not update_df.empty:
        print('有{}条数据更新'.format(len(update_df)))
        # 新建csv文件
        update_df.to_csv(save_file_path + 'update_' + file_name)
    else:
        print('没有数据更新')

    return add_df, update_df

基本思路是：

先把新增的数据移除，并保存成新的csv文件；
两个新旧的dataframe进行比较；
通过行列转化的形式获取changed,并从中获得发生改变的数据的index；
将更新的数据保存成新的csv文件。

最后补充一下：

上边的changed能做得更多，它还能识别数据哪个字段发生了改变；
中间还试了别的方法去完成这个功能，但是没有跑通，不过思路还是挺有趣的，代码丢上来mark一下

df_3 = pd.concat([new_df, old_df])
df_gpby = df_3.groupby(list(df_3.columns))
idx = [x[0] for x in df_gpby.groups.values() if len(x) == 1]
df_3.iloc[idx]

参考：

「xadmin开发笔记」后台添加数据时外键的回填

发表于 2017-12-06 | 分类于 Web开发

在xadmin中关于user一般都设置了relfield_style = 'fk-ajax'，项目中遇到有个实际情况就是，在新建一条用户的数据时，在编辑页面自动回填介绍人

— 未完待续 —

参考：

xadmin文档

「xadmin开发笔记」Admin中重载方法可实现的一些功能

发表于 2017-12-05 | 分类于 Web开发

方便举例，现在有个编辑教师的Admin

class TeacherAdmin(object):
	pass
	
xadmin.site.register(Teacher, TeacherAdmin)

在List页面将数据过滤显示

可以通过重载queryset()去实现列表的过滤

class TeacherAdmin(object):
    def queryset(self):
        qs = super(TeacherAdmin, self).queryset()
        qs = qs.filter(is_active=True)  # 过滤掉失效的数据
    return qs

现实项目中的用途可以用来过滤那些已经过期或者失效的数据。因为这个queryset()的方法不仅仅只是作用于List页面的过滤，还有其他的xadmin的功能，像关联字段的搜索。
同时，如果是有多个学校，学校的网站管理者只能操作管理员用户所属学校的教师数据，也可以在queryset()中实现。

class TeacherAdmin(object):
    def queryset(self):
        qs = super(TeacherAdmin, self).queryset()
        qs = qs.filter(is_active=True, school=self.user.school)
    return qs

将Edit页面中的外键进行过滤显示

可以通过重载get_context()去实现过滤
如果“A学校”跟“B学校”有很多不同的班级，须要对老师的grade的外键进行限制选择，那么就可以像这样，在编辑的页面对grade进行过滤。

class TeacherAdmin(object):
	...
    def get_context(self):
        context = super(TeacherAdmin, self).get_context()
        if "form" in context:
            context["form"].fields["grade"].queryset = Grade.objects.filter(
                school=self.user.school)  # 实现 grade 字段显示的过滤
        return context

上边这个get_context()只作用于普通的外键样式进行过滤，如果在GradeAdmin中对grade设置了relfield_style = 'fk-ajax'，那么须要用到上边提到的queryset()进行过滤了。

Edit页面`readonly_fields`、`exclude`的设置

在编辑页面设置某些只读的字段或者不显示的字段可以分别通过readonly_fields跟exclude去设置

class TeacherAdmin(object):
    readonly_fields = ('name', )
    exclude = ('school', )
    ...

但是如果须要根据自己的业务逻辑，根据当前操作管理员去动态改变该表单的字段的readonly_fields以及exclude的话，那么须要重载以下两个方法去实现功能

class TeacherAdmin(object):
    # readonly_fields = ('name', )
    # exclude = ('school', )
    
    # 动态设置readonly_fields
    def get_readonly_fields(self):
        if SUPERUSER in self.user.groups.all():  # 随便加入自己的逻辑
            readonly_fields = ()
        else:
            readonly_fields = ('name', )
        return readonly_fields
        
    # 动态设置 exclude
    def get_model_form(self, **kwargs):
        if SUPERUSER in self.user.groups.all():  # 随便加入自己的逻辑
            self.exclude = ()
        else:
            self.exclude = ('school', )
        return super(StudentProfileAdmin, self).get_model_form(**kwargs)
    
    ...

基本上通过上边的对From的过滤以及动态设置readonly_fields、exclude，可以满足到很大的一部分定制开发的需求了。
再加上xadmin中的inlines、重载save_models()定制Admin保存数据逻辑以及重载get_form_layout()完成页面样式配置，后端只需要修改小部分代码，就能实现很多功能。

参考：

xadmin文档

前言：

解决思路：

实现：

在List页面将数据过滤显示

将Edit页面中的外键进行过滤显示

Edit页面readonly_fields、exclude的设置

Edit页面`readonly_fields`、`exclude`的设置