web框架学习记录

[TOC]

---

django

个人包装练习的项目

runoob - jango 教程

django官方文档

1️⃣ 框架概述

• Django 是一个用 Python 编写的开源 Web 框架。
• 它遵循 MVC/MVT 架构模式（Model-View-Template）。
• 目标是快速开发、安全可靠、可扩展的 Web 应用。
• 官方网站：https://www.djangoproject.com/

2️⃣ 主要特点

1. 快速开发：内置管理后台（Admin）、ORM 映射数据库，免写大量 SQL、内置表单、认证、会话、缓存支持
2. 安全性高：自动防止 SQL 注入、防止跨站脚本 (XSS)、防止跨站请求伪造 (CSRF)、提供用户认证和权限管理
3. 可扩展性强：支持插件和第三方库、模块化设计，项目可拆分为多个 App、可与 REST API、GraphQL 等接口集成
4. 丰富的生态：有 Django REST Framework（DRF）支持 API 开发、Django Channels 支持 WebSocket、支持 Celery 异步任务队列

3️⃣ 架构模式：MVT（Model-View-Template）

组件	功能
Model	数据模型，负责数据库操作（ORM 映射表）
View	业务逻辑处理，接收请求，返回响应
Template	前端模板，负责展示 HTML 页面

类似 MVC：

• Controller 的角色由 View + URL Dispatcher 承担
• Template 对应 View 层的展示部分

4️⃣ 项目结构

一个典型 Django 项目结构：

myproject/
├── manage.py             # 命令行工具
├── myproject/            # 项目配置
│   ├── settings.py       # 配置文件
│   ├── urls.py           # URL 路由
│   ├── wsgi.py           # 部署用
│   └── asgi.py           # 异步部署用
├── apps/                 # 自定义应用
│   └── blog/
│       ├── models.py     # 数据模型
│       ├── views.py      # 视图/逻辑
│       ├── urls.py       # 应用路由
│       └── admin.py      # 后台管理
└── templates/            # 模板文件

---

5️⃣ 核心组件

1. ORM（Object Relational Mapping）：Python 类映射数据库表、支持关系、外键、联合唯一约束、多对多关系
2. Admin 管理后台：自动生成增删改查后台界面、可通过 admin.py 配置显示字段、搜索、过滤
3. URL 路由：将请求 URL 映射到 View 函数或类、支持动态参数和命名空间
4. 中间件（Middleware）：请求和响应处理链、可做认证、日志、压缩、跨域等处理
5. 模板系统（Template）：渲染 HTML 页面、支持模板继承和自定义标签

6️⃣ 应用场景

• Web 应用系统（博客、电商、CMS）
• RESTful API 开发（结合 DRF）
• 后台管理系统
• 数据可视化系统
• 社交平台和微服务

✅ 总结：

• Django 是 Python 最流行的 Web 框架之一
• 提供全栈开发能力（前端模板 + 后端逻辑 + 数据库）
• 特点：快速、安全、可扩展、模块化
• 核心理念：“Don’t repeat yourself（DRY）”

新建项目

1. 新建项目

   # 新建项目
   django-admin startproject myproject
   
   myproject/
   │
   ├── manage.py
   └── myproject/
       ├── __init__.py
       ├── settings.py
       ├── urls.py
       ├── asgi.py
       └── wsgi.py
   
   # 启动开发服务器
   python manage.py runserver
       
   # 访问浏览器
   http://127.0.0.1:8000/

   生产环境下可以用uvicorn启动

   #!/usr/bin/env Python
   # -- coding: utf-8 --
   
   """
   @version: v1.0
   @author: huangyc
   @file: uvicorn_start.py
   @Description: asgi异步服务器
   @time: 2025/10/4 5:49
   """
   import os
   
   import uvicorn
   from dotenv import load_dotenv
   
   from configs.project_config import apps
   
   # 加载.env文件
   load_dotenv()
   
   if __name__ == "__main__":
       host = os.getenv("DJANGO_HOST", "0.0.0.0")
       port = int(os.getenv("DJANGO_PORT", 80))
       workers = int(os.getenv("DJANGO_WORKERS", 1))
   
       uvicorn.run("django_payin.asgi:application",  # 注意这里是 module:application
                   host=host, port=port, workers=workers)

2. 新建app

   # 新建文件夹apps（用于分开存放app）
   python manage.py startapp myapp apps/myapp
   
   myapp/
   ├── admin.py
   ├── apps.py
   ├── models.py
   ├── views.py
   ├── urls.py   # 需要手动创建
   └── ...
   
   apps

   在项目settings下配置

   INSTALLED_APPS = [
       'django.contrib.admin',
       'django.contrib.auth',
       ...
       'myapp',  # ← 新增
   ]

   配置url

   # 项目urls
   urlpatterns = [
       path('admin/', admin.site.urls),
       path('api/', include('myapp.urls')),  # ← 添加你的 app 路由
   ]
   
   
   # 应用urls
   urlpatterns = [
       path('hello/', views.hello, name='hello'),
   ]

django-admin

常用 django-admin 命令详解

1. 创建新项目

   django-admin startproject <项目名称> [目标目录]

   | 参数 | 作用 | 示例 |
   | :—————— | :——————————————————————- | :—————————————————————————————- |
   | <项目名称> | 必填，项目名称（会生成同名目录） | django-admin startproject mysite |
   | [目标目录] | 可选，指定项目存放目录 | django-admin startproject mysite /opt/myproject |
   | --template | 使用自定义项目模板 | django-admin startproject --template=my_template.zip mysite |
   | --extension | 指定文件扩展名（如 .py, .txt） | django-admin startproject --extension=py,txt mysite |
   | --name | 指定文件名模式（如 Dockerfile, README.md） | django-admin startproject --name=Dockerfile mysite |

   django-admin startproject mysite  # 创建默认项目
   django-admin startproject mysite /opt/code  # 指定目录
   django-admin startproject --template=https://example.com/my_template.zip mysite  # 使用远程模板

