# django

# 环境搭建 env

# 工具版本

  • macOS High Sierra 10.13.6
  • Python 3.8.3
  • Django 3.1.5
  • PyCharm 2020.2 (Community Edition)

PyCharm的断点调试特别方便

# 代码环境

  1. fork 官方django源码,拉取到本地,新建分支read_django_source
  2. 新建虚拟环境dj_source_env,软连接本地django源码到虚拟环境的site-packages
  3. 进入虚拟环境,新建调试项目mysite, 那么所依赖的django环境为 https://github.com/ni-ning/django.git,从而动态调试django源码
# 建立当前用户学习目录 learning
cd && mkdir learning


# 拉取fork源码,建立注解分支 read_django_source
git clone https://github.com/ni-ning/django.git
git fetch origin stable/3.1.x
git checkout stable/3.1.x
git checkout -b read_django_source


# 建立虚拟环境
cd ~/learning && mkdir envs
cd ~/learning/envs
virtualenv -p /Library/Frameworks/Python.framework/Versions/3.8/bin/python3 dj_source_env
source ~/learning/envs/dj_source_env/bin/activate


# django的注解分支 映射到 虚拟环境中方便断点调试
ln -s ~/learning/django/django ~/learning/envs/dj_source_env/lib/python3.8/site-packages/django


# 创建测试项目 mysite
python ~/learning/django/django/bin/django-admin.py startproject mysite


# 创建项目时需要额外安装依赖
pip install asgiref
pip install pytz
pip install sqlparse


# 建立快捷命令 vim ~/.bash_profile
alias dsource='source /Users/nining/learning/envs/dj_source_env/bin/activate'
alias cddjango='cd /Users/nining/learning/django/'
alias cdmysite='cd /Users/nining/learning/mysite/'

# 代码结构 structure

外层项目结构可以打包、测试、文档等,重点是内部的源码结构是我们重点研究的对象

# 项目结构

根据名称就可以推断目录的作用,可总结一下打包发布到官方pypi流程

  • setup.py 打包

distuils是标准库中负责建立Python第三方库的安装器,对于简单的分发很有用,但功能缺少。 _setuptools_是distuitls增强版, 不包括在标准库中, 是一个优秀的, 可靠的Python包安装与分发工具

本地开发者 --> 打包上传pypi服务器 --> 使用者直接 pip install <package>

# 1. 在用户目录下创建.pypirc文件。
# 添加权限:chmod 600 ~/.pypirc
[distutils]
index-servers = pypi

[pypi]
repository=https://upload.pypi.org/legacy/
username=******
password=******

