Django 基础知识汇总
使用
安装
pip install django
创建新项目
django-admin.py startproject my_project
创建新的App
# 在Django项目(my_project)的根目录下执行
python manage.py startapp my_app
创建用户
python manage.py createsuperuser
http://127.0.0.1:8888/admin
进入 shell 调试
python manage.py shell
from moments.models import Status
statuses = Status.objects.all()
print statuses.query
print statuses
statuses[0].user
启动服务
python manage.py runserver 80
数据库操作
配置
测试环境，没有mysql时，可以将 mysql 切换为 sqlite3
import os
from settings import APP_ID
from settings import BASE_DIR
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3', # 默认用mysql
'NAME': os.path.join(BASE_DIR, APP_ID),
...
}
}
升级数据库
# make new migrations
python manage.py makemigrations
# apply all migrations
python manage.py migrate
显示 SQL
$ python manage.py sqlmigrate my_app 0001
BEGIN;
--
-- Create model Status
--
CREATE TABLE "moments_status" ("id" integer NOT NULL PRIMARY KEY AUTOINCREMENT, "text" varchar(140) NOT NULL, "pics" varchar(400) NULL, "pub_time" datetime NOT NULL);
--
-- Create model WeChatUser
--
CREATE TABLE "moments_wechatuser" ("id" integer NOT NULL PRIMARY KEY AUTOINCREMENT, "motto" varchar(500) NULL, "pic" varchar(200) NULL, "region" varchar(50) NULL, "user_id" integer NOT NULL UNIQUE REFERENCES "auth_user" ("id"));
--
-- Add field user to status
--
ALTER TABLE "moments_status" RENAME TO "moments_status__old";
CREATE TABLE "moments_status" ("id" integer NOT NULL PRIMARY KEY AUTOINCREMENT, "text" varchar(140) NOT NULL, "pics" varchar(400) NULL, "pub_time" datetime NOT NULL, "user_id" integer NOT NULL REFERENCES "moments_wechatuser" ("id"));
INSERT INTO "moments_status" ("text", "pub_time", "user_id", "id", "pics") SELECT "text", "pub_time", NULL, "id", "pics" FROM "moments_status__old";
DROP TABLE "moments_status__old";
CREATE INDEX "moments_status_user_id_d38910b1" ON "moments_status" ("user_id");
COMMIT;
celery
启动
python manage.py celery worker --settings=settings -l info
python manage.py celerybeat --settings=settings
使用
- 异步任务
# home_application.celery_tasks 方法
@task()
def async_task(x, y):
"""
定义一个 celery 异步任务
"""
logger.error(u"celery 定时任务执行成功，执行结果：{:0>2}:{:0>2}".format(x, y))
return x + y
>>> from home_application.celery_tasks import async_task
>>> import datetime
>>> now = datetime.datetime.now()
>>> async_task.apply_async(args=[now.hour, now.minute], eta=now + datetime.timedelta(seconds=60))
<AsyncResult: 2350fd61-1f32-46c0-b9d9-da0de2a918e1>
输出如下：
[2019-07-25 15:21:40,658: INFO/MainProcess] Received task: home_application.celery_tasks.async_task[2350fd61-1f32-46c0-b9d9-da0de2a918e1] eta:[2019-07-25 15:22:32.038230]
[2019-07-25 15:22:32,746: ERROR/Worker-10] celery 定时任务执行成功，执行结果：15:21
[2019-07-25 15:22:32,771: INFO/MainProcess] Task home_application.celery_tasks.async_task[2350fd61-1f32-46c0-b9d9-da0de2a918e1] succeeded in 0.0256261270006s: 36
from home_application import celery_tasks
celery_tasks.async_task(5, 8) # 实时输出
celery_tasks.async_task.dely(5, 8) # 延时执行
celery_tasks.async_task.apply_async(args=(5, 8))
import datatime
celery_tasks.async_task.apply_async(args=(5, 8), eta=datatime.datatime.now() + datatime.timedeta(seconds=60)) # 指定时间执行
- 定时任务
@periodic_task(run_every=crontab(minute='*/1', hour='*', day_of_week="*"))
def task_name():
celery 添加 task 后，需要在 conf/default.py 中，添加 CELERY_IMPORTS
- 定时任务触发
# docker exec -it cbf /bin/bash
# su apps
# /cache/.py/env/bin/python /data/app/code/manage.py shell
>>> from ntds import celery_tasks
>>> celery_tasks.<periodic_task_name>.apply()
<EagerResult: c1adbc71-cc09-4901-818e-8281251178a1>
Django Admin
Django Admin Commands
应用能通过 manage.py 注册自定义活动，如调用django里models.py或者views.py的函数对数据库做些增删改查的操作，可以加入crontab做管理，也可以用django commands来搞。
示例：
tree test/
test/
├── __init__.py
├── admin.py
├── apps.py
├── management
│ ├── __init__.py
│ └── commands
│ ├── __init__.py
│ ├── delete_author.py
├── migrations
│ ├── 0001_initial.py
│ ├── __init__.py
├── models.py
├── tests.py
└── views.py
3 directories, 11 files
简单写一个函数，功能是删除test_author表的一条记录:
cat test/management/commands/delete_author.py
from django.core.management.base import BaseCommand, CommandError
from test.models import Author
class Command(BaseCommand):
help = 'delete author'
def add_arguments(self, parser):
parser.add_argument('author_id', nargs='+', type=int)
def handle(self, *args, **options):
for author_id in options['author_id']:
try:
author = Author.objects.get(pk=author_id)
except Author.DoesNotExist:
raise CommandError('Author "%s" does not exist' % author_id)
author.delete()
self.stdout.write(self.style.SUCCESS('Successfully delete author "%s"' % author_id))
执行
python manage.py delete_author 2
Successfully delete author "2"
ModelAdmin
作用：列表页优化和排序，admin.py如下：
from django.contrib import admin
class ArticleControl(admin.ModelAdmin):
# 显示的字段
list_display = ('title', 'body', 'auth', 'create_time', 'update_time')
# 搜索条件
search_fields = ('title',)
# 按字段排序 -表示降序
ordering = ('-create_time',)
# 每页显示10条
list_per_page = 10
# 可编辑字段
list_editable = ('auth',)
# 设置哪些字段可以点击进入编辑界面
list_display_links = ('title', 'body')
# 点右侧作者名称，快速找到相关内容
list_filter = ('auth', 'title')
# 时间分层
date_hierarchy = 'create_time'
# 控制actions的下拉框出现在页面的位置
actions_on_bottom = True
actions_on_top = True
# 是否在actions下拉框右侧显示选中的对象的数量
actions_selection_counter = True
# 注册Article表
admin.site.register(models.Article, ArticleControl)
或
from django.contrib import admin
class BaseControl(admin.ModelAdmin):
ordering = ('-id',)
list_per_page = 20
class ArticleControl(BaseControl):
# 显示的字段
list_display = ('title', 'body', 'auth', 'create_time', 'update_time')
# 注册Article表
admin.site.register(models.Article, ArticleControl)
工具
根据模型字段，生成："A", "B", "C",这种格式的字符串。
python manage.py shell
from home_application.models import *
def gen_doc(obj):
l = obj.__dict__["__doc__"].replace(" ", "").split("(")[1].split(")")[0].split(",")
print l
r = ""
for i in l:
r += "\""
r += i
r += "\", "
print r
gen_doc(Object)
其他
去重
def remove_duplicates(old_str):
old_list = old_str.split(',')
new_list = []
for a in old_list:
if a not in new_list:
new_list.append(a)
print ",".join(new_list)
remove_duplicates("a,b,c,a")
错误
升级问题
报 django.db.utils.InternalError: (1050, u"Table 'django_content_type' already exists") 错误：
File "/cache/.py/env/lib/python2.7/site-packages/pymysql/connections.py", line 1138, in read
first_packet = self.connection._read_packet()
File "/cache/.py/env/lib/python2.7/site-packages/pymysql/connections.py", line 906, in _read_packet
packet.check_error()
File "/cache/.py/env/lib/python2.7/site-packages/pymysql/connections.py", line 367, in check_error
err.raise_mysql_exception(self._data)
File "/cache/.py/env/lib/python2.7/site-packages/pymysql/err.py", line 120, in raise_mysql_exception
_check_mysql_exception(errinfo)
File "/cache/.py/env/lib/python2.7/site-packages/pymysql/err.py", line 115, in _check_mysql_exception
raise InternalError(errno, errorvalue)
django.db.utils.InternalError: (1050, u"Table 'django_content_type' already exists")
------FAILURE: Migrate Database------
[172.19.11.14]20190409-162750 105 [JOB FAILURE]
解决办法：
pip install -r requirements.txt -i http://<url>/simple --trusted-host <url>
# 1. 执行
python manage.py migrate --fake
# 2. 再次执行
python manage.py migrate