分类目录归档:Python

Django admin export inline sub models

Django 导入导出大部分情况都可以使用django-import-export扩展,但是实际业务中经常需要导出子模型数据,比如导出订单同时需要导出订单商品信息。django-import-export扩展默认不支持导出inline子模型,这时候需要自己实现resource的export方法:

自定义django-import-export的export方法

class OrderResource(resources.ModelResource):
    class Meta:
        model = Order

    def export(self, queryset=None, *args, **kwargs):
        if queryset is None:
            queryset = self.get_queryset()
        ds = tablib.Dataset()
        data = []
        for order in queryset:
            for item in order.items.all():
                row = {
                    'order_id': order.id,
                    "item_id": item.idd,
                }
                data.append(row)
        ds.dict = data
        return ds

参考

django-q — 比django-celery和django-rq更简单又不失强大的django队列

django-q介绍

Django Q is a native Django task queue, scheduler and worker application using Python multiprocessing.

Features

  • Multiprocessing worker pools
  • Asynchronous tasks
  • Scheduled, cron and repeated tasks
  • Signed and compressed packages
  • Failure and success database or cache
  • Result hooks, groups and chains
  • Django Admin integration
  • PaaS compatible with multiple instances
  • Multi cluster monitor
  • Redis, Disque, IronMQ, SQS, MongoDB or ORM
  • Rollbar and Sentry support

Django Q is tested with: Python 3.7 and 3.8, Django 2.2.x and 3.1.x

安装django-q模块

pip install django-q

增加 django_q 到 INSTALLED_APPS:

修改settings.py

INSTALLED_APPS = (
    # other apps
    'django_q',
)

创建数据库表:

运行 Django migrations

$ python manage.py migrate

配置一个broker

使用django orm数据库作为broker

Q_CLUSTER = {
    'name': 'DjangORM',
    'workers': 1,
    'timeout': 90,
    'retry': 120,
    'queue_limit': 50,
    'bulk': 10,
    'orm': 'default'
}

使用redis作为broker

Q_CLUSTER = {
    'redis': 'redis://h:asdfqwer1234asdf@ec2-111-1-1-1.compute-1.amazonaws.com:111'
}

使用django_redis作为broker

Q_CLUSTER = {
    'name': 'DJRedis',
    'workers': 4,
    'timeout': 90,
    'django_redis': 'default'
}

其他broker: https://django-q.readthedocs.io/en/latest/brokers.html

使用qcluster处理异步任务

python manage.py qcluster

增加异步任务到队列

from django_q.tasks import async_task, result

# create the task
async_task('math.copysign', 2, -2)

# or with import and storing the id
import math.copysign

task_id = async_task(copysign, 2, -2)

# get the result
task_result = result(task_id)

# result returns None if the task has not been executed yet
# you can wait for it
task_result = result(task_id, 200)

# but in most cases you will want to use a hook:

async_task('math.modf', 2.5, hook='hooks.print_result')

# hooks.py
def print_result(task):
    print(task.result)

管理后台页面

Django Q不使用自定义页面,而是默认使用Django模型管理员提供的功能。当您打开Django Q的管理页面时,您将看到三种模型:

成功的任务

意味着他们在执行过程中没有遇到任何错误。

file

失败的任务

失败的任务遇到错误,阻止其完成执行。

计划任务

在这里,您可以检查计划任务的状态,创建,编辑或删除它们。

file

排队的任务

仅当您使用Django ORM代理时才出现

参考

通过Xvfb在命令行界面运行GUI程序

wkhtmltopdf使用的qt版本比较久,不支持很多css3和html5特性,于是用qt写了一个html转PDF工具,需要常驻后台运行,采用supervisord守护,无需安装图形界面,这时候需要用到Xvfb。

Xvfb介绍

Xvfb or X virtual framebuffer is a display server implementing the X11 display server protocol. In contrast to other display servers, Xvfb performs all graphical operations in virtual memory without showing any screen output.

简单说就是Xvfb是一个虚拟的X11显示服务,在虚拟内存中执行不需要显示图像。可以用来做远程左面显示,以及将桌面程序在非桌面环境使用(如headless的图形界面测试)。

Xvfb基本使用

安装xvfb

sudo apt-get install xvfb

命令行运行虚拟桌面

