Django 进阶主题

主题	难度	前置知识
REST Framework	中级	Django 基础，序列化
异步视图	高级	Python 异步编程
Django Channels	高级	WebSocket，异步编程
Celery	中级	消息队列，任务调度
高级 ORM	中级	Django 模型，数据库

Django REST Framework (DRF)

Django REST Framework 是一个强大而灵活的工具包，用于构建 Web API。

安装和配置

# 安装 DRF

pip

install

djangorestframework

# settings.py

INSTALLED_APPS

= [

# ...

'rest_framework'

,

'rest_framework.authtoken'

,
]

# DRF 配置

REST_FRAMEWORK

= {

'DEFAULT_AUTHENTICATION_CLASSES'

: [

'rest_framework.authentication.SessionAuthentication'

,

'rest_framework.authentication.TokenAuthentication'

,
],

'DEFAULT_PERMISSION_CLASSES'

: [

'rest_framework.permissions.IsAuthenticated'

,
],

'DEFAULT_PAGINATION_CLASS'

:

'rest_framework.pagination.PageNumberPagination'

,

'PAGE_SIZE'

:

20

,
}

序列化器

# serializers.py

from

rest_framework

import

serializers

from

.models

import

Product, Category

class

CategorySerializer

(serializers.ModelSerializer):

class

视图和视图集

# views.py - 基于类的视图

from

rest_framework

import

viewsets, permissions, status

from

rest_framework.decorators

import

action

from

rest_framework.response

import

Response

from

.models

import

Product

from

.serializers

import

ProductSerializer

class

ProductViewSet

(viewsets.ModelViewSet):
    queryset = Product.objects.all()
    serializer_class = ProductSerializer
    permission_classes = [permissions.IsAuthenticatedOrReadOnly]

def

get_queryset

(

self

):

# 过滤查询集

queryset = Product.objects.filter(is_available=

True

)
category = self.request.query_params.get(

'category'

)

if

category:
queryset = queryset.filter(category_id=category)

return

queryset

@action(detail=

True

, methods=[

'post'

])

def

purchase

(

self

, request, pk=

None

):

# 自定义动作

product = self.get_object()

# 处理购买逻辑

return

Response({

'status'

:

'purchased'

})

# urls.py

from

django.urls

import

path, include

from

rest_framework.routers

import

DefaultRouter

from

.views

import

ProductViewSet

# 创建路由器并注册视图集

router = DefaultRouter()
router.register(

'products'

, ProductViewSet)

urlpatterns = [
path(

'api/'

, include(router.urls)),
path(

'api-auth/'

, include(

'rest_framework.urls'

)),
]

API 认证和权限

# 获取认证令牌

from

rest_framework.authtoken.views

import

ObtainAuthToken

from

rest_framework.authtoken.models

import

Token

from

rest_framework.response

import

Response

class

CustomAuthToken

(ObtainAuthToken):

def

post

(

self

, request, *args, **kwargs):
        serializer = self.serializer_class(
            data=request.data,
            context={

'request'

: request}
)
serializer.is_valid(raise_exception=

True

)
user = serializer.validated_data[

'user'

]
token, created = Token.objects.get_or_create(user=user)

return

Response({

'token'

: token.key,

'user_id'

: user.pk,

'email'

: user.email
})

异步视图和异步支持

Django 3.1+ 引入了对异步视图的官方支持，允许您编写高性能的异步代码。

异步视图基础

# 异步视图示例

from

django.http

import

JsonResponse

import

asyncio

import

httpx

async

def

async_api_call

(url):

async

with

httpx.AsyncClient()

as

client:
response =

await

client.get(url)

return

response.json()

async

def

async_view

(request):

# 并行执行多个异步任务

tasks = [
async_api_call(

'https://api.example.com/data1'

),
async_api_call(

'https://api.example.com/data2'

),
async_api_call(

'https://api.example.com/data3'

),
]
results =

await

asyncio.gather(*tasks)

return

JsonResponse({

'results'

: results})

# 同步视图中的异步调用

from

asgiref.sync

import

async_to_sync

def

sync_view_with_async

(request):
results = async_to_sync(

lambda

: asyncio.gather(*tasks))()

return

JsonResponse({

'results'

: results})

异步 ORM 操作

# Django 4.1+ 支持异步 ORM

from

django.http

import

JsonResponse

from

.models

import

Product

async

def

async_products_view

(request):

# 异步查询

products =

await

Product.objects.filter(
is_available=

True

    ).async()

# 异步迭代

product_list = []

async

for

product

in

products:
product_list.append({

'name'

: product.name,

'price'

: product.price
        })

return

JsonResponse({

'products'

: product_list})

# 异步数据库事务

from

django.db

import

transaction

async

def

async_create_product

(request):

async

with

transaction.atomic():
product =

await