2. 创建新应用，虽然通常使用 manage.py 来创建应用，但也可以通过 django-admin：

   django-admin startapp <应用名称> [目标目录]

   这会创建一个新的 Django 应用，包含：

   • migrations/：数据库迁移文件目录
   • __init__.py
   • admin.py：管理后台配置
   • apps.py：应用配置
   • models.py：数据模型定义
   • tests.py：测试代码
   • views.py：视图函数

   | 参数 | 作用 | 示例 |
   | :—————- | :————————————————————————— | :————————————————————————————— |
   | <应用名称> | 必填，应用名称（会生成 models.py, views.py 等） | django-admin startapp blog |
   | [目标目录] | 可选，指定应用存放目录 | django-admin startapp blog /opt/myapp |
   | --template | 使用自定义应用模板 | django-admin startapp --template=my_app_template.zip blog |

   django-admin startapp blog  # 创建默认应用
   django-admin startapp blog /opt/myapp  # 指定目录
   django-admin startapp --template=my_template.zip blog  # 使用模板

3. 检查项目配置

   django-admin check

4. 数据库迁移，Django 使用迁移系统来管理数据库模式变更：

   django-admin makemigrations  # 创建迁移文件
   django-admin migrate        # 应用迁移到数据库

5. 创建超级用户，会引导你创建一个可以访问 Django 管理后台的超级用户

   django-admin createsuperuser

6. django-admin 常用命令

   | 命令 | 作用 | 示例 |
   | :———————— | :—————————————- | :—————————————————— |
   | startproject | 创建一个新 Django 项目 | django-admin startproject myproject |
   | startapp | 创建一个新 Django 应用 | django-admin startapp myapp |
   | runserver | 启动开发服务器 | python manage.py runserver |
   | makemigrations | 生成数据库迁移文件 | python manage.py makemigrations |
   | migrate | 执行数据库迁移 | python manage.py migrate |
   | createsuperuser | 创建管理员账号 | python manage.py createsuperuser |
   | shell | 启动 Django 交互式 Shell | python manage.py shell |
   | collectstatic | 收集静态文件（用于生产环境） | python manage.py collectstatic |
   | test | 运行单元测试 | python manage.py test |

7. django-admin runserver（启动开发服务器）

   python manage.py runserver [IP:端口]

   | 参数 | 作用 | 示例 |
   | :—————- | :————————————————————————— | :———————————————————— |
   | [IP:端口] | 可选，指定监听的 IP 和端口（默认 127.0.0.1:8000） | python manage.py runserver 0.0.0.0:8080 |
   | --noreload | 禁用自动重载（调试时使用） | python manage.py runserver --noreload |
   | --insecure | 强制静态文件服务（非 DEBUG 模式） | python manage.py runserver --insecure |

   python manage.py runserver  # 默认启动（127.0.0.1:8000）
   python manage.py runserver 0.0.0.0:8000 # 允许外部访问
   python manage.py runserver 8080 # 仅修改端口

8. django-admin migrate（数据库迁移）

   python manage.py migrate [应用名] [迁移版本]

   | 参数 | 作用 | 示例 |
   | :———————- | :—————————————————- | :———————————————————— |
   | [应用名] | 可选，指定要迁移的应用 | python manage.py migrate blog |
   | [迁移版本] | 可选，指定迁移版本号 | python manage.py migrate blog 0002 |
   | --fake | 标记迁移为已执行（不实际修改数据库） | python manage.py migrate --fake |
   | --fake-initial | 仅当表已存在时标记为已执行 | python manage.py migrate --fake-initial |

   python manage.py migrate  # 执行所有未应用的迁移
   python manage.py migrate blog  # 仅迁移 blog 应用
   python manage.py migrate blog 0002  # 迁移到特定版本

9. 其他常用命令

   | 命令 | 作用 | 示例 |
   | :———————— | :———————————————- | :————————————————- |
   | createsuperuser | 创建管理员用户 | python manage.py createsuperuser |
   | shell | 进入 Django Shell（带 ORM 支持） | python manage.py shell |
   | test | 运行测试用例 | python manage.py test blog |
   | collectstatic | 收集静态文件（生产环境部署） | python manage.py collectstatic |

库初始化

Django 使用 migrate 命令来初始化数据库。这个命令会应用所有已定义的迁移文件，创建数据库表。

配置数据库

在 settings.py 文件中，配置数据库连接。默认情况下，Django 使用 SQLite 数据库：

# settings.py

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': BASE_DIR / 'db.sqlite3',
    }
}

如果你使用其他数据库（如 PostgreSQL、MySQL 等），需要安装相应的数据库驱动并配置连接参数。例如，使用 PostgreSQL：

pip install psycopg2

然后在 settings.py 中配置：

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'mydatabase',
        'USER': 'mydatabaseuser',
        'PASSWORD': 'mypassword',
        'HOST': '127.0.0.1',
        'PORT': '5432',
    }
}

多数据配置（进阶）

运行迁移

1. 生成迁移文件（根据你定义的模型生成数据库表结构）：

python manage.py makemigrations

运行以下命令来初始化数据库：

python manage.py migrate