Xvfb :2 -screen 0 1024x768x16 &
x11vnc -listen 0.0.0.0 -rfbport 5900 -noipv6 -passwd xxxxxx -display 2
export DISPLAY=:2

使用Python xvfbwrapper

from xvfbwrapper import Xvfb

with Xvfb(width=1920, height=1024, colordepth=24) as xvfb:
    # launch stuff inside virtual display here.
    # Xvfb will stop when this block completes

参考

Django工作流(状态管理)

本文通过简单的例子比较了django状态管理扩展的使用,涵盖django-fsm,xworkflows,django_transitions以及参考

django-fsm 使用例子

https://github.com/viewflow/django-fsm

state = FSMField(
    default=State.DRAFT,
    verbose_name='Publication State',
    choices=State.CHOICES,
    protected=True,
)
@transition(field=state, source=[State.APPROVED, State.EXPIRED],
    target=State.PUBLISHED,
    conditions=[can_display])
def publish(self):
    '''
    Publish the object.
    '''
    email_the_team()
    update_sitemap()
    busta_cache()

django-fsm-admin

https://github.com/gadventures/django-fsm-admin

from fsm_admin.mixins import FSMTransitionMixin

class YourModelAdmin(FSMTransitionMixin, admin.ModelAdmin):
    # The name of one or more FSMFields on the model to transition
    fsm_field = ['state',]
    readonly_fields = ['state', ]

admin.site.register(YourModel, YourModelAdmin)

xworkflows

https://github.com/rbarrois/xworkflows

import xworkflows

class MyWorkflow(xworkflows.Workflow):
    # A list of state names
    states = (
        ('foo', "Foo"),
        ('bar', "Bar"),
        ('baz', "Baz"),
    )
    # A list of transition definitions; items are (name, source states, target).
    transitions = (
        ('foobar', 'foo', 'bar'),
        ('gobaz', ('foo', 'bar'), 'baz'),
        ('bazbar', 'baz', 'bar'),
    )
    initial_state = 'foo'

class MyObject(xworkflows.WorkflowEnabled):
    state = MyWorkflow()

    @xworkflows.transition()
    def foobar(self):
        return 42

    # It is possible to use another method for a given transition.
    @xworkflows.transition('gobaz')
    def blah(self):
        return 13

django_transitions

LiveStatus Status

class LiveStatus(StatusBase):
    """Workflow for Lifecycle."""

    # Define the states as constants
    DEVELOP = 'develop'
    LIVE = 'live'
    MAINTENANCE = 'maintenance'
    DELETED = 'deleted'

    # Give the states a human readable label
    STATE_CHOICES = (
        (DEVELOP, 'Under Development'),
        (LIVE, 'Live'),
        (MAINTENANCE, 'Under Maintenance'),
        (DELETED, 'Deleted'),
    )

    # Define the transitions as constants
    PUBLISH = 'publish'
    MAKE_PRIVATE = 'make_private'
    MARK_DELETED = 'mark_deleted'
    REVERT_DELETED = 'revert_delete'

    # Give the transitions a human readable label and css class
    # which will be used in the django admin
    TRANSITION_LABELS = {
        PUBLISH : {'label': 'Make live', 'cssclass': 'default'},
        MAKE_PRIVATE: {'label': 'Under maintenance'},
        MARK_DELETED: {'label': 'Mark as deleted', 'cssclass': 'deletelink'},
        REVERT_DELETED: {'label': 'Revert Delete', 'cssclass': 'default'},
    }

    # Construct the values to pass to the state machine constructor

    # The states of the machine
    SM_STATES = [
        DEVELOP, LIVE, MAINTENANCE, DELETED,
    ]

    # The machines initial state
    SM_INITIAL_STATE = DEVELOP

    # The transititions as a list of dictionaries
    SM_TRANSITIONS = [
        # trigger, source, destination
        {
            'trigger': PUBLISH,
            'source': [DEVELOP, MAINTENANCE],
            'dest': LIVE,
        },
        {
            'trigger': MAKE_PRIVATE,
            'source': LIVE,
            'dest': MAINTENANCE,
        },
        {
            'trigger': MARK_DELETED,
            'source': [
                DEVELOP, LIVE, MAINTENANCE,
            ],
            'dest': DELETED,
        },
        {
            'trigger': REVERT_DELETED,
            'source':  DELETED,
            'dest': MAINTENANCE,
        },
    ]

