分类目录归档：Python-Dev
解决方法： 使用on_commit注册事务提交后的回调
多线程
from django.db import transaction
transaction.on_commit(lambda: self._start_query_thread(log.id))
Django 的一个常见陷阱是立即触发任务，而不是等到 数据库事务的结束，这意味着 Celery 任务可能会运行 在所有更改都保存到数据库之前。
DeepSeek R1部署到debian,ubuntu，的无页面系统中
curl -fsSL https://ollama.com/install.sh | sh
使用ollama拉取并且运行(根据自己配置选择)
# 下载 1.5B 版本（适用于 CPU 及低配 GPU）
ollama run deepseek-r1:1.5b
# 下载 8B 版本
ollama run deepseek-r1:8b
# 下载 14B 版本
ollama run deepseek-r1:14b
# 下载 32B 版本
ollama run deepseek-r1:32b
# 下载 文件上传使用drf序列化注意事项
假如文件字段的数据是
{
"id": 1,
"name": test,
"file": /media/xxxxxxx/test.txt
})
序列化器要添加context={"request": request}返回的字段信息才是 (后端域名 + 文件路径)
ser = StockItemImpWorkSerializer(instance, context={"request": request})
上下文
###########
## seriimport asyncio
import time
from functools import wraps
from concurrent.futures import ThreadPoolExecutor
from typing import List, Callable, Any
# 创建数据库连接池
pool_executor = ThreadPoolExecutor(max_workers=10)
def to_async(func):
@wraps(func)
async def wrapper(*args, *pip install django-sslserver
xxxx_APPS = [
"sslserver"
]
python
manage.py
"runsslserver"
"--certificate"
"local.chenhaifeng.top_public.crt"
"--key"
"local.chenhaifeng.top.key"
"后端
@action(detail=False, methods=['get'], schema=AutoDocSchema())
def get_big_files(self, request):
def file_iterator(file_path, start_byte=0, chunk_size=256):
with open(file_path, 'rb') as f:
f.seek(我这里是发送邮件, 定时晚上6点执行
from datetime import datetime, timedelta
from django.conf import settings
from django.utils import timezone
from backend.cews.tasks import send_email_when_check_eform
import pytz
def schedule_email(recipient_email, subject, message):
"""
django中celery使用定时任务
celery_app.py
import os
from celery import Celery
# set the default Django settings module for the 'celery' program.
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.local")
app = Celery("backend")
# Using a string here me开启重试需要添加 bind=True参数
才能绑定到self来进行重试
@celery_app.task(bind=True, max_retries=3, default_retry_delay=5)
def send_email_when_check_eform(self, recipient_email, subject, message):
password = 'G@9742ns'
server_email = 'no-reply@gnstec.com.hk'
from_addr = forma我这里是定义数据结构模板， 来限制用户的dict内容
from ninja import Schema
from typing import TypeVar, Generic
T = TypeVar('T')
class ApiResSchema(Schema, Generic[T]):
data: T = ""
code: int = 0
msg: str = ""
django-ninja使用的时候可以通过[]写入类型， 框架自动序列化
@message_router.get(pat