Product.objects.acreate(
name=

"新产品"

,
price=

99.99

,
is_available=

True

)

return

JsonResponse({

'id'

: product.id})

注意: 异步 ORM 支持需要 Django 4.1+ 版本，并且某些数据库后端可能有特定要求。

异步性能对比

0ms 开始请求

+10ms → 异步 API 调用 1 开始

+10ms → 异步 API 调用 2 开始

+10ms → 异步 API 调用 3 开始

+100ms ← 所有 API 调用完成

+100ms 请求完成 (总时间: 100ms)

同步版本总时间: ~300ms (3倍!)

Django Channels

Django Channels 扩展了 Django 的能力，使其能够处理 WebSocket、聊天协议、IoT 协议等。

安装和配置

# 安装 Channels

pip

install

channels channels-redis

# settings.py

INSTALLED_APPS

= [

# ...

'channels'

,
]

# 配置 ASGI 应用

ASGI_APPLICATION

=

'myproject.asgi.application'

# 通道层配置（使用 Redis）

CHANNEL_LAYERS

= {

'default'

: {

'BACKEND'

:

'channels_redis.core.RedisChannelLayer'

,

'CONFIG'

: {

"hosts"

: [(

'127.0.0.1'

,

6379

)],
},
},
}

ASGI 配置

# asgi.py

import

os

from

channels.auth

import

AuthMiddlewareStack

from

channels.routing

import

ProtocolTypeRouter, URLRouter

from

channels.security.websocket

import

AllowedHostsOriginValidator

from

django.core.asgi

import

get_asgi_application

from

django.urls

import

path

from

.consumers

import

ChatConsumer

os.environ.setdefault(

'DJANGO_SETTINGS_MODULE'

,

'myproject.settings'

)

django_asgi_app = get_asgi_application()

application = ProtocolTypeRouter({

# HTTP 请求仍然由 Django 处理

"http"

: django_asgi_app,

# WebSocket 请求由 Channels 处理

"websocket"

: AllowedHostsOriginValidator(
        AuthMiddlewareStack(
            URLRouter([
                path(

"ws/chat/<room_name>/"

, ChatConsumer.as_asgi()),
            ])
        )
    ),
})

WebSocket 消费者

# consumers.py

import

json

from

channels.generic.websocket

import

AsyncWebsocketConsumer

from

channels.db

import

database_sync_to_async

from

django.contrib.auth.models

import

User

class

ChatConsumer

(AsyncWebsocketConsumer):

async

def

connect

(

self

):
self.room_name = self.scope[

'url_route'

][

'kwargs'

][

'room_name'

]
self.room_group_name =

f'chat_

{self.room_name}

'

# 加入房间组

await

self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )

await

self.accept()

async

def

disconnect

(

self

, close_code):

# 离开房间组

await

self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )

async

def

receive

(

self

, text_data):
text_data_json = json.loads(text_data)
message = text_data_json[

'message'

]
username = text_data_json[

'username'

]

# 保存消息到数据库

await

self.save_message(username, message)

# 发送消息到房间组

await

self.channel_layer.group_send(
            self.room_group_name,
            {

'type'

:

'chat_message'

,

'message'

: message,

'username'

: username
            }
        )

async

def

chat_message

(

self

, event):

# 发送消息到 WebSocket

await

self.send(text_data=json.dumps({

'message'

: event[

'message'

],

'username'

: event[

'username'

]
        }))

    @database_sync_to_async

def

save_message

(

self

, username, message):
user = User.objects.get(username=username)

# 保存消息逻辑

pass

前端 WebSocket 连接

// JavaScript WebSocket 客户端

const

chatSocket =

new

WebSocket(

'ws://'

+ window.location.host +

'/ws/chat/lobby/'

);

chatSocket.onmessage =

function

(e) {

const

data = JSON.parse(e.data);

const

message = data.message;

const

username = data.username;

// 显示消息

const

messageElement = document.createElement(

'div'

);
messageElement.innerHTML =

`<strong>

${username}

</strong>:

${message}

`

;
document.querySelector(

'#chat-log'

).appendChild(messageElement);
};

chatSocket.onclose =

function

(e) {
console.error(

'Chat socket closed unexpectedly'

);
};

// 发送消息

document

.querySelector(

'#chat-message-input'

).focus();

document

.querySelector(

'#chat-message-input'

).onkeyup =

function

(e) {

if

(e.keyCode === 13) {

// Enter 键

const

messageInputDom = document.querySelector(

'#chat-message-input'

);

const

message = messageInputDom.value;
        chatSocket.send(JSON.stringify({
            message: message,
            username:

'当前用户'

}));
messageInputDom.value =

''

;
}
};

提示: Channels 还支持 HTTP2、背景任务、定时任务等高级功能，非常适合构建实时应用。

常用第三方包