Model:

class Lifecycle(LifecycleStateMachineMixin, models.Model):
    """
    A model that provides workflow state and workflow date fields.

    This is a minimal example implementation.
    """

    class Meta:  # noqa: D106
        abstract = False

    wf_state = models.CharField(
        verbose_name = 'Workflow Status',
        null=False,
        blank=False,
        default=LiveStatus.SM_INITIAL_STATE,
        choices=LiveStatus.STATE_CHOICES,
        max_length=32,
        help_text='Workflow state',
    )

    wf_date =  models.DateTimeField(
        verbose_name = 'Workflow Date',
        null=False,
        blank=False,
        default=timezone.now,
        help_text='Indicates when this workflowstate was entered.',
    )

Admin

# -*- coding: utf-8 -*-
"""Example django admin."""

from django_transitions.admin import WorkflowAdminMixin
from django.contrib import admin

from .models import Lifecycle

class LifecycleAdmin(WorkflowAdminMixin, admin.ModelAdmin):
    """
    Minimal Admin for Lifecycles Example.

    You probably want to make the workflow fields
    read only so yo can not change these values
    manually.

    readonly_fields = ['wf_state', 'wf_date']
    """

    list_display = ['wf_date', 'wf_state']
    list_filter = ['wf_state']

admin.site.register(Lifecycle, LifecycleAdmin)

参考

著作权代码收集工具

使用Python编写,可以设置源码目录,排除目录,收集文件类型,输出目标文件,是否允许空白,编码等。

使用示例:

python zhuzuoquan.py -s D:\pythoncode\projectname \
-e debug,release \
-t .h,.cpp \
-en gb18030 \
-o result.txt

zhuzuoquan.py 代码

#!/usr/bin/python
# -*- coding: utf-8 -*-

import argparse
import os

parser = argparse.ArgumentParser(
    description='著作权代码收集工具', usage='''
    zhuzuoquan.py -s 源码目录 -e 排除目录 -t 文件类型 -o 输出文件
    ''')
parser.add_argument('-s', help='源文件目录', required=True)
parser.add_argument('-e', help='排除目录,多个用半角逗号分割', )
parser.add_argument('-t', help='扩展名,多个用半角逗号分割', required=True)
parser.add_argument('-o', help='输出文件')
parser.add_argument('-el', help='允许空白行')
parser.add_argument('-en', help='编码', default='utf-8')

def filter_files(root_dir, excluded_dirs, allow_types):
    valid_files = []
    for root, dirs, files in os.walk(root_dir):

        excluded = False
        d = root[root_dir.__len__():]
        for ex in excluded_dirs:
            if d.startswith(ex):
                excluded = True
                continue
        if excluded:
            continue

        for f in files:
            allow = False
            for t in allow_types:
                if f.endswith(t):
                    allow = True
            if allow:
                filepath = os.path.join(root, f)
                valid_files.append(filepath)
    return valid_files

if __name__ == '__main__':
    args = parser.parse_args()

    if args.e:
        excluded_dirs = args.e.split(',')
    else:
        excluded_dirs = []

    allow_types = args.t.split(',')

    root_dir = args.s

    output_file = args.o

    allow_empty_line = args.el

    encoding = args.en

    files = filter_files(root_dir, excluded_dirs, allow_types)
    total_files = total_lines = 0
    if output_file:
        with open(output_file, encoding=encoding, mode='w') as op:
            op.write("")

    for f in files:
        total_files = total_files + 1
        try:
            with open(f, encoding=encoding) as fh:
                content = fh.read()
                if not allow_empty_line:
                    content = "\n".join([s for s in content.splitlines() if s.strip()])
                lines = len(content.splitlines())
                relative_path = f[root_dir.__len__():]
                print(relative_path, lines)
                total_lines = total_lines + lines
                if output_file:
                    fh.seek(0)
                    with open(output_file, encoding=encoding, mode='a+') as op:
                        op.write(relative_path + ":\n\n" + content + "\n\n\n")
        except:
            pass

    print("Total: files %d, lines %d" % (total_files, total_lines))

Matplotlib使用中文字体(linux)

查看系统已有中文字体

fc-list :lang=zh