# 2. 安装twine
pip install twine
# 3. 在项目根目录执行
python setup.py sdist bdist_wheel
# 4. 上传打包文件
twine upload dist/*
  • setup.py 常用配置

详细说明参考官文文档 (opens new window)

PS:被忽视的攻击面:Python package 钓鱼,https://paper.seebug.org/326/ (opens new window)

# 源码结构

django/django
- apps    # django的思想是把app作为独立单元来运作的,对外接口AppConfig和apps
- bin     # django-admin -> management.execute_from_command_line()
- conf    # app_template和project_template创建app和project模块;global_settings.py 全局默认配置;可以重点研究一下 __init__中的LazySettings
- contrib # 可插拔
    |- admin    # resister
    |- auth     # authenticate,login,logout,get_user_model,get_user
    |- sessions # Session
    |- ....
- core
    |- cache    # 缓存 内容变化不快 但访问量大的场景,如首页
    |- checks   # 如检查数据库配置是否可以正常连接
    |- files    # 文件上传相关
    |- handlers # wsgi实现
    |- mail     # 邮件
    |- management    # python manage.py <命令>
    |- serializers   # 序列化相关 queryset甚至可以为yaml
    |- servers       # HTTP server
    |- exceptions.py # 全局自定义异常
    |- paginator.py  # 分页实现逻辑
    |- signals.py    # 全局信号
    |- signing.py    # 签名
    |- validators.py # 相关通用正则验证
    |- wsig.py       # wsig 接口 get_wsgi_application
- db  # 数据库相关 重点研究
  |- backends # 策略支持不同数据库
  |- models   # 重点研究
- dispathch   # 如信号具体实现
- forms       # form表单映射
- http        # HttpRequest, HttpResponse, Http404, SimpleCookie 重点研究
- middleware  # 重点研究
- template
- templatetags
- test        # 单元测试
- urls        # 现代化的path, re_path, include
- utils       # 实用工具 重点研究
- views       # 视图 重点研究 catch csrf Class-based View
- shortcuts.py  # 快捷键
  • pip install本质是把django/django源码文件拷贝到site-package中

# 项目创建 project

创建项目 django-admin startproject mysite 和创建应用django-admin startapp app 基本流程

# django/bin/django-admin.py 命令脚本实际执行逻辑
from django.core import management
management.execute_from_command_line()
  • from django.core.management import execute_from_command_line 当这行代码开始执行时,首先运行django.core.management.__init__.py整个文件,找到execute_from_command_line函数并将其导入当前程序的命名空间中;
  • 不可忽略的是djanog.core.management.__init__.py也包含import语句

djanog.core.management.__init__.py

...
import django                   # 这里
from django.apps import apps    # 这里
from django.conf import settings

class ManagementUtility:
    def __init__(self, argv=None):
        pass
    def main_help_text(self):
        # 帮助信息
    def fetch_command(self, subcommand):
        pass
    def autocomplete(self):
        # 自动补全
    def execute(self):
        # 执行入口函数

def execute_from_command_line(argv=None):
    utility = ManagementUtility(argv)
    utility.execute()
  • import django 运行了文件 django.__init__.py

  • from django.apps import apps 运行了文件 django.apps.__init__.py

  • 整个django的开端就从这里开始

  • execute_from_command_line 工厂函数,负责指挥ManagementUtility类利用execute方法来解析参数和启动wsgi服务,重点在于self.fetch_command(subcommand).run_from_argv(self.argv)

    • self.fetch_command 利用django内置的命令管理区去匹配到具体的模块,self.fetch_command('startproject')相当于django.core.management.commands.startproject.Command()
    • run_from_argv(self.argv) app_or_project 从django/conf/模板经过Context渲染,然后shutil.copyfile

# 本地启动 runserver

python manage.py runserver

Django框架的整个数据流向、服务启动、端口监听等基础核心功能都是按照WSGI标准进行开发的

  • WSGI, Web Server Gateway Interface, Web服务器和Web框架或应用的协议接口
  • 主要作用是接收客户端(浏览器/软件/APP/调试程序)发送的请求和转发请求给上层服务(例如Django)
  • 采用select网络模型进行数据的接收和分发,可以利用操作系统的非堵塞和线程池等特性,因此非常高效,如开源项目uWSGI是C语言代码实现
  • Django采用wsgi是纯python代码实现,实际项目部署使用 uWSGI 或者 Gunicorn

# checks 校验

一个完整的功能设计可包括:

  • 功能定义 如django/core/checks/registry.py中定义 数据字典 class Tags, 定义类并实例化为功能函数,作为对外提供的接口
  • 功能使用 如django/core/checks/database.py中 使用 register,且要符合功能定义时规范,如返回列表issues
  • 嵌入到整体中,成为一部分,如self.check(display_num_erros)、self.check_migrations() 启动初始化校验

# 内部流程

  • self.fetch_command 利用django内置的命令管理区去匹配到具体的模块,self.fetch_command('runserver')相当于django.contrib.staticfiles.management.commands.runserver.Command(),django中的命令工具代码组织采用的是策略模式+接口模式,也就是说django.core.management.commands这个目录下面存在各种命令工具,每个工具下面都有一个Command接口,当匹配到’runserver’时调用’runserver’命令工具的Command接口,当匹配到’migrate’时调用’migrate’命令工具的Command接口
  • run_from_argv(self.argv) 初始化中间件、启动服务,也就是拉起wgsi
# 先捋清楚 继承关系链
WSGIServer -> simple_server.WSGIServer -> http.server.HTTPServer -> scokectserver.TCPServer -> socketserver.BaseServer

小结:对于开发测试环境runserver而言,Django统一实现的 WSGIHandler + SocketServer + App 对外提供服务;实际部署时nginx + uwsgi承担WSGI,而Django只是作为App

# runserver.py中 def inner_run中代码
try:
    handler = self.get_handler(*args, **options)
    run(self.addr, int(self.port), handler,
        ipv6=self.use_ipv6, threading=threading, server_cls=self.server_cls)
except OSError as e:
    pass
  • 初始化WSGIHandler时调用self.load_middleware()
  • WSGIHandler中会处理Request和Response所有逻辑,其中就涉及到url解析

PS:模型示例

  • 内部实现原理可用下面模型总结 基于IO多路复用的socket的服务端
# coding: utf-8
import socket
import select

class Handler(object):
    """ 业务功能代码 """

    def __init__(self, conn):
        self.conn = conn

    def execute(self):
        """
        处理客户端的请求
        :return:  False,断开连接;True,继续执行当前客户端发来的请求
        """
        data = self.conn.recv(1024)
        content = data.decode('utf-8')

        if content.upper() == "Q":
            return False

        # 1.处理登录
        # "login linda"
        # 2.注册
        # "register linda"

        self.conn.sendall(b'OK')
        return True


class Server(object):
    """ 基于普通socket的服务端 """

    def run(self, handler_class):
        # socket服务端
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind(("IP", "端口"))
        sock.listen(5)

        while True:
            # 等待,客户端发来连接
            conn, address = sock.accept()

            # 新客户端到来。 Handler对象
            instance = handler_class(conn)
            # 处理客户端的请求。 Handler对象.execute
            while True:
                result = instance.execute()
                if not result:
                    break
            conn.close()
        sock.close()


class SelectServer(object):
    """ 基于IO多路复用的socket的服务端 """

    def run(self, handler_class):
        server_object = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_object.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server_object.setblocking(False)
        server_object.bind(("IP", "端口"))
        server_object.listen(5)

        socket_object_list = [server_object, "客户端socket对象1", "客户端socket对象2"]

        conn_handler_map = {}
        # conn_handler_map = {
        #     "客户端socket对象1": Handler(conn1),
        #     "客户端socket对象2": Handler(conn2),
        # }

        while True:
            # r = ["客户端socket对象4", ]
            r, w, e = select.select(socket_object_list, [], [], 0.05)

            for sock in r:
                # sock="客户端socket对象4"

                # 新连接到来,执行 handler的 __init__ 方法
                if sock == server_object:
                    print("新客户端来连接")
                    conn, address = server_object.accept()
                    socket_object_list.append(conn)
                    # 实例化handler类,即:类(conn)
                    conn_handler_map[conn] = handler_class(conn)
                    continue

                # 一旦有请求发来,找到相关的 handler对象,执行他的 execute方法。
                #  execute方法返回False,则意味着此客户端要断开连接。
                handler_object = conn_handler_map[sock]   # 自己Handler(conn1),

                # 找到execute去处理各自的业务逻辑
                result = handler_object.execute()
                if not result:
                    socket_object_list.remove(sock)
                    del conn_handler_map[sock]
        sock.close()

if __name__ == '__main__':
    # server = Server()
    server = SelectServer()
    server.run(Handler)

从服务端代码角度,socket可以分为自己的服务端socket和客户端socket

  • 服务端socket有个额外功能,可以接收accept客户端socket
  • 服务端socket和客户端socket统一交给select.select进行选择判断,看看哪个socket准备好了
  • 准备好了的意思是说:
    • 判定服务端socket可以接收新的请求
    • 判定客户端socket可以收发数据
  • 每次socket的切换上下文的保存,也就是如何再定位到原先的socket,细节是由select.select实现的,我们只要实现socket收发业务逻辑就行

# 解析django.setup()

调用apps.populate(settings.INSTALLED_APPS)

  • 加载app模块
  • 加载app下的models模块
  • 调用加载完成的ready事件

# 调试工具 pdb

pdb - The Python Debugger

print --> Python 自带的Debug工具:pdb

  • 非侵入式方法(不用额外修改源代码,在命令行下直接运行就能调试)
python -m pdb filename.py
  • 侵入式方法(需要在被调试的代码中添加一行代码然后再正常运行代码
import pdb; pdb.set_trace()

当你在命令行看到下面这个提示符时,说明已经正确打开了pdb

(Pdb) h or help command

l(ist) [first [,last] | .]    # List source code for the current file.
ll     # List the whole source code for the current funtion or frame.

b/tbreak/cl     # 添加/临时/清除断点

p expression    # Print the value of the expression   

 # 逐行调试命令 - 执行下一行
 s    # 能够进入函数体
 n    # 不会进入函数体
 r    # 在函数中时会直接执行到函数返回处
 
 # 非逐行调试命令
 c    # 持续执行下去,直到遇到一个断点
 unt lineno  # 持续执行直到运行到指定行(或遇到断点)
 j lineno    # 直接跳转到指定行(注意,被跳过的代码不执行)
 
 
 a    # 在函数中时打印函数的参数和参数的值
 whatis expression    # 打印表达式的类型,常用来打印变量值
 interact             # 启动交互式解释器
 w     # 打印堆栈信息
 
 enter # 执行最近命令

PS:pip install ipdb (opens new window),基于ipython的pdb,带颜色

# 路由匹配 url

全局搜索配置文件中 ROOT_URLCONF,定位于 class BaseHandler中的 def _get_response

def _get_response(self, request):
    """
    Resolve and call the view, then apply view, exception, and
    template_response middleware. This method is everything that happens
    inside the request/response middleware.
    """
    response = None
    callback, callback_args, callback_kwargs = self.resolve_request(request)

    # Apply view middleware
    for middleware_method in self._view_middleware:
        response = middleware_method(request, callback, callback_args, callback_kwargs)
        if response:
            break

    if response is None:
        wrapped_callback = self.make_view_atomic(callback)
        # If it is an asynchronous view, run it in a subthread.
        if asyncio.iscoroutinefunction(wrapped_callback):
            wrapped_callback = async_to_sync(wrapped_callback)
        try:
            response = wrapped_callback(request, *callback_args, **callback_kwargs)
        except Exception as e:
            response = self.process_exception_by_middleware(e, request)
            if response is None:
                raise

    # Complain if the view returned None (a common error).
    self.check_response(response, callback)

    # If the response supports deferred rendering, apply template
    # response middleware and then render the response
    if hasattr(response, 'render') and callable(response.render):
        for middleware_method in self._template_response_middleware:
            response = middleware_method(request, response)
            # Complain if the template response middleware returned None (a common error).
            self.check_response(
                response,
                middleware_method,
                name='%s.process_template_response' % (
                    middleware_method.__self__.__class__.__name__,
                )
            )
        try:
            response = response.render()
        except Exception as e:
            response = self.process_exception_by_middleware(e, request)
            if response is None:
                raise

    return response
  • callback, callback_args, callback_kwargs = self.resolve_request(request)

# 数据模型 models

django/db

- django/db
    |- backends  # 支持多种数据库驱动
    |- models    # 模型相关,如manager.py、query.py等
    |- init.py   # 对外导入接口    
    |- transaction.py   # 事务相关操作
    |- utils.py  # 连接connection相关操作
django

# 基本使用

# 基础理论

  • ORM 实现数据模型与数据库的解耦,即数据模型的设计不依赖于特定的数据库
  • ORM,Object Relation Mapping, "对象-关系-映射"
# 创建表
create table employee(
    id int primary key auto_increment,
    name varchar(32) not null,
    gender bool default 1 not null,
    birthday date,
    department varcher(32) not null,
    salary decimal(8, 2)
);


insert employee (name, gender, birthday, department, salary) values ('linda', 0, '2000-01-01', 'Sale', 8000.00);

select * from employee where name = 'linda';
update employ set salary = 10000.00 where name = 'lidna';
delete from employ where name = 'linda';

# python类
class Employee(models.Model):
    id = models.AutoField(primary_key=True)
    name = models.CharField(max_length=32)
    gener = models.BooleanField(default=True)
    birthday = models.DateField(null=True, blank=True)
    department = models.CharField(max_length=32)
    salary = models.DecimalField(max_digits=8, decimal_places=2, default=0)
    
emp = Employee(name='linda', gender=False, birthday='Sale', salary=8000.00)
Employee.objects.filter(name='linda')
Employee.objects.filter(name='linda').update(salary=10000.00)
Employ.objects.filter(name='linda').delete()
  • 名称,见名识意
  • 类型,数值型 字符型 日期型
  • 约束,主键 外键 默认值 唯一性 非空
  • 设计的字段是用来确定唯一性的标识字段,还是唯一性基础上的属性说明字段
  • 关联关系的 往上字典 往下列表

A = models.ForeignKey(B, related_name='a_foreigh_key_b')

  • 正向查询: A --------> B
  • 反向查询: B --------> A

# 多表分组查询

分组查询SQL: 查询每一个出版社的名称以及出版的书籍个数
    select publish.name, count('title')
    from book inner join publish
    on book.publish_id = publish.id
    group by publish.id, publish.name, publish.addr, publish.email  # 主键publish.id唯一非空,效果相同

Models查询:查询每一个出版社的名称以及出版的书籍个数
                        # select id from publish,即按照select 语句进行分组
                        # <QuerySet [{'id': 1, 'book_count': 100}, {'id': 2, 'book_count': 200}]> 有几个字典代表对应出版社个数
                        # values('其实可展示 Publish 对象字段 和 聚合值 book_count')
                        # Count('book__title') 反向查询表名小写__字段名
    Publish.objects.values('id').annotate(book_count=Count('book__title')).values('name', 'email', 'book_count')
                
                        # <QuerySet [<Publish: 人民出版社>, <Publish: 南京出版社>]>
    Publish.objects.all()                       
                        # <QuerySet [<Publish: 人民出版社>, <Publish: 南京出版社>]> 显示对象
    Publish.objects.all().annotate(book_count=Count('book__title'))
    
                        # select id, name, publish, addr, email 与 id 结果相同
    Publish.objects.all().annotate(book_count=Count('book__title')).values('name', 'email', 'book_count')
    Publish.objects.annotate(book_count=Count('book__title')).values('name', 'email', 'book_count')
                        
                        # .filter(book_count__gt=1) having book_count > 1 统计完了之后再过滤 having
    Publish.objects.annotate(book_count=Count('book__title')).values('name', 'email', 'book_count').filter(book_count__gt=1)
    
总结 跨表的分组查询的模型:
- 每一个后的表模型.objects.values('pk').annotate(聚合函数(关联表__统计字段)).values('表模型的所有字段以及统计字段')
- 每一个后的表模型.objects.annotate(聚合函数(关联表__统计字段)).values('表模型的所有字段以及统计字段')

# aggregate 和 annotate

annotate()可以得到一些我们手动计算得到的值,并将其作为Queryset中Item的一个属性来调用

aggregate()可以使得查询的返回值由一个Queryset变成一个dict,每个key和对应的value由自己计算得到。

Case/When

Staff.objects.annotate(
    age_tag = Case(
        When(
            age__lt=18,
            then=1),
        When(
            age__lt=30,
            then=2,
        )
        When(
            age__lt=60,
            then=3,
        )
        default=0,
        output_field=IntegerField(),
    )
)

# F查询 与 Q查询

model进行查询时通常是 某个字段和一个常量 对比,涉及到变量时,可用F查询

from django.db.models import F, Q

# 评论数大于阅读数 
SQL: select * from book where comment_num > read_num
Model: Book.objects.filter(comment__num__gt=F('read_num'))

# 每一本书价格加10
Book.objects.all().update(price=F('price') + 10)

# 且关系 等同于 Q('红楼梦') & Q(price=100)
Book.objects.filter(title='红楼梦', price=100)
# 或关系
Book.objects.filter(Q(title='红楼梦') | Q(price=100))
# 非关系
Book.objects.filter(~Q(title='红楼梦')
# 注意顺序 先 Q,后字段
Book.objects.filter(Q(title='红楼梦') | Q(price=100), comment_num__gte=100)

组合技

import operator
from django.db.models import Q

q1 = reduce(operator.and_, Q(gender='F'))
q2 = reduce(operator.or_, Q(gender='M'))

# 最佳实践

  • 当需要同时写多张表的时候,为了保证数据的一致性,需要使用事务
# 定义事务时间点
sid = transaction.savepoint(using=settings.WRITE_DATABASE)
# 成功提交
transaction.savepoint_commit(sid, using=settings.WRITE_DATABASE)
# 失败回滚
transaction.savepoint_rollback(sid, using=settings.WRITE_DATABASE)

# with 上下文形式
with transaction.atomic(using=settings.WRITE_DATABASE):
    pass
  • default=None -> isnull=True
# 定义MySQL数据字段类型
ALTER TABLE `teacher` ADD COLUMN first_class_time DATETIME DEFAULT NULL;

# 定义Model字段
first_class_time = models.DateTimeField(help_text='老师的第一节课', default=None)

# 使用Model字段查询
 models.Teacher.objects.filter(first_class_time__isnull=False).count()
  • 排序 主键索引 order_by('-id')

  • collections.Counter()

course_ids = list(orders.values_list('course_id', flat=True))
times_of_course_info = collections.Counter(course_ids)
times_of_course_info = dict(times_of_course_info.most_common(10))

# backends

- backends
    |- base
    |- dummy
    |- mysql
        |- base.py # 统一定义 class DatabaseWrapper
    |- oracle
    |- postgresql
    |- sqlite3
  • 规定好目录接口,对外暴露统一的接口,实现支持多种数据库,而使用方感知不到
  • backends/base中的 BaseDatabaseWrapper 定义connect函数,而具体的mysql/base中的 def get_new_connection(self, conn_params): 进行实际的MySQLdb.connect(**conn_params)

# models

query_set = Book.objects.all()
print(query_set.count())
print(len(query_set))

Book.objects.all().select_related('category')
  • 查询也是可以放到Model上来处理,但是这样会导致Model操作功能较多,所以objects为Manager实例,单独来处理数据库操作
  • Manager是QeurySet完整代理

# transaction.py

  • 当需要同时写多张表的时候,为了保证数据的一致性,需要使用事务
# 定义事务时间点
sid = transaction.savepoint(using=settings.WRITE_DATABASE)
# 成功提交
transaction.savepoint_commit(sid, using=settings.WRITE_DATABASE)
# 失败回滚
transaction.savepoint_rollback(sid, using=settings.WRITE_DATABASE)

# with 上下文形式
with transaction.atomic(using=settings.WRITE_DATABASE):
    pass

# utils.py

def load_backend(backend_name):
    # setttings中配置策略匹配到 backends中 DatabaseWrapper
    try:
        return import_module('%s.base' % backend_name)
    except ImportError as e_user:
        raise

# 自定义Handler类,让使用起来类似一个字典
class ConnectionHandler:
    def __init__(self, databases=None):
        self._databases = databases
        self._connections = Local(thread_critical=True)

    @cached_property
    def databases(self):
        if self._databases is None:
            self._databases = settings.DATABASES
        if self._databases == {}:
            self._databases = {
                DEFAULT_DB_ALIAS: {
                    'ENGINE': 'django.db.backends.dummy',
                },
            }
        if DEFAULT_DB_ALIAS not in self._databases:
            raise ImproperlyConfigured("You must define a '%s' database." % DEFAULT_DB_ALIAS)
        if self._databases[DEFAULT_DB_ALIAS] == {}:
            self._databases[DEFAULT_DB_ALIAS]['ENGINE'] = 'django.db.backends.dummy'
        return self._databases

    def __getitem__(self, alias):
        if hasattr(self._connections, alias):
            return getattr(self._connections, alias)

        self.ensure_defaults(alias)
        self.prepare_test_settings(alias)
        db = self.databases[alias]
        backend = load_backend(db['ENGINE'])
        conn = backend.DatabaseWrapper(db, alias)
        setattr(self._connections, alias, conn)
        return conn

# 信号系统 signal

高内聚 低耦合

  • Django 内置信号使用
from django.db import models
from django.db.models.signals import post_save
from django.dispatch import receiver, Signal

class Category(models.Model):
    name = models.CharField(max_length=128, verbose_name='类名')
    updated_time = models.DateTimeField(auto_now=True)
    created_time = models.DateTimeField(auto_now_add=True)

class Book(models.Model):
    title = models.CharField(max_length=128, verbose_name='书名')
    category = models.ForeignKey(Category, on_delete=models.DO_NOTHING)
    updated_time = models.DateTimeField(auto_now=True)
    created_time = models.DateTimeField(auto_now_add=True)

# django下的已定义好的post_save
@receiver(post_save, sender=Book)
def notify_thirty(sender, **kwargs):
    print(sender, kwargs)
  • 自定义信号使用
from django.dispatch import receiver, Signal

# 定义signal方式
pizza_done = Signal(providing_args=['size'])

# 触发操作点
def handler(self, *arg, **kw):
    sender = self.__class__
    pizza_done.send(sender=sender, size=10)

# 触发回调函数
@receiver(pizza_done)
def launch(sender, **kw):
    print(sender, kw)

# or 触发回调函数
# pizza_done.connect(launch)
  • 源码所在模块 django/dispatch/dispatcher.py不是很复杂,可以精读一下,观察者模式

# 多级缓存 caches

源码位置

  • django/core/cache 根据settings.CACHES封装对外的caches、cache
  • django/middleware/cache.py 实现页面级别缓存
MIDDLEWARE = [
    'django.middleware.cache.UpdateCacheMiddleware',
    ...
    'django.middleware.cache.FetchFromCacheMiddleware'
]

# 认证权限 auth

你是谁,能干啥

源码位置

MIDDLEWARE = [
    'django.contrib.sessions.middleware.SessionMiddleware',
    ...
    'django.contrib.auth.middleware.AuthenticationMiddleware',
]
上次更新: 8/5/2021, 4:09:57 PM