server {
    listen 80;
    server_name domain.com;

    location / {
        proxy_pass http://127.0.0.1:8080;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        # SSE Support
        proxy_buffering off;
        proxy_cache off;
    }
}

定义模型

class TaskTracking(models.Model):
    task_id = models.CharField(max_length=100, null=False, blank=False, verbose_name='任务id', unique=True)
    task_type = models.CharField(max_length=100, null=False, blank=False, verbose_name='任务类型')
    task_pk = models.IntegerField(verbose_name='任务主键', null=False, blank=False, default=0)
    status = models.CharField(max_length=100, null=False, blank=False, verbose_name='任务状态')
    result = models.CharField(max_length=100, null=True, blank=True, verbose_name='任务结果', )
    created_at = models.DateTimeField(verbose_name='创建时间', editable=False, null=False, auto_now_add=True)
    updated_at = models.DateTimeField(verbose_name='更新时间', editable=False, null=False, auto_now=True)

    class Meta:
        verbose_name = "任务追踪"
        verbose_name_plural = verbose_name

迁移

python manage.py makemigrations && python manage.py migrate

定义装饰器

from functools import wraps
from celery.app.task import Task
from study.celery import app


def task_tracker(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 获取方法名作为 task_type
        task_type = func.__name__

        # 获取任务ID
        task_id = None
        if isinstance(args[0], Task):  # 检查第一个参数是否是 Celery 任务实例
            task_id = args[0].request.id

        # 获取 x 和 y 的值
        task_pk = args[1] if len(args) > 1 else kwargs.get('pk')

        # 任务开始，记录到数据库（包括 x 和 y 的值）
        task = TaskTracking.objects.create(task_id=task_id, task_type=task_type, task_pk=task_pk, status='started')

        try:
            result = func(*args, **kwargs)
            # 任务成功，更新数据库状态
            task.status = 'completed'
            task.result = result
            task.save()
            return result
        except Exception as e:
            # 任务失败，更新数据库状态
            task.status = 'failed'
            task.result = str(e)
            task.save()
            raise e

    return wrapper

use

@app.task(bind=True)
@task_tracker
def dosomething(self, pk):
    pass

Install the django-celery-results library:

pip install django-celery-results

Add django_celery_results to INSTALLED_APPS in your Django project’s settings.py:

INSTALLED_APPS = (
    ...,
    'django_celery_results',
)

Note that there is no dash in the module name, only underscores.

Create the Celery database tables by performing a database migrations:

python manage.py migrate django_celery_results

Configure Celery to use the django-celery-results backend.

Assuming you are using Django’s settings.py to also configure Celery, add the following settings:

CELERY_RESULT_BACKEND = 'django-db'
# For the cache backend you can use:

CELERY_CACHE_BACKEND = 'django-cache'

We can also use the cache defined in the CACHES setting in django.

# celery setting.
CELERY_CACHE_BACKEND = 'default'

# django setting.
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.db.DatabaseCache',
        'LOCATION': 'my_cache_table',
    }
}

获取到crt和key文件

mkdir /etc/nginx/cert
cp ssl.crt /etc/nginx/cert/ssl.crt
cp ssl.key /etc/nginx/cert/ssl.key

配置nginx

server {
        listen 443 ssl;
        server_name somedomain.com;

        ssl_certificate  /etc/nginx/cert/ssl.crt;
        ssl_certificate_key /etc/nginx/cert/ssl.key;
        ssl_protocols TLSv1 TLSv1.1 TLSv1.2;
        ssl_ciphers         HIGH:!aNULL:!MD5;

        #...
}

server {
    listen 80;
    server_name somedomain.com;
    # 跳转到https
    rewrite ^(.*)$  https://$host$1 permanent;
}

重启nginx

nginx -t
nginx -s reload

from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad

def encrypt_text(text: str, key_hex: str, iv_hex: str) -> str:
    key = bytes.fromhex(key_hex)
    iv = bytes.fromhex(iv_hex)
    text = text.encode('utf-8')
    cipher = AES.new(key, AES.MODE_CBC, iv)
    encrypted_data = cipher.encrypt(pad(text, AES.block_size))
    return encrypted_data.hex()


def decrypt_text(text: str, key_hex: str, iv_hex: str) -> str:
    key = bytes.fromhex(key_hex)
    iv = bytes.fromhex(iv_hex)
    text = bytes.fromhex(text)
    cipher = AES.new(key, AES.MODE_CBC, iv)
    decrypted_data = unpad(cipher.decrypt(text), AES.block_size)
    return decrypted_data.decode('utf-8')