1. 结论

该版本能在Django4.1.13上正常运行，且支持连接池

1.1. 如何配置Django使其支持达梦数据库和连接池

使用前需要安装dmPython、dmPython_pool、dmDjango三个包

配置Django的数据库时settings.py书写格式如下：

DATABASES = {
    'default': {
        'ENGINE': 'dmDjango',
        'NAME': 'DAMENG',
        'USER': 'SYSDBA',
        'PASSWORD': 'Dameng123',
        'HOST': 'localhost',
        'PORT': '5236',
        'OPTIONS': {
            'pool': {
                'max_size':100, # 连接池最大连接数（硬上限）
                'min_size':0, # 连接池最小连接数（可预热）
            }
        }
    }
}

支持的参数：

1.2. 如何使用连接池

dmDjango有三种使用连接池的方式，都是在 Django 项目内编写，申请连接都是自动的，区别在于谁负责归还。

一、原生SQL，使用connection.cursor

from django.db import connection

# 申请：首次使用 connection 时自动向池子借出连接
with connection.cursor() as cursor:
    cursor.execute('select count(*) from T_FIELD_TYPES')
    print(cursor.fetchone())
# with 块结束只释放游标，并不归还连接

# 释放：调用 close() 把连接归还连接池
connection.close()

二、ORM查询，申请和释放全部由Django自动完成

from django.http import HttpResponse
from dm.models import AllTypes

def index(request):
    books = AllTypes.objects.all()   # 自动向池子申请连接
    total = books.count()            # 复用同一条连接
    return HttpResponse(str(total))  # 请求结束 Django 自动归还

第一次执行查询时，Django 自动向连接池申请连接。请求结束时，Django将该调用归还到连接池

三、Django 管理命令 / 脚本，自动申请连接，可提前手动释放连接

from django.core.management.base import BaseCommand
from django.db import connection
from dm.models import AllTypes

class Command(BaseCommand):
    def handle(self, *args, **options):
        list(AllTypes.objects.all()[:1])   # 自动向池子申请连接
        with connection.cursor() as cursor:
            cursor.execute('select count(*) from T_FIELD_TYPES')
        # ... 执行其他操作 ...
        connection.close()                  # 提前归还池子（规范做法）

运行方式：

python manage.py 命令名

命令执行结束时 Django 的 BaseCommand 会统一调用 connections.close_all() 归还连接，因此不一定需要手动close()；手动 close() 用于命令中途提前释放连接，或在长时间运行场景下及时归还连接

2. 问题背景

基于客户需求，开发那边提供了一个支持DmPythonPool的DmDjango驱动，需要测试一下是否能够正常对各种常见数据类型进行增删查改，以及多个线程尝试连接时是否能正常使用

2.1. 测试环境

内容	版本/类型
操作系统	Kylin-Server-V10-SP3-2403
CPU	x86_64bit
达梦数据库	DM8
Python	3.10
Django	4.1.13
dmPython	2.5.38
dmPython_pool	1.0.0
dmDjango	3.0.10

3. 使用步骤示例

3.1. 安装

3.1.1. 安装dmPython

进入dmPython文件夹执行

python setup.py install

如果报错找不到libdpi.so，需要配置LD_LIBRARY_PATH，将达梦安装目录下面的bin目录添加到LD_LIBRARY_PATH

export LD_LIBRARY_PATH=/dm/dmdbms/bin:$LD_LIBRARY_PATH

3.1.2. 安装dmPython_pool

进入dmPython_pool文件夹执行

python setup.py install

3.1.3. 安装dmDjango

python setup.py install

3.2. 使用

3.2.1. 创建项目

创建项目，这里以项目名为mysite为例

django-admin startproject mysite
tree

Django创建了一个mysite目录，该目录包含mysite文件夹与manage.py文件

.
└── mysite
    ├── manage.py
    └── mysite
        ├── asgi.py
        ├── __init__.py
        ├── settings.py
        ├── urls.py
        └── wsgi.py

2 directories, 6 files

3.2.2. 配置数据库

将Django的默认数据库设置为达梦数据库，需要修改settings.py中的DATABASES

vi mysite/mysite/settings.py

修改为如下内容：

DATABASES = {
    'default': {
        'ENGINE': 'dmDjango',
        'NAME': 'DAMENG',
        'USER': 'SYSDBA',
        'PASSWORD': 'Dameng123',
        'HOST': 'localhost',
        'PORT': '5237',
        'OPTIONS': {
            'pool': {
                'max_size':100,
                'min_size':0,
            }
        }
    }
}

3.2.3. 创建应用

这里以应用名为dm为例

cd mysite
python manage.py startapp dm

3.2.4. 创建模型

编辑models.py文件

vi dm/models.py

创建的模型AllTypes映射到表T_FIELD_TYPES，表内有达梦常见数据类型：

• 字符类型：CharField映射为NVARCHAR2，TextField映射为TEXT
• 数值类型：DecimalField映射为 NUMBER，IntegerField映射为INTEGER，SmallIntegerField映射为SMALLINT，BigIntegerField映射为 BIGINT
• 布尔与位串：BooleanField映射为TINYINT（0/1）
• 日期时间：DateField映射为DATE，TimeField、DateTimeField映射为TIMESTAMP(6)
• 大字段：BinaryField映射为BLOB，TextField映射为TEXT