查询matplotlib默认配置示例

import matplotlib
matplotlib.matplotlib_fname()

查看当前系统matplotlib配置文件路径

import matplotlib
matplotlib.get_configdir()

全局修改

在系统如今下新增matplotlibrc文件,然后修改font.sans-serif,添加中文字体

font.sans-serif : WenQuanYi Micro Hei, Bitstream Vera Sans, ...

只修改当前程序

import matplotlib
matplotlib.rcParams['font.sans-serif'] = 'WenQuanYi Micro Hei'

参考

Python进入交互模式4种方法

使用python -i参数

添加-i参数

python -i myapp.py

使用code.interact()

    import code
    # before
    code.interact()
    # after

使用pdb lib

for thing in set_of_things:
    import pdb;
    pdb.set_trace()
    do_stuff_to(thing)

使用IPython embed

from IPython import embed

for thing in set_of_things:
  embed()
  do_stuff_to(thing)

Python 修改PDF文档尺寸以及去除水印图片

# -*- coding: UTF-8 -*-

import sys
import os

from pdfrw import PageMerge, PdfReader, PdfWriter, IndirectPdfDict
import fitz


# resize
def adjust(page):
    info = PageMerge().add(page)
    x1, y1, x2, y2 = info.xobj_box
    viewrect = ((x2 - 421) / 2, (y2 - 595) / 2, 421, 595)
    page = PageMerge().add(page, viewrect=viewrect)
    return page.render()

fin, = sys.argv[1:]
fout = 'mid.' + os.path.basename(fin)
reader = PdfReader(fin)
writer = PdfWriter(fout)
for p in reader.pages:
    writer.addpage(adjust(p))
writer.trailer.Info = IndirectPdfDict(reader.Info or {})
writer.write()

# trip backgroud images
doc = fitz.open(fout)
for i in range(len(doc)):
    imglist = doc.getPageImageList(i)
    for img in imglist:
        xref = img[0]
        if xref==51:
            doc._deleteObject(xref)
        print(img)
doc.save('new.' + os.path.basename(fin))

Python通用logging方法:控制台输出以及文件自动切分

import logging
from logging.handlers import TimedRotatingFileHandler

import sys

def get_logger():
    logger = logging.getLogger()
    hd = logging.StreamHandler(sys.stdout)
    # hd = TimedRotatingFileHandler('mylog.log', when='D',maxBytes=10240000 backupCount=30)
    formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
    hd.setFormatter(formatter)
    logger.addHandler(hd)
    logger.setLevel(logging.DEBUG)
    return logger

django rest framework token验证指南

安装app

修改settings.py增加rest_framework.authtoken

INSTALLED_APPS = (
    'rest_framework',
    'rest_framework.authtoken',
    'myapp',
)

增加权限验证

REST_FRAMEWORK = {
    'DEFAULT_PERMISSION_CLASSES': (
        'rest_framework.permissions.IsAdminUser',
    ),
    'DEFAULT_AUTHENTICATION_CLASSES': (
        'rest_framework.authentication.SessionAuthentication',
        'rest_framework.authentication.TokenAuthentication',
    )
}

升级表可以看到增加了authtoken_token表

python manage.py migrate

添加url

编辑urls.py

from rest_framework.authtoken import views

urlpatterns = [
    path('admin/', admin.site.urls),
    url(r'^', include(router.urls)),
    url(r'^api-auth/', include('rest_framework.urls', namespace='rest_framework')),
    url(r'^api-token-auth/', views.obtain_auth_token),
]

获取token

http POST 127.0.0.1:8000/api-token-auth/ username=’admin’ password=’password’

{
    "token": "9d1ff379e5e380c143ceadb66dde26b2b09dd4ab"
}

查看验证

http GET 127.0.0.1:8000/users/

{
    "detail": "身份认证信息未提供。"
}

http GET 127.0.0.1:8000/users/ ‘Authorization:Token 9d1ff379e5e380c143ceadb66dde26b2b09dd4ab’

{
    "count": 1,
    "next": null,
    "previous": null,
    "results": [
        {
            "date_joined": "2018-09-12T22:46:35.919532+08:00",
            "id": 1,
            "url": "http://127.0.0.1:8000/users/1/",
            "username": "admin"
        }
    ]
}

参考