这个命令会创建所有必要的数据库表，包括 Django 自带的用户认证系统表。

扫描app识别列表

python manage.py showmigrations

清空数据

使用 Django 管理命令

Django 提供了一个 flush 命令，可以清空所有应用的数据。但这个命令会清空整个数据库，而不是特定的应用。如果你只想清空特定应用的数据，可以结合 migrate 命令来实现。

1. 回滚指定应用的迁移：首先，回滚指定应用的迁移，这将删除该应用的所有表。

   python manage.py migrate myapp zero

   这个命令会将 myapp 的迁移回滚到初始状态，删除该应用的所有表。

2. 重新应用迁移：然后，重新应用迁移，创建空表。

   python manage.py migrate myapp

   这个命令会重新应用 myapp 的迁移，创建空表

创建管理员用户

使用 createsuperuser 命令创建一个管理员用户：

python manage.py createsuperuser

运行这个命令后，系统会提示你输入用户名、邮箱和密码。例如：

Username (leave blank to use 'yourusername'): admin
Email address: admin@example.com
Password: 
Password (again): 
Superuser created successfully.

启动开发服务器

启动 Django 开发服务器，访问项目：

python manage.py runserver

打开浏览器，访问 http://127.0.0.1:8000/admin，使用刚才创建的管理员用户登录。

模型定义

主键（Primary Key）

每个 Django 模型默认有一个 id 主键字段：

id = models.AutoField(primary_key=True)

可以自定义主键：

book_id = models.CharField(max_length=20, primary_key=True)

主键保证表中每条记录唯一，用于 ORM 查询、关联外键等。

复合主键（Multi-column PK）

Django 不支持原生多列主键。

方法 1（推荐）：

• 给表增加单列主键（id/UUID）。

• 原来的多列字段加唯一约束：

  class A(models.Model):
      id = models.AutoField(primary_key=True)
      field1 = models.CharField(max_length=20)
      field2 = models.CharField(max_length=20)
      field3 = models.CharField(max_length=20)
  
      class Meta:
          unique_together = ('field1', 'field2', 'field3')

• 外键直接关联主键 id。

• 优点：查询、级联、admin 都方便。

方法 2（不推荐）：

• B 表存三列字段对应 A 表的三列。

• 手动验证组合是否存在：

  if not A.objects.filter(field1=f1, field2=f2, field3=f3).exists():
      raise ValidationError

• 缺点：无法自动级联，admin 和 ORM 查询不方便。

外键（ForeignKey）

1. 一对一：表示 一条记录对应另一条记录。

   class UserProfile(models.Model):
       user = models.OneToOneField(User, on_delete=models.CASCADE)
       birthday = models.DateField()

   常用于扩展 User 表

   不想用主键来关联的情况

   在 Django 中，OneToOneField 通常用于建立一对一的关系，它默认关联到目标模型的主键。如果你需要通过非主键字段建立一对一关系，可以使用 ForeignKey 并设置 unique=True。这样可以实现类似一对一的关系，但实际上是通过非主键字段进行关联。

   from django.db import models
   from django.contrib.auth.models import User
   
   class UserProfile(models.Model):
       user = models.ForeignKey(User, on_delete=models.CASCADE, unique=True)
       birthday = models.DateField()
   
       def __str__(self):
           return f"{self.user.username}'s profile"

   在这个例子中：

   • user 字段是一个 ForeignKey，关联到 User 模型。
   • unique=True 确保每个 User 只能关联一个 UserProfile，从而实现一对一的关系。

   为什么不能直接使用 OneToOneField 指定非主键字段？

   OneToOneField 是 ForeignKey 的一个特例，它默认关联到目标模型的主键。Django 的 OneToOneField 不支持直接指定非主键字段作为关联字段。如果你需要通过非主键字段建立一对一关系，必须使用 ForeignKey 并设置 unique=True。

   示例：通过非主键字段建立一对一关系

   假设你有一个 User 模型和一个 UserProfile 模型，你希望通过 User 的 email 字段建立一对一关系。你可以这样实现：

   from django.db import models
   from django.contrib.auth.models import User
   
   class UserProfile(models.Model):
       user = models.ForeignKey(User, on_delete=models.CASCADE, unique=True, to_field='email')
       birthday = models.DateField()
   
       def __str__(self):
           return f"{self.user.username}'s profile"

   在这个例子中：

   • to_field='email' 指定了 User 模型的 email 字段作为关联字段。
   • unique=True 确保每个 User 的 email 只能关联一个 UserProfile

2. 一对多关系：默认关联目标模型的 主键字段（id）。

   class Author(models.Model):
       name = models.CharField(max_length=50)
   
   class Book(models.Model):
       title = models.CharField(max_length=100)
       author = models.ForeignKey(Author, on_delete=models.CASCADE)

   数据库中生成 author_id 列。

   on_delete 常用选项：

   • CASCADE：父表删除，子表删除
   • SET_NULL：父表删除，子表字段置 NULL
   • PROTECT：阻止删除父表
   • SET_DEFAULT：设置默认值

   • DO_NOTHING：不做操作

   • 可用 related_name 设置反向查询：

     author = models.ForeignKey(Author, related_name="books", on_delete=models.CASCADE)

3. 多对多（ManyToManyField）：表示多条记录与多条记录的关系。

   class Book(models.Model):
       title = models.CharField(max_length=100)
   
   class Reader(models.Model):
       name = models.CharField(max_length=50)
       books = models.ManyToManyField(Book, related_name="readers")

   Django 会自动创建中间关联表，可通过 through 指定自定义关联表。

Django 外键实际关联原理