写入如下内容：

from django.db import models

class AllTypes(models.Model):
    class Meta:
        db_table = 'T_FIELD_TYPES'

    # 主键：INTEGER IDENTITY(1,1)
    ID = models.AutoField(primary_key=True, db_column='ID')

    # ---- 字符类型 ----
    C_CHAR = models.CharField(max_length=20, null=True)       # -> NVARCHAR2(20)
    C_VARCHAR2 = models.CharField(max_length=50, null=True)   # -> NVARCHAR2(50)
    C_TEXT = models.TextField(null=True)                      # -> TEXT

    # ---- 数值类型 ----
    N_NUMBER = models.DecimalField(max_digits=10, decimal_places=2, null=True)  # -> NUMBER(10,2)
    N_INT = models.IntegerField(null=True)                    # -> INTEGER
    N_SMALLINT = models.SmallIntegerField(null=True)          # -> SMALLINT
    N_BIGINT = models.BigIntegerField(null=True)              # -> BIGINT
    N_TINYINT = models.BooleanField(null=True)                # -> TINYINT（0/1）

    # ---- 位串（布尔，0/1）----
    B_BIT = models.BooleanField(default=False)                # -> TINYINT（0/1）

    # ---- 日期时间 ----
    D_DATE = models.DateField(null=True)                      # -> DATE
    T_TIME = models.TimeField(null=True)                      # -> TIMESTAMP(6)
    T_DATETIME = models.DateTimeField(null=True)              # -> TIMESTAMP(6)

    # ---- 大字段 ----
    L_BLOB = models.BinaryField(null=True)                    # -> BLOB
    L_CLOB = models.TextField(null=True, db_column='L_CLOB')  # -> TEXT

    # ---- 其他常见 ----
    O_IP = models.GenericIPAddressField(null=True)            # -> VARCHAR(39)
    O_UUID = models.UUIDField(null=True)                      # -> VARCHAR(32)
    O_JSON = models.JSONField(null=True)                      # -> JSON

还需要修改settings.py文件中的INSTALLED_APPS，添加’dm.apps.DmConfig’，将dm应用安装到Django 项目里：

INSTALLED_APPS = [
    ...
    'dm.apps.DmConfig',
]

3.2.5. 进行数据迁移

在项目的mysite目录下执行该命令

python manage.py makemigrations dm
python manage.py migrate

3.2.6. 对数据库进行操作

在dm目录中新建一个services.py，实现增删查改

from .models import Book
#查询操作
def fun_get_books():
	books = Book.objects.all() #返回 QuerySet
	total_rows=books.count()
	print('查询到'+str(total_rows)+'条记录')
	for book in books:
		print(book.NAME, book.AUTHOR, book.PUBLISHER)
		print('')
#增加操作
def fun_insert_book():
	NewBook = Book()
	NewBook.NAME = '水浒传'
	NewBook.AUTHOR = '施耐庵，罗贯中'
	NewBook.PUBLISHER = '中华书局'
	NewBook.PUBLISHTIME = '2005-4-1'
	NewBook.PRODUCTNO = '9787101046137'
	NewBook.SAFETYSTOCKLEVEL = '10'
	NewBook.ORIGINALPRICE = '19'
	NewBook.NOWPRICE = '14.3'
	NewBook.DISCOUNT = '7.5'
	NewBook.DESCRIPTION = '''
	《水浒传》是宋江起义故事在民间长期流传基础上产生出来的，吸收了民间文学的营养。
	'''
	NewBook.PHOTO = ''
	NewBook.TYPE = '16'
	NewBook.PAPERTOTAL = '922'
	NewBook.WORDTOTAL = '912000'
	NewBook.SELLSTARTTIME = '2006-03-20'
	NewBook.SELLENDTIME = '2021-07-12'
	NewBook.save()
	print('插入成功')
#更新操作
def fun_update_book():
	try:
		obj = Book.objects.get(PRODUCTID=1) #返回结果有且只有一个
		obj.NAME='水浒'
		obj.save()
		print('更新成功')
	except Book.DoesNotExist:
		print('PRODUCTID=1 的记录不存在，未更新任何数据')
#删除操作
def fun_delete_book():
	obj = Book.objects.filter(NAME='水浒') #返回 QuerySet
	del_rows=obj.count()
	obj.delete()
	print('删除'+str(del_rows)+'条记录')

在dm/views.py中新增

from django.shortcuts import render

# Create your views here.
from django.http import HttpResponse
from .services import fun_get_books
from .services import fun_insert_book
from .services import fun_update_book
from .services import fun_delete_book
def index(request):
	# 查询
	fun_get_books()
	# 增加
	fun_insert_book()
	fun_get_books()
	# 更新
	fun_update_book()
	fun_get_books()
	# 删除
	fun_delete_book()
	fun_get_books()
	return HttpResponse("执行结束，请到控制台查看输出结果。")