Django 生态系统拥有丰富的第三方包，可以大大加速开发过程。

必备第三方包

Django REST Framework

⭐ 23k+ 📦 1M+

强大的 Web API 工具包，提供序列化、视图、认证等功能。

pip install djangorestframework

Django Channels

⭐ 5k+ 📦 200k+

处理 WebSocket、HTTP2、协议等的扩展，支持实时功能。

pip install channels channels-redis

Celery

⭐ 20k+ 📦 2M+

分布式任务队列，用于处理异步任务和定时任务。

pip install celery django-celery-results

Django Debug Toolbar

⭐ 7k+ 📦 300k+

开发调试工具，显示请求/响应周期中的各种调试信息。

pip install django-debug-toolbar

Celery 配置示例

# celery.py

import

os

from

celery

import

Celery

# 设置 Django 默认设置模块

os.environ.setdefault(

'DJANGO_SETTINGS_MODULE'

,

'myproject.settings'

)

app = Celery(

'myproject'

)

# 使用 Django 的配置

app.config_from_object(

'django.conf:settings'

, namespace=

'CELERY'

)

# 自动发现任务

app.autodiscover_tasks()

@app.task(bind=

True

)

def

debug_task

(

self

):
print(

f'Request:

{self.request!r}

'

)

# settings.py

CELERY_BROKER_URL

=

'redis://localhost:6379/0'

CELERY_RESULT_BACKEND

=

'django-db'

CELERY_ACCEPT_CONTENT

= [

'json'

]

CELERY_TASK_SERIALIZER

=

'json'

CELERY_RESULT_SERIALIZER

=

'json'

CELERY_TIMEZONE

=

'Asia/Shanghai'

# tasks.py

from

celery

import

shared_task

from

django.core.mail

import

send_mail

from

django.utils

import

timezone

@shared_task

def

send_welcome_email

(user_id):

# 发送欢迎邮件的异步任务

send_mail(

'欢迎加入我们！'

,

'感谢您注册我们的服务。'

,

'noreply@example.com'

,
[user.email],
fail_silently=

False

,
)

@shared_task

def

process_data

(data):

# 模拟长时间运行的任务

import

time
time.sleep(10)

# 模拟处理时间

return

{

'processed'

:

True

,

'timestamp'

: timezone.now()}

高级功能

Django 提供许多高级功能，可以帮助您构建更复杂的应用程序。

自定义中间件

# middleware.py

import

time

from

django.utils.deprecation

import

MiddlewareMixin

from

django.core.cache

import

cache

class

TimingMiddleware

(MiddlewareMixin):

def

process_request

(

self

, request):
request.start_time = time.time()

def

process_response

(

self

, request, response):

if

hasattr(request,

'start_time'

):
duration = time.time() - request.start_time
response[

'X-Request-Duration'

] =

f'

{duration:.2f}

s'

return

response

class

RateLimitMiddleware

(MiddlewareMixin):

def

process_request

(

self

, request):

# 简单的速率限制

ip = request.META.get(

'REMOTE_ADDR'

)
key =

f'rate_limit:

{ip}

'

requests = cache.get(key,

0

)

if

requests >

100

:

// 每分钟 100 次请求

from

django.http

import

HttpResponseForbidden

return

HttpResponseForbidden(

'Rate limit exceeded'

)

cache.set(key, requests +

1

,

60

)

// 60 秒过期

# settings.py

MIDDLEWARE

= [

# ...

'myapp.middleware.TimingMiddleware'

,

'myapp.middleware.RateLimitMiddleware'

,

# ...

]

# signals.py - 自定义信号

from

django.dispatch

import

Signal

# 定义自定义信号

user_logged_in = Signal()
order_created = Signal()

# 信号处理器

def

send_login_notification

(sender, user, request, **kwargs):

# 发送登录通知

print(

f"用户

{user.username}

在

{request.META['REMOTE_ADDR']}

登录"

)

def

update_user_activity

(sender, user, **kwargs):

# 更新用户最后活动时间

user.last_activity = timezone.now()
user.save()

# 连接信号处理器

user_logged_in.connect(send_login_notification)
user_logged_in.connect(update_user_activity)

提示: 中间件和信号是 Django 强大的扩展机制，可以用于实现横切关注点。

高级 ORM 功能

# 复杂查询和优化

from

django.db.models

import

(
Q, F, Count, Sum, Avg, Subquery, OuterRef,
Case, When, Value, IntegerField
)

# 使用 F 表达式避免竞态条件

from

django.db.models

import

F
Product.objects.filter(id=product_id).update(
views=F(

'views'

) +

1

)

# 条件表达式

products = Product.objects.annotate(
price_category=Case(
When(price__lt=

50

, then=Value(

'cheap'

)),
When(price__lt=

200

, then=Value(

'medium'

)),
default=Value(

'expensive'

),
output_field=CharField(),
)
)

# 子查询

latest_prices = PriceHistory.objects.filter(
product=OuterRef(

'pk'

)
).order_by(

'-created_at'

).values(

'price'

)[:

1

]

products_with_latest_price = Product.objects.annotate(
latest_price=Subquery(latest_prices)
)

# 数据库特定功能

from

django.db.models

import

Func, JSONField

from

django.contrib.postgres.fields

import

JSONField

from

django.contrib.postgres.aggregates

import

JSONBAgg

# PostgreSQL JSON 查询

class

JsonExtract

(Func):
function =

'JSON_EXTRACT'

template =

"%(function)s(%(expressions)s, '%(key)s')"

def

__init__

(

self

, expression, key, **extra):
super().__init__(expression, key=key, **extra)

# 使用 JSON 字段

products_with_metadata = Product.objects.annotate(
color=JsonExtract(

'metadata'

,

'color'

)
).filter(color=

'red'

)

# 全文搜索 (PostgreSQL)

from

django.contrib.postgres.search

import

(
SearchVector, SearchQuery, SearchRank
)

vector = SearchVector(

'name'

, weight=

'A'

) +

\

SearchVector(

'description'

, weight=

'B'

)
query = SearchQuery(

'django'

)

results = Product.objects.annotate(
search=vector,
rank=SearchRank(vector, query)
).filter(search=query).order_by(

'-rank'

)

最佳实践和性能优化

性能优化技巧

数据库查询优化
- 使用 select_related() 和 prefetch_related()
- 避免 N+1 查询问题
- 使用 only() 和 defer() 限制字段
缓存策略
- 使用 Redis 或 Memcached
- 实现多级缓存
- 缓存模板片段
异步任务
- 将耗时操作移出请求-响应周期
- 使用 Celery 处理后台任务

代码组织最佳实践

应用结构
- 保持应用单一职责
- 使用自定义管理器和管理类
- 合理使用中间件和信号
API 设计
- 使用版本控制 (v1/, v2/)
- 实现适当的错误处理
- 提供清晰的文档
安全考虑
- 实施适当的认证和授权
- 验证和清理所有输入
- 使用 HTTPS 和安全的 Cookie

监控和调试

# 性能监控配置

LOGGING

= {

'version'

:

1

,

'handlers'

: {

'console'

: {

'class'

:

'logging.StreamHandler'

,
        },
    },

'loggers'

: {

'django.db.backends'

: {

'level'

:

'DEBUG'

,

'handlers'

: [

'console'

],
},
},
}

# Django Debug Toolbar 配置

DEBUG_TOOLBAR_PANELS

= [

'debug_toolbar.panels.history.HistoryPanel'

,

'debug_toolbar.panels.versions.VersionsPanel'

,

'debug_toolbar.panels.timer.TimerPanel'

,

'debug_toolbar.panels.settings.SettingsPanel'

,

'debug_toolbar.panels.headers.HeadersPanel'

,

'debug_toolbar.panels.request.RequestPanel'

,

'debug_toolbar.panels.sql.SQLPanel'

,

'debug_toolbar.panels.staticfiles.StaticFilesPanel'

,

'debug_toolbar.panels.templates.TemplatesPanel'

,

'debug_toolbar.panels.cache.CachePanel'

,

'debug_toolbar.panels.signals.SignalsPanel'

,

'debug_toolbar.panels.logging.LoggingPanel'

,

'debug_toolbar.panels.redirects.RedirectsPanel'

,

'debug_toolbar.panels.profiling.ProfilingPanel'

,
]

部署和生产环境

服务器配置
- 使用 Gunicorn 或 uWSGI
- 配置 Nginx 反向代理
- 启用 HTTPS 和 HTTP/2
监控和日志
- 设置应用性能监控 (APM)
- 配置结构化日志
- 设置错误跟踪
扩展性
- 使用负载均衡
- 实现数据库读写分离
- 使用 CDN 加速静态资源

持续学习: Django 生态系统不断发展，关注官方文档、社区博客和会议，持续学习新技术和最佳实践。

django教程

Django 进阶主题

Django 进阶开发

REST API

异步编程

第三方包

高级功能

进阶主题概览

学习路径

Django REST Framework (DRF)

安装和配置

序列化器

视图和视图集

API 认证和权限

异步视图和异步支持

异步视图基础

异步 ORM 操作

异步性能对比

Django Channels

安装和配置

ASGI 配置

WebSocket 消费者

前端 WebSocket 连接

常用第三方包

必备第三方包

Django REST Framework

Django Channels

Celery

Django Debug Toolbar

Celery 配置示例

高级功能

自定义中间件

高级 ORM 功能

最佳实践和性能优化

性能优化技巧

代码组织最佳实践

监控和调试

部署和生产环境