• 外键字段对应 目标模型的主键（默认 id）。

• 数据库中存储 字段名_id。

• 查询时 Django 会自动返回关联对象：

  book = Book.objects.get(id=1)
  print(book.author.name)

• 反向查询：

  author = Author.objects.get(id=1)
  author.book_set.all()  # 默认 related_name=modelname_set

关键点总结表

关系类型	Django 字段类型	对应数据库结构	示例用途
一对多	ForeignKey	表 A.id ↔ 表 B.a_id	订单 → 用户
一对一	OneToOneField	表 A.id ↔ 表 B.a_id	用户 → 用户详情
多对多	ManyToManyField	中间关联表	读者 ↔ 书
主键唯一标识	AutoField/UUIDField	表的唯一列	每条记录唯一
复合主键（推荐方式）	AutoField + unique_together	单列主键 + 多列唯一约束	组合字段唯一，方便外键关联

Django Admin

Django 自带的 Admin 管理界面非常强大，可以方便地管理数据库中的数据。以下是一些基本配置：

注册模型

在你的应用的 admin.py 文件中，注册你的模型以便在 Admin 界面中管理。例如：

# myapp/admin.py
from django.contrib import admin
from .models import MyModel

admin.site.register(MyModel)

自定义 Admin 界面

你可以自定义 Admin 界面的显示方式。例如：

# myapp/admin.py
from django.contrib import admin
from .models import MyModel

@admin.register(MyModel)
class MyModelAdmin(admin.ModelAdmin):
    list_display = ('id', 'name', 'created_at')
    search_fields = ('name',)
    list_filter = ('created_at',)

迁移模型

在 Django 中修改模型后，需要通过迁移（migrations）来更新数据库结构。以下是详细的步骤：

修改模型

首先，修改你的模型。例如，假设你有一个 UserProfile 模型，你想要添加一个新的字段 phone_number：

# models.py
from django.db import models
from django.contrib.auth.models import User

class UserProfile(models.Model):
    user = models.OneToOneField(User, on_delete=models.CASCADE)
    birthday = models.DateField()
    phone_number = models.CharField(max_length=15, blank=True, null=True)  # 新添加的字段

生成迁移文件

运行以下命令生成迁移文件：

python manage.py makemigrations

这个命令会检查你的模型定义，并生成相应的迁移文件。迁移文件会保存在应用的 migrations 文件夹中。例如：

myapp/
    migrations/
        0001_initial.py
        0002_add_phone_number.py  # 新生成的迁移文件

查看迁移文件

打开生成的迁移文件，确认生成的内容是否符合你的预期。例如：

from django.db import migrations, models

class Migration(migrations.Migration):
    dependencies = [
        ('myapp', '0001_initial'),
    ]

    operations = [
        migrations.AddField(
            model_name='userprofile',
            name='phone_number',
            field=models.CharField(max_length=15, blank=True, null=True),
        ),
    ]

应用迁移

运行以下命令应用迁移，更新数据库结构：

python manage.py migrate

这个命令会应用所有未应用的迁移文件，更新数据库结构。例如：

Operations to perform:
  Apply all migrations: myapp
Running migrations:
  Applying myapp.0002_add_phone_number... OK

验证迁移

访问 Django Admin 或使用 Django Shell 验证迁移是否成功。例如：

python manage.py shell

在 Shell 中，你可以检查新字段是否已经添加：

from myapp.models import UserProfile
UserProfile.objects.create(user=User.objects.get(id=1), birthday='1990-01-01', phone_number='1234567890')

处理数据迁移

如果你需要在迁移过程中处理数据，可以使用 RunPython 操作。例如，假设你需要在添加新字段时初始化一些数据：

# myapp/migrations/0002_add_phone_number.py

from django.db import migrations, models

def add_default_phone_number(apps, schema_editor):
    UserProfile = apps.get_model('myapp', 'UserProfile')
    for profile in UserProfile.objects.all():
        profile.phone_number = '0000000000'
        profile.save()

class Migration(migrations.Migration):
    dependencies = [
        ('myapp', '0001_initial'),
    ]

    operations = [
        migrations.AddField(
            model_name='userprofile',
            name='phone_number',
            field=models.CharField(max_length=15, blank=True, null=True),
        ),
        migrations.RunPython(add_default_phone_number),
    ]

回滚迁移

如果你需要回滚迁移，可以使用以下命令：

python manage.py migrate myapp 0001

这个命令会将 myapp 的迁移回滚到 0001_initial 状态。

信号

django-cleanup项目学习

django官方文档-signal

Django 的 信号机制（Signals） 是一个非常实用、优雅的“事件通知系统”，让不同模块之间可以在不直接耦合的情况下进行通信。

什么是信号（Signal）

定义：信号是 Django 提供的一种机制，当某个事件发生时，系统自动“发射（send）信号”，而其他模块可以提前“监听（connect）”这个信号，从而执行某些函数。

你可以把它理解成一种 “观察者模式”（Observer Pattern）：事件源发射信号 → 所有关心这个事件的监听者被通知并执行回调函数。

信号的主要组成部分

组件	作用
Signal 对象	表示一种可触发的事件，比如 django.db.models.signals.post_save
发送信号（send）	某个动作发生后调用 .send()，通知监听者
接收信号（receiver）	一个普通函数，用 @receiver 装饰后自动执行
连接信号（connect）	手动将接收函数注册到信号上

Django 内置常用信号

信号	触发时机
pre_save / post_save	模型对象保存前 / 保存后
pre_delete / post_delete	模型对象删除前 / 删除后
m2m_changed	多对多关系变化时
request_started / request_finished	请求开始 / 请求结束时
user_logged_in / user_logged_out / user_login_failed	用户登录事件

---

基本使用示例

假设我们要在用户注册（User 对象保存）后自动发送欢迎邮件。

1. 创建信号接收器：在 users/signals.py 文件：

   from django.db.models.signals import post_save
   from django.dispatch import receiver
   from django.contrib.auth.models import User
   
   @receiver(post_save, sender=User)
   def send_welcome_email(sender, instance, created, **kwargs):
       if created:  # 只在新建用户时执行
           print(f"欢迎新用户：{instance.username}")
           # send_email(instance.email, '欢迎加入我们！')

2. 注册信号模块：在 users/apps.py 中（推荐做法），Django 启动时会执行 ready()，自动注册信号

      from django.apps import AppConfig
      
      class UsersConfig(AppConfig):
          default_auto_field = 'django.db.models.BigAutoField'
          name = 'users'
      
          def ready(self):
              import users.signals  # 导入信号处理模块
   
   3. 自定义信号，如果内置信号不够用，你也可以定义自己的信号
   
      ```cmd
      from django.dispatch import Signal, receiver
      
      # 定义一个自定义信号
      order_paid = Signal()
      
      # 定义监听器
      @receiver(order_paid)
      def handle_order_paid(sender, order_id, **kwargs):
          print(f"订单 {order_id} 已付款，通知发货系统。")
      
      # 发射信号
      order_paid.send(sender=None, order_id=123)

信号执行的时机，信号执行是 同步的（synchronous）：

• 也就是说：send() 会立即执行所有接收器函数，等它们完成后才返回。
• 如果接收器出错，会导致信号触发处抛异常。

✅ 若你希望信号异步执行（例如发邮件），可以结合：

• Celery 异步任务；
• 或者 Django 4.1+ 的 async def 接收器。

常见使用场景

场景	示例
用户注册后自动初始化资料	@receiver(post_save, sender=User)
删除对象时清理文件 / 缓存	@receiver(post_delete, sender=MyModel)
日志或审计记录	监听数据库变动信号
电商中订单状态更新通知	自定义信号触发回调
模块解耦	不直接导入另一个模块，只通过信号交互

使用注意事项

1. 不要滥用信号：信号逻辑过多会使系统难以维护（“魔法太多”）。
2. 确保导入执行：如果没有在 apps.py 里导入信号模块，信号不会生效。
3. 信号是同步的，耗时操作（如发邮件）最好用异步任务队列。
4. 调试时可用 django.db.models.signals 查看有哪些信号被注册。

总结类比

框架	类似机制
Django	Signals（观察者模式）
FastAPI	无内置信号系统，一般用事件钩子或中间件实现
Flask	有 Blinker 信号库，可实现类似机制

django test

常见测试类

1. django.test.TestCase
   • 最常用，继承自 unittest.TestCase，会自动使用测试数据库（运行前建库，结束后清理）。
   • 自带断言方法（如 self.assertContains、self.assertTemplateUsed）。
2. django.test.SimpleTestCase
   • 不需要数据库时使用（更快）。
   • 适合测试一些工具函数、纯逻辑。
3. django.test.TransactionTestCase
   • 会在数据库层执行事务测试，适合验证事务回滚逻辑。
   • 比 TestCase 慢。
4. django.test.LiveServerTestCase
   • 启动一个临时服务，结合 Selenium 等工具做端到端（E2E）测试。

安全相关

美团二面：Cookie、Session、Token、JWT究竟有什么区别？JWT的原理分析及避坑指南

jwt

token

1. 添加伪admin，诱导攻击

   urlpatterns = [
       # 真后台（已更名）
       path('__ready_ad__/', admin.site.urls),
   
       # 诱饵后台（可选，需安装 django-admin-honeypot）
       path('admin/', include('admin_honeypot.urls', namespace='admin_honeypot')),
   ]
   
   pip install django-honeypot-admin==2.0.0

2. nginx 访问速率限制

3. php攻击拦截

4. .env等拦截

5. 其他一些加密

6. 容器内获取真实访问IP（正在弄）

7. php模拟攻击测试（还未测试）

web通识

获取公网真实IP

1. 场景1：基于docker的nginx + 基于docker的django服务

   django获取真实IP的中间件

   import ipaddress
   import re
   
   from basic_support.logger.logger_config import logger
   
   
   class RealIPFromForwardedForMiddleware:
       def __init__(self, get_response):
           self.get_response = get_response
   
       @staticmethod
       def validate_ip(ip, fallback):
           try:  # 尝试解析，如果合法就返回标准格式字符串
               return str(ipaddress.ip_address(ip))
           except ValueError:  # 不合法就返回 fallback
               return fallback
   
       @staticmethod
       def get_real_ip(request):
           """
           根据多层 header 获取真实客户端 IP。
           优先级：
           1. Forwarded header
           2. X-Forwarded-For
           3. 其他常见代理 header
           4. REMOTE_ADDR
           """
           fallback = request.META.get('REMOTE_ADDR')
           try:
               ip = request.META.get('HTTP_FORWARDED')
               if ip and 'for=' in ip:
                   match = re.search(r'for=([\d\.]+)', ip)
                   if match:
                       return RealIPFromForwardedForMiddleware.validate_ip(match.group(1), fallback)
   
               # X-Forwarded-For 第一条才是客户端真实 IP
               ip = request.META.get('HTTP_X_FORWARDED_FOR')
               if ip and ip.lower() != 'unknown':
                   return RealIPFromForwardedForMiddleware.validate_ip(ip.split(',')[0].strip(), fallback)
   
               # 其他可能的代理头
               for header in [
                   'HTTP_PROXY_CLIENT_IP',
                   'HTTP_WL_PROXY_CLIENT_IP',
                   'HTTP_CLIENT_IP',
                   'HTTP_X_FORWARDED_FOR'
               ]:
                   ip = request.META.get(header)
                   if ip and ip.lower() != 'unknown':
                       return RealIPFromForwardedForMiddleware.validate_ip(ip, fallback)
   
               # 最后 fallback
               return fallback
           except Exception as e:
               logger.error("get_real_ip ERROR", exc_info=e)
               return fallback
   
       def __call__(self, request):
           # 打印请求信息
           # self.print_request_info(request)
   
           # 获取真实 IP 并覆盖 REMOTE_ADDR
           real_ip = self.get_real_ip(request)
           logger.trace(f"获取真实IP: {real_ip}")
           request.META['REMOTE_ADDR'] = real_ip
   
           return self.get_response(request)
   
       @staticmethod
       def print_request_info(request):
           """
           打印请求的完整信息，包括 REMOTE_ADDR、PATH、HOST、所有 HTTP_ 头
           """
           logger.trace("=== Request Info Start ===")
           logger.trace(f"REMOTE_ADDR: {request.META.get('REMOTE_ADDR')}")
           logger.trace(f"PATH_INFO: {request.path}")
           logger.trace(f"HOST: {request.META.get('HTTP_HOST')}")
           logger.trace(f"SERVER_PROTOCOL: {request.META.get('SERVER_PROTOCOL')}")
           logger.trace(f"REQUEST_METHOD: {request.META.get('REQUEST_METHOD')}")
   
           # 打印所有 HTTP_ 开头的头，并还原常规格式
           for key, value in sorted(request.META.items()):
               if key.startswith("HTTP_"):
                   header_name = key[5:].replace('_', '-').title()
                   logger.trace(f"{header_name}: {value}")
   
           # CONTENT_TYPE 和 CONTENT_LENGTH
           for key in ["CONTENT_TYPE", "CONTENT_LENGTH"]:
               if key in request.META:
                   logger.trace(f"{key}: {request.META[key]}")
   
           logger.trace("=== Request Info End ===\n")

   django基于中间件的IP控制

   import copy
   import socket
   import threading
   import time
   
   from basic_support.logger.logger_config import logger
   
   from configs.project_config import security
   
   
   def refresh_domain_ips(middleware_instance, interval=600):
       """自动根据 domain 刷新 IP 白名单"""
       while True:
           time.sleep(interval)
           new_ips = set()
           for domain in middleware_instance.allowed_domains:
               try:
                   ip_list = socket.gethostbyname_ex(domain)[2]
                   new_ips.update(ip_list)
                   # logger.debug(f"[IP 白名单刷新成功] {domain} -> {ip_list}")
               except socket.gaierror as e:
                   logger.warning(f"[DNS 解析失败] {domain}: {e}")
               except Exception as e:
                   logger.error(f"[IP 白名单刷新异常] {domain}: {e}")
           middleware_instance.allowed_ips = middleware_instance.cfg_allowed_ips | new_ips
   
   
   class IPWhitelistMiddleware:
       def __init__(self, get_response):
           self.get_response = get_response
           # 允许 IP 列表
           self.allowed_ips = set(security['allowed_ips'])
           self.cfg_allowed_ips = copy.deepcopy(self.allowed_ips)
   
           # 允许 Domain 列表
           self.allowed_domains = set(security['allowed_domains'])
           # 想要排除白名单检查的路径
           self.exempt_paths = set(security['exempt_paths'])
   
           logger.trace(
               f"配置ip白名单的数量为: {len(self.allowed_ips)}, domain白名单的数量为: {len(self.allowed_domains)}")
           threading.Thread(target=refresh_domain_ips, args=(self,), daemon=True).start()
   
       def get_remote_addr(self, request) -> str:
           # 直接用中间件覆盖后的 REMOTE_ADDR
           ip = request.META.get('REMOTE_ADDR')
           logger.trace(f"白名单中间件 REMOTE_ADDR: {ip}")
           return ip
   
       def __call__(self, request):
           # 排除指定接口
           if request.path in self.exempt_paths:
               return self.get_response(request)
   
           ip = self.get_remote_addr(request=request)
           logger.info(f"====================> [收到请求]，IP地址为: {ip}, path: {request.path}")
           # if ip not in self.allowed_ips:
           #     msg = f"Your IP【{ip}】 is not allowed."
           #     logger.warning(msg)
           #     from django.http import HttpResponseForbidden
           #     return HttpResponseForbidden(msg)
   
           return self.get_response(request)

   nginx中也要配置

   # 定义限速区（每个 IP 每秒最多 30 个请求）
   limit_req_zone $http_x_real_ip zone=req_limit_per_ip:10m rate=30r/s;
   
   # 定义 Django upstream 后端池
   upstream django_backend {
       # 后端 Django 服务地址
       server django_app:8001;
   
       # TCP 健康检查
       # interval=3000    每 3 秒检查一次
       # rise=2           连续 2 次成功标记为 UP
       # fall=5           连续 5 次失败标记为 DOWN
       # timeout=1000     TCP 超时时间 1 秒
       # type=tcp         TCP 健康检查
       check interval=3000 rise=2 fall=5 timeout=1000 type=tcp;
   }
   
   server {
       # HTTP 监听端口
       listen 80;
       server_name api.ptxpay.com;
   
       # 限制客户端请求体最大 5MB
       client_max_body_size 5M;
   
       # === 恶意请求防护 start ===
       location ~* (allow_url_include|auto_prepend_file|eval-stdin\.php|think\\\\app\\\\invokefunction|phpunit|php://input|base64_decode) {
           access_log off;
           return 403;
       }
       # === 恶意请求防护 end ===
   
       # 拒绝访问隐藏文件 (.env, .git 等)
       location ~ /\. {
           deny all;
       }
   
       # 拒绝访问 PHP 文件（即使被上传）
       location ~ \.php$ {
           return 403;
       }
   
       # 静态文件配置
       location /static/ {
           alias /usr/local/nginx/staticfiles/;
           expires 30d;
           add_header Cache-Control "public, immutable";
       }
   
       # 媒体文件配置
       # location /media/ {
       #     alias /path/to/your/mediafiles/;
       # }
   
       # 所有其他请求代理到 Django upstream
       location / {
           # 请求限速（结合上面的 limit_req_zone）
           limit_req zone=req_limit_per_ip burst=60 nodelay;
   
           proxy_pass http://django_backend;
   
           # 保留客户端真实 IP
           proxy_set_header Host $host:$server_port;
           proxy_set_header X-Real-IP $remote_addr;
           proxy_set_header REMOTE-HOST $remote_addr;
           proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
           proxy_set_header X-Forwarded-Proto $scheme;
   
           # WebSocket 支持
           proxy_http_version 1.1;
           proxy_set_header Upgrade $http_upgrade;
           proxy_set_header Connection "upgrade";
   
           # 超时时间设置
           proxy_connect_timeout 10s;
           proxy_read_timeout 30s;
           proxy_send_timeout 10s;
   
           # 出错时可尝试下一个 upstream
           proxy_next_upstream error timeout invalid_header http_500 http_502 http_503 http_504;
       }
   }

   主要是这一段

   # 保留客户端真实 IP
   proxy_set_header Host $host:$server_port;
   proxy_set_header X-Real-IP $remote_addr;
   proxy_set_header REMOTE-HOST $remote_addr;
   proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
   proxy_set_header X-Forwarded-Proto $scheme;
   
   # WebSocket 支持
   proxy_http_version 1.1;
   proxy_set_header Upgrade $http_upgrade;
   proxy_set_header Connection "upgrade";

   测试curl命令如下（如果在本机测试要加上-H 'X-Forwarded-For: 10.20.30.40'，模拟公网IP）：

   curl -X POST 'https://xx.xx.com/xx/order/' -H 'accept: application/json' -H 'Content-Type: application/json' H 'X-CSRFTOKEN: lE3WXHdYpWIMU9dC37oMM9aLuSWHxqyLveW7l9N3h0lbZtgXlYoHg7FWbmAO' -H 'X-Forwarded-For: 10.20.30.40' -d '{
        "merchantNo": "Ssoryn",
        "method": "BAR01",
        "merchantOrderNo": "I-038642b2fc9b4ac7809f0d123e6600",
        "payAmount": "20"
   }'

2. 请求真实IP测试，用curl http://xx.xx.com去访问

   #!/usr/bin/env Python
   # -- coding: utf-8 --
   
   """
   @version: v1.0
   @author: huangyc
   @file: test_ip_server.py
   @Description: 
   @time: 2025/10/10 10:27
   """
   import datetime
   import ipaddress
   import json
   import re
   import socketserver
   from http.server import BaseHTTPRequestHandler
   from socketserver import ThreadingMixIn
   
   from basic_support.logger.logger_config import logger
   
   PORT = 80
   
   
   def validate_ip(ip, fallback):
       try:
           # 尝试解析，如果合法就返回标准格式字符串
           return str(ipaddress.ip_address(ip))
       except ValueError:
           # 不合法就返回 fallback
           return fallback
   
   
   class Request:
       def __init__(self, handler):
           self.META = {
               'REMOTE_ADDR': handler.client_address[0],
               'HTTP_X_FORWARDED_FOR': handler.headers.get('X-Forwarded-For'),
               'HTTP_X_REAL_IP': handler.headers.get('X-Real-IP'),
               'HTTP_FORWARDED': handler.headers.get('Forwarded'),
           }
   
   
   def get_real_ip(request):
       """
       根据多层 header 获取真实客户端 IP。
       优先级：
       1. Forwarded header
       2. X-Forwarded-For
       3. 其他常见代理 header
       4. REMOTE_ADDR
       """
       fallback = request.META.get('REMOTE_ADDR')
       try:
           ip = request.META.get('HTTP_FORWARDED')
           if ip and 'for=' in ip:
               match = re.search(r'for=([\d\.]+)', ip)
               if match:
                   return validate_ip(match.group(1), fallback)
   
           # X-Forwarded-For 第一条才是客户端真实 IP
           ip = request.META.get('HTTP_X_FORWARDED_FOR')
           if ip and ip.lower() != 'unknown':
               return validate_ip(ip.split(',')[0].strip(), fallback)
   
           # 其他可能的代理头
           for header in [
               'HTTP_PROXY_CLIENT_IP',
               'HTTP_WL_PROXY_CLIENT_IP',
               'HTTP_CLIENT_IP',
               'HTTP_X_FORWARDED_FOR'
           ]:
               ip = request.META.get(header)
               if ip and ip.lower() != 'unknown':
                   return validate_ip(ip, fallback)
   
           # 最后 fallback
           return fallback
       except Exception as e:
           logger.error("get_real_ip ERROR", exc_info=e)
           return fallback
   
   
   class Handler(BaseHTTPRequestHandler):
       def _send_json(self, data):
           body = json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")
           self.send_response(200)
           self.send_header("Content-Type", "application/json; charset=utf-8")
           self.send_header("Content-Length", str(len(body)))
           # allow javascript/curl from remote
           self.send_header("Access-Control-Allow-Origin", "*")
           self.end_headers()
           self.wfile.write(body)
   
       def do_GET(self):
           request = Request(self)
           real_ip = get_real_ip(request)
           logger.trace(f"获取真实IP: {real_ip}")
           # socket-level client address (what the TCP connection shows)
           client_ip = self.client_address[0]
   
           # headers of interest
           xff = self.headers.get("X-Forwarded-For")
           xreal = self.headers.get("X-Real-IP")
   
           data = {
               "timestamp": datetime.datetime.utcnow().isoformat() + "Z",
               "path": self.path,
               "client_ip_socket": client_ip,
               "x_forwarded_for": xff,
               "x_real_ip": xreal,
               "all_headers": dict(self.headers),
           }
   
           # Log to console (stdout)
           print("=" * 80)
           print("Request from:", client_ip, "path:", self.path)
           print("X-Forwarded-For:", xff)
           print("X-Real-IP:", xreal)
           print("All headers:")
           for k, v in self.headers.items():
               print(f"  {k}: {v}")
           print("=" * 80)
   
           self._send_json(data)
   
       # also allow POST if needed
       def do_POST(self):
           request = Request(self)
           real_ip = get_real_ip(request)
           logger.trace(f"获取真实IP: {real_ip}")
   
           content_length = int(self.headers.get("Content-Length", 0))
           body = self.rfile.read(content_length) if content_length else b""
           # reuse GET behavior but include body
           client_ip = self.client_address[0]
           xff = self.headers.get("X-Forwarded-For")
           xreal = self.headers.get("X-Real-IP")
           data = {
               "timestamp": datetime.datetime.utcnow().isoformat() + "Z",
               "path": self.path,
               "client_ip_socket": client_ip,
               "x_forwarded_for": xff,
               "x_real_ip": xreal,
               "all_headers": dict(self.headers),
               "body": body.decode(errors="replace"),
           }
           print("=" * 80)
           print("POST from:", client_ip, "path:", self.path)
           print("X-Forwarded-For:", xff)
           print("X-Real-IP:", xreal)
           print("Body:", body[:1000])
           print("=" * 80)
           self._send_json(data)
   
   
   class ThreadingHTTPServer(ThreadingMixIn, socketserver.TCPServer):
       allow_reuse_address = True
   
   
   if __name__ == "__main__":
       server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
       print(f"Listening on 0.0.0.0:{PORT}  -- Ctrl-C to stop")
       try:
           server.serve_forever()
       except KeyboardInterrupt:
           print("\nShutting down.")
           server.shutdown()
           server.server_close()

fastapi

🌟FastAPI 简介

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API。
它基于 Python 3.7+，使用 标准类型提示（type hints） 来实现自动验证、序列化和文档生成。

✅ FastAPI 的核心特点

1. 高性能
   • 基于 Starlette（异步 Web 框架） 和 Pydantic（数据验证库）。
   • 性能接近 Node.js 和 Go，远高于传统 Python 框架（如 Django / Flask）。
2. 自动生成 API 文档
   • 内置 Swagger UI 和 ReDoc。
   • 启动后可直接访问：
     • http://127.0.0.1:8000/docs
     • http://127.0.0.1:8000/redoc
3. 类型安全
   • 利用 Python 的类型提示，实现参数验证、自动补全、错误提示等。
   • 与 IDE 配合非常友好。
4. 异步支持
   • 原生支持 async/await，并发性能极佳。
   • 适合高并发的 API（如金融交易、物联网、实时服务等）。
5. 极简语法
   • 写法简洁直观，非常接近 Flask，但功能更强。

---

🚀FastAPI 示例

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 定义请求体模型
class Item(BaseModel):
    name: str
    price: float
    in_stock: bool = True

@app.get("/")
def read_root():
    return {"message": "Hello FastAPI"}

@app.post("/items/")
def create_item(item: Item):
    return {"item": item}

启动命令：

uvicorn main:app --reload

然后访问：

• 🌐 http://127.0.0.1:8000/docs → 自动交互式文档（Swagger UI）

---

⚖️FastAPI vs Django 对比

对比项	FastAPI	Django
定位	轻量级、高性能 API 框架	全功能 Web 框架（模板、ORM、Admin 全包）
性能	✅ 非常高（异步 IO 支持）	❌ 较慢（同步模型）
异步支持	✅ 原生支持 async/await	⚠️ 从 Django 3.1 开始才部分支持
ORM	可选（可用 SQLAlchemy、Tortoise ORM 等）	内置强大 ORM
表单/模板	❌ 无内置模板系统	✅ 完整模板系统
管理后台	❌ 需自建或用第三方	✅ 自带 Django Admin
自动文档	✅ 内置 Swagger/ReDoc	❌ 需插件（如 DRF + drf-yasg）
学习曲线	平滑，代码简洁	略陡，概念多
适用场景	高性能 API、微服务、AI/ML、金融系统	传统网站、管理后台、信息系统

---

💡总结建议

场景	推荐框架
想快速搭建完整网站（含后台、模板）	Django
专注于 API、数据交互、性能高	FastAPI
要写异步服务（如策略回测系统、数据接口）	FastAPI
教学项目、内容管理类网站	Django
需要前后端分离 + 高性能 API	FastAPI + Vue/React