在 dm 目录里新建 urls.py 文件

from . import views
from django.urls import path
urlpatterns = [
	path('', views.index, name='index'),
]

在 mysite\urls.py 文件的 urlpatterns 列表中指定创建的 dm.urls 模块

from django.contrib import admin
from django.urls import path, include
urlpatterns = [
	path('dm/',include('dm.urls')),
	path('admin/', admin.site.urls),
]

3.2.7. 启动

在命令行工具中输入以下命令，启动 Django 自带的简易服务器

python manage.py runserver

说明对常见数据类型的增删查改均正常

3.2.8. 测试连接池在多线程时的效果

编写一个Django管理命令pooltest来验证连接池效果

在dm目录中新建management/commands子目录：

mkdir -p dm/management/commands

创建两个空的__init__.py文件：

touch dm/management/__init__.py dm/management/commands/__init__.py

在dm/management/commands/目录中新建pooltest.py，写入如下内容：

import threading

from django.core.management.base import BaseCommand
from django.db import connection
from dm.models import AllTypes


class Command(BaseCommand):
    help = '测试 dmDjango 连接池：单进程复用 + 并发压测 + 数据库会话数。用法: python manage.py pooltest [线程数] [每线程查询数]'

    def add_arguments(self, parser):
        parser.add_argument('threads', nargs='?', type=int, default=5, help='并发线程数')
        parser.add_argument('queries', nargs='?', type=int, default=20, help='每线程查询次数')

    def handle(self, *args, **options):
        pool = connection.pool
        threads_count = options['threads']
        queries_per_thread = options['queries']
        self.stdout.write('池子对象类型: %s (max_size=100, 等待超时30s)\n' % type(pool).__name__)

        def show(tag):
            s = pool.stats()
            self.stdout.write(
                '%s: total=%d created_total=%d borrowed_total=%d idle=%d in_use=%d' % (
                    tag, s['total'], s['created_total'], s['borrowed_total'],
                    s['idle'], s['in_use']))

        # ---------- 1. 单进程连接复用 ----------
        self.stdout.write(self.style.MIGRATE_HEADING('[1] 单进程连接复用测试 (10轮借出-归还)'))
        show('初始')
        for _ in range(10):
            list(AllTypes.objects.all()[:1])
            connection.close()
        show('10轮借还后')
        s1 = pool.stats()
        reuse_ok = (s1['created_total'] == 1 and s1['borrowed_total'] >= 10)
        self.stdout.write('结论: 只创建了 %d 条物理连接, 复用了 %d 次 -> %s\n' % (
            s1['created_total'], s1['borrowed_total'],
            self.style.SUCCESS('正常') if reuse_ok else self.style.ERROR('异常')))

        # ---------- 2. 并发压测 ----------
        self.stdout.write(self.style.MIGRATE_HEADING(
            '[2] 并发压测 (%d线程 x %d次查询, max_size=100)' % (threads_count, queries_per_thread)))
        before = self._session_count()
        errors = []

        def worker():
            from django.db import connection as th_conn
            try:
                for _ in range(queries_per_thread):
                    list(AllTypes.objects.all()[:1])
            except Exception as e:
                errors.append('%s: %s' % (type(e).__name__, str(e)[:80]))
            finally:
                th_conn.close()

        threads = [threading.Thread(target=worker) for _ in range(threads_count)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        show('并发结束后')
        s2 = pool.stats()
        after = self._session_count()
        self.stdout.write('SYSDBA 数据库会话数: before=%d -> after=%d (max_size=100)' % (before, after))
        if errors:
            self.stdout.write(self.style.ERROR('并发期间报错 %d 条:' % len(errors)))
            for e in errors[:5]:
                self.stdout.write('  ' + e)
        else:
            self.stdout.write('并发期间无报错 -> %s' % self.style.SUCCESS('正常'))
        total_queries = threads_count * queries_per_thread
        self.stdout.write('共 %d 次查询, 累计创建 %d 条物理连接 (无池子会是 %d 条)' % (
            total_queries, s2['created_total'], total_queries))

        # ---------- 3. 汇总 ----------
        self.stdout.write('\n' + self.style.MIGRATE_HEADING('[3] 汇总'))
        all_ok = reuse_ok and not errors
        if all_ok:
            self.stdout.write(self.style.SUCCESS(
                '连接池工作正常: 连接被复用, 并发安全, 会话数受池子控制'))
        else:
            self.stdout.write(self.style.ERROR('连接池存在异常, 请检查'))

    def _session_count(self):
        cur = connection.cursor()
        cur.execute("select count(*) from V$SESSIONS where USER_NAME=%s", ['SYSDBA'])
        return cur.fetchone()[0]

运行方式如下，第1个参数是线程数，第2个参数是每线程查询数，不传默认5线程×20次：

python manage.py pooltest
python manage.py pooltest 200 20

判断方法：复用测试中created_total始终保持1而borrowed_total增长，说明连接被复用；并发压测无报错且创建的物理连接数远小于查询总数，说明并发安全、池子正常。

运行结果如下：

说明dmDjango的连接池在多线程情况下能正常使用