当前位置： 首页 > news >正文

python简单操作达梦数据库

news 2025/8/2 8:08:23

Python数据库接口

Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本语言。运行 Python 程序需要解释器的支持，只要在不同的平台安装不同的解释器，Python 代码就可以跨平台运行，可移植性强，不用担心任何兼容性问题。

DMPython 是 DM 提供的依据 Python DB API version 2.0 中 API 使用规定而开发的数据库访问接口。DMPython 实现这些 API，使 Python 应用程序能够对 DM 数据库进行访问。

DMPython 通过调用 DM DPI 接口完成 Python 模块扩展。在其使用过程中，除 Python 标准库以外，还需要 DPI 的运行环境。

数据库连接

Python 接口登录、登出示例程序 py_conn.py 如下（注意把端口号和密码更换）：

#!/usr/bin/python

#coding:utf-8

import dmPython

try:

    conn = dmPython.connect(user='SYSDBA', password='*****', server='localhost',  port=51236)

    cursor  = conn.cursor()

    print('python: conn success!')

    conn.close()

except (dmPython.Error, Exception) as err:

    print(err)

  

开发示例

基础操作示例

Python 接口增、删、改、查四个基本操作，示例程序 py_dml.py 如下：

#!/usr/bin/python

#coding:utf-8

import dmPython

try:

    conn = dmPython.connect(user='SYSDBA', password='*****', server='localhost',  port=51236)

    cursor  = conn.cursor()

    try:

        #清空表，初始化测试环境

        cursor.execute ('delete from PRODUCTION.PRODUCT_CATEGORY')

    except (dmPython.Error, Exception) as err:

        print(err)

    try:

        #插入数据

        cursor.execute ("insert into PRODUCTION.PRODUCT_CATEGORY(NAME) values('语文'), ('数学'), ('英语'), ('体育')")

        print('python: insert success!')

        #删除数据

        cursor.execute ("delete from PRODUCTION.PRODUCT_CATEGORY where name='数学'")

        print('python: delete success!')

        #更新数据

        cursor.execute ('update PRODUCTION.PRODUCT_CATEGORY set name = \'英语-新课标\' where name=\'英语\'')

        print('python: update success!')

        #查询数据

        cursor.execute ("select name from PRODUCTION.PRODUCT_CATEGORY")

        res = cursor.fetchall()

        for tmp in res:

            for c1 in tmp:

                print(c1)

        print('python: select success!')

    except (dmPython.Error, Exception) as err:

        print(err)

    conn.close()

except (dmPython.Error, Exception) as err:

    print(err)

 

绑定变量 

Python 接口绑定变量示例程序 py_bind.py 如下：

#!/usr/bin/python

#coding:utf-8

import dmPython

try:

    conn = dmPython.connect(user='SYSDBA', password=*****, server='localhost',  port=51236)

    cursor  = conn.cursor()

    try:

        #清空表，初始化测试环境

        cursor.execute ('delete from PRODUCTION.PRODUCT_CATEGORY')

    except (dmPython.Error, Exception) as err:

        print(err)

    try:

        #插入数据

        values = ('物理')

        cursor.execute ("insert into PRODUCTION.PRODUCT_CATEGORY(name) values(?)", values)

        print('python: insert success!')

        #查询数据

        cursor.execute ("select name from PRODUCTION.PRODUCT_CATEGORY")

        res = cursor.fetchall()

        for tmp in res:

            for c1 in tmp:

                print(c1)

        print('python: select success!')

    except (dmPython.Error, Exception) as err:

        print(err)

    conn.close()

except (dmPython.Error, Exception) as err:

    print(err)

 

 大字段操作

Python 接口操作大字段（本例以 blob、clob 为例）示例程序 py_blob.py 如下：

import sys

longstring = ""

longstring += 'ABCDEF0123456789' * 500

cvalue = longstring

if sys.version_info[0] >= 3 :

    bvalue = longstring.encode("ascii")

else :

    bvalue = longstring

import dmPython

conn = dmPython.connect(user='SYSDBA', password='*****', server='localhost',  port=51236)

cursor = conn.cursor()

try:

    #清理测试环境

    cursor.execute("select object_id from all_objects where object_type='TABLE' and OBJECT_NAME='BIG_DATA';")

    bigdata_id = cursor.fetchone()

    if(bigdata_id):

        cursor.execute('drop table PRODUCTION.BIG_DATA;')

        print('drop table success')

    cursor.execute('create table PRODUCTION.BIG_DATA(c1 blob, c2 clob)')

    print('create table success!')

    cursor.execute('insert into PRODUCTION.BIG_DATA values(?, ?)', bvalue, cvalue)

    print('insert success!')

    cursor.execute('select * from PRODUCTION.BIG_DATA')

    print('select success!')

    cursor.description

    row = cursor.fetchone()

    (blob, clob) = row

    if sys.version_info[0] >= 3 :

        type(blob)

        type(clob)

    blob

    clob

except(dmPython.Error,Exception) as err:

    print(err)

conn.close

 

多线程 

Python 接口利用多线程连接数据库示例程序 py_multi.py 如下：

#!/usr/bin/python

# -*- coding: UTF-8 -*-

import dmPython

import _thread

import time

# 为线程定义一个函数

def print_time( threadName, delay):

    count = 0

    while count < 3:

        try:

            conn = dmPython.connect(user='SYSDBA',password='*****',server='localhost',port=51236)

            cursor = conn.cursor()

            print ("%s: %s" % ( threadName, time.ctime(time.time()) ))

            conn.close()

            time.sleep(delay)

            count += 1

        except(dmPython.Error, Exception) as err:

            print(err)

# 创建两个线程

try:

    _thread.start_new_thread(print_time, ("Thread-1", 1,))

    _thread.start_new_thread(print_time, ("Thread-2", 2,))

except:

    print ("Error: unable to start thread")

while 1:

    pass

SQLAlchemy 框架 

SQLAlchemy 是 Python 社区中最广泛使用的 ORM 框架之一，该框架建立在数据库 DBAPI 之上，使用关系对象映射进行数据库操作，即将对象转换成 SQL，然后使用数据库 DBAPI 执行 SQL 并获取执行结果。SQLAlchemy 操纵的是 Python 对象而不是 SQL 查询，也就是在代码层面考虑的是对象，而不是 SQL，体现的是一种程序化思维，这样使得 Python 程序更加简洁易读

开发示例

基础操作示例
创建数据库 PRODUCT 表
CREATE TABLE "SYSDBA"."PRODUCT"
(
"PRODUCTID" INT IDENTITY(1, 1) NOT NULL,
"NAME" VARCHAR(100) NOT NULL,
"AUTHOR" VARCHAR(25) NOT NULL,
"PUBLISHER" VARCHAR(50) NOT NULL,
"PUBLISHTIME" DATE NOT NULL,
"PRODUCTNO" VARCHAR(25) NOT NULL,
"SATETYSTOCKLEVEL" SMALLINT NOT NULL,
"ORIGINALPRICE" DEC(19,4) NOT NULL,
"NOWPRICE" DEC(19,4) NOT NULL,
"DISCOUNT" DECIMAL(2,1) NOT NULL,
"DESCRIPTION" TEXT,
"TYPE" VARCHAR(5),
"PAPERTOTAL" INT,
"WORDTOTAL" INT,
"SELLSTARTTIME" DATE NOT NULL,
"SELLENDTIME" DATE,
NOT CLUSTER PRIMARY KEY("PRODUCTID"),
UNIQUE("PRODUCTNO")) STORAGE(ON "BOOKSHOP", CLUSTERBTR);
创建 Product.py 文件
# coding: utf-8
from sqlalchemy import Column, Integer, String,Date,Numeric,Text
from sqlalchemy.ext.declarative import declarative_base
# 创建对象的基类:
Base = declarative_base()class Product(Base):# 表的名字:__tablename__ = 'product'# 表的结构:PRODUCTID = Column(Integer,autoincrement=True, primary_key=True)NAME = Column(String(100))AUTHOR = Column(String(25))PUBLISHER = Column(String(50))PUBLISHTIME = Column(Date)PRODUCTNO = Column(String(25))SATETYSTOCKLEVEL = Column(Integer)ORIGINALPRICE = Column(Numeric(19,4))NOWPRICE = Column(Numeric(19,4))DISCOUNT = Column(Numeric(2,1))DESCRIPTION = Column(Text)TYPE = Column(String(5))PAPERTOTAL = Column(Integer)WORDTOTAL = Column(Integer)SELLSTARTTIME = Column(Date)
SELLENDTIME = Column(Date)
创建 dm_sqlalchemy.py 文件
# coding: utf-8
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from Product import Product
def main():#dialect 是SQLAlchemy用来与各种类型的DBAPI和数据库通信的系统。conn_url = 'dm+dmPython://SYSDBA: *****@192.168.201.118:5236'#创建Engine对象engine = create_engine(conn_url)#创建DBSession对象DBSession = sessionmaker(bind=engine)fun_select_all(DBSession)# 插入fun_insert(DBSession)fun_select_all(DBSession)# 更新fun_update(DBSession)fun_select_all(DBSession)# 删除fun_delete(DBSession)fun_select_all(DBSession)def fun_select_all(DBSession):# 创建Sessionsession = DBSession()# 查询所有的list_product = session.query(Product).all()print('查询所有结果：')for product in list_product:print(product.NAME, product.AUTHOR, product.PUBLISHER )print('')session.close()def fun_insert(DBSession):# 创建Sessionsession = DBSession()new_product = Product()new_product.NAME = '水浒传'new_product.AUTHOR = '施耐庵，罗贯中'new_product.PUBLISHER = '中华书局'new_product.PUBLISHTIME = '2005-4-1'new_product.PRODUCTNO = '9787101046137'new_product.SATETYSTOCKLEVEL = '10'new_product.ORIGINALPRICE = '19'new_product.NOWPRICE = '14.3'new_product.DISCOUNT = '7.5'new_product.DESCRIPTION = '''  《水浒传》是宋江起义故事在民间长期流传基础上产生出来的，吸收了民间文学的营养。'''new_product.PHOTO = ''new_product.TYPE = '16'new_product.PAPERTOTAL = '922'new_product.WORDTOTAL = '912000'new_product.SELLSTARTTIME = '2006-03-20'new_product.SELLENDTIME = ''session.add(new_product)session.commit()print('插入成功')session.close()def fun_update(DBSession):# 创建Sessionsession = DBSession()product = session.query(Product).filter(Product.NAME == '水浒传').one()product.NAME = '水浒'session.commit()print('更新成功')session.close()def fun_delete(DBSession):# 创建Sessionsession = DBSession()session.query(Product).filter(Product.NAME == '水浒').delete()session.commit()print('删除成功')session.close()if __name__ == '__main__':main()

Flask 框架

 Flask 是一个轻量级的可定制框架，使用 Python 语言编写，较其他同类型框架更为灵活、轻便、安全且容易上手。它可以很好地结合 MVC 模式进行开发，开发人员分工合作，小型团队在短时间内就可以完成功能丰富的中小型网站或 Web 服务的实现。另外，Flask 还有很强的定制性，用户可以根据自己的需求来添加相应的功能，在保持核心功能简单的同时实现功能的丰富与扩展，其强大的插件库可以让用户实现个性化的网站定制，开发出功能强大的网站。Flask-SQLAlchemy 集成了 SQLAlchemy，它简化了连接数据库服务器、管理数据库操作会话等各类工作，让 Flask 中的数据处理体验变得更加轻松。

创建测试表

create table "SYSDBA"."FLASK"
("PRODUCTID" INTEGER not null ,"NAME" VARCHAR(50),"AUTHOR" VARCHAR(50),"PUBLISHER" VARCHAR(50),"PUBLISHTIME" DATE,"PRODUCTNO" VARCHAR(50),"SATETYSTOCKLEVEL" INTEGER,"ORIGINALPRICE" DECIMAL(22, 6),"NOWPRICE" DECIMAL(22, 6),"DISCOUNT" DECIMAL(22, 6),"DESCRIPTION" TEXT,"PHOTO" BLOB,"TYPE" VARCHAR(50),"PAPERTOTAL" INTEGER,"WORDTOTAL" INTEGER,"SELLSTARTTIME" DATE,"SELLENDTIME" DATE,primary key("PRODUCTID")
)
storage(initial 1, next 1, minextents 1, fillfactor 0)
;

Flask 应用开发

新建 Flask 项目

使用 PyCharm 工具新建 Flask 项目。

初始化 Flask 和 SQLAlchemy
编辑 FlaskDm/application/_init _.py 文件。
# -*- coding: utf-8 -*-from flask import Flask
from flask_sqlalchemy import SQLAlchemydb = SQLAlchemy()def init_app():app = Flask(__name__)# 解决页面中问乱码问题app.config['JSON_AS_ASCII'] = False# 设置数据库链接地址app.config['SQLALCHEMY_DATABASE_URI'] = "dm+dmPython://SYSDBA:*****@127.0.0.1:5236"# 设置显示底层执行的sql语句app.config['SQLALCHEMY_ECHO'] = Trueapp.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True# 初始化组件对象，关联flask应用db.init_app(app)return app
构建模型类
编辑 FlaskDm/application/models/ProductModel.py 文件。
# -*- coding: utf-8 -*-# 构建模型类
from application._init_ import db# 构建模型类
class Product(db.Model):# 设置表名，表名默认为类名小写__tablename__ = 'FLASK'PRODUCTID = db.Column(db.Integer, primary_key=True, autoincrement=True)NAME = db.Column(db.String(100))AUTHOR = db.Column(db.String(25))PUBLISHER = db.Column(db.String(50))PUBLISHTIME = db.Column(db.Date)PRODUCTNO = db.Column(db.String(25))SATETYSTOCKLEVEL = db.Column(db.Integer)ORIGINALPRICE = db.Column(db.Numeric(19, 4))NOWPRICE = db.Column(db.Numeric(19, 4))DISCOUNT = db.Column(db.Numeric(2, 1))DESCRIPTION = db.Column(db.Text)# 图片转化为二进制数据进行存储PHOTO = db.Column(db.LargeBinary)TYPE = db.Column(db.String(5))PAPERTOTAL = db.Column(db.Integer)WORDTOTAL = db.Column(db.Integer)SELLSTARTTIME = db.Column(db.Date)SELLENDTIME = db.Column(db.Date)def __repr__(self):  # 自定义 交互模式 & print() 的对象打印return "(%d, %s, %s, %s)" % (self.PRODUCTID, self.NAME, self.AUTHOR, self.DESCRIPTION)
完整代码
编辑 FlaskDm/FlaskDm.py 文件。
# -*- coding: utf-8 -*-
from flask import Response, jsonfrom application._init_ import init_app, db
from application.ProductModel import Productapp = init_app()@app.route("/query_all")
def query_all():# 查询所有用户数据, 返回User列表,data = Product.query.all()response_data = []for item in data:info = dict()info['PRODUCTID'] = item.PRODUCTIDinfo['NAME'] = item.NAMEinfo['AUTHOR'] = item.AUTHORinfo['DESCRIPTION'] = item.DESCRIPTIONresponse_data.append(info)return Response(json.dumps(response_data), mimetype='application/json')@app.route("/query_first")
def query_filter():# 查询所有用户数据, 返回User列表,data = Product.query.filter_by(NAME='面纱').first()response_data = dict()response_data['PRODUCTID'] = data.PRODUCTIDresponse_data['NAME'] = data.NAMEresponse_data['AUTHOR'] = data.AUTHORresponse_data['DESCRIPTION'] = data.DESCRIPTIONreturn Response(json.dumps(response_data), mimetype='application/json')@app.route("/add_one")
def add_one():# 添加一条数据# 将图片转化为二进制数据进行存储product = Product(PRODUCTID=1,NAME='万历十五年',AUTHOR='黄仁宇',PUBLISHER='中华书局',PUBLISHTIME='1981-09-01',PRODUCTNO='9787101046126',SATETYSTOCKLEVEL=10, ORIGINALPRICE=39.8000, NOWPRICE=20.0000,DISCOUNT=5.0,DESCRIPTION='《万历十五年》以1587年为关节点，在历史的脉络中延伸，从政治、经济、军事等各个方面的历史大事与人物着手，记叙了明朝中晚期的种种社会矛盾和开始走向衰败的迹象。',PHOTO=open('C:/Users/Administrator/PycharmProjects/FlaskDm/application/OIP.jpg', 'rb').read(),TYPE='16', PAPERTOTAL=684, WORDTOTAL=68000,SELLSTARTTIME='2006-03-20', SELLENDTIME='2023-11-1')db.session.add(product)db.session.commit()return Response('数据插入成功，本次插入数据的PRODUCTID为{}'.format(product.PRODUCTID), mimetype='application/json')@app.route("/add_list")
def add_list():# 一次添加多条数据data_list = [Product(PRODUCTID=2,NAME='面纱', AUTHOR='威廉·萨默塞特·毛姆', PUBLISHER='人民文学出版社', PUBLISHTIME='2020-04-01',PRODUCTNO='9787101046127',SATETYSTOCKLEVEL=10, ORIGINALPRICE=39.8000, NOWPRICE=20.0000,DISCOUNT=5.0,DESCRIPTION='小说的故事发生在香港和一个叫“湄潭府”的地方。女主人公凯蒂·费恩因为和香港助理布政司查理通奸，被丈夫瓦尔特（细菌学家）发现后胁迫她去了霍乱横行的湄潭府，最终瓦尔特不幸染病死去，凯蒂回到香港，重投查理怀抱后羞愧不已，最终回到英国和父亲和解，并和父亲同往巴哈马群岛生活。',PHOTO=open('C:/Users/Administrator/PycharmProjects/FlaskDm/application/ORF.jpg', 'rb').read(),TYPE='16', PAPERTOTAL=684, WORDTOTAL=68000, SELLSTARTTIME='2006-03-20',SELLENDTIME='2023-11-1'),Product(PRODUCTID=3,NAME='乖，摸摸头', AUTHOR='大冰', PUBLISHER='湖南文艺出版社', PUBLISHTIME='2014-10-01',PRODUCTNO='9787101046129',SATETYSTOCKLEVEL=10, ORIGINALPRICE=39.8000, NOWPRICE=20.0000,DISCOUNT=5.0,PHOTO=open('C:/Users/Administrator/PycharmProjects/FlaskDm/application/OWW.jpg', 'rb').read(),DESCRIPTION='此书记录了大冰十余年的江湖游历，以及他和他朋友们的爱与温暖的传奇故事。',TYPE='16', PAPERTOTAL=684, WORDTOTAL=68000, SELLSTARTTIME='2006-03-20',SELLENDTIME='2023-11-1')]db.session.add_all(data_list)db.session.commit()return Response('数据插入成功，本次共插入{}条数据'.format(len(data_list)))@app.route("/updata_info")
def updata_info():# 直接根据条件修改Product.query.filter(Product.PRODUCTID == 2).update({Product.NAME: '面纱(中文)'})db.session.commit()# 查询修改后的值data = Product.query.filter_by(PRODUCTID=2).first()response_data = dict()response_data['PRODUCTID'] = data.PRODUCTIDresponse_data['NAME'] = data.NAMEreturn Response(json.dumps(response_data), mimetype='application/json')@app.route("/delete_info")
def delete_info():# 直接根据条件修改Product.query.filter(Product.PRODUCTID == 1).delete()db.session.commit()# 查询修改后的值t_count = Product.query.count()return Response('数据删除成功！目前表中还有{}条数据！'.format(t_count))@app.route("/")
def HELLO_DM():return 'HELLO DM'if __name__ == '__main__':app.run(host="127.0.0.1", port=5000, debug=True)

 

 

Django 框架 

Django 是一个开放源代码的 Python Web 应用框架，采用 MVT 的软件设计模式，即模型（Model），视图（View）和模板（Template）。Django 将 MVC 中的视图进一步分解为 Django 视图和 Django 模板两个部分。MVC 中的控制器由 Django 框架的 URLconf 来实现。

Django 被认为是"大而全"的重量级 Web 框架，其自带大量的常用工具和组件（比如数据库 ORM 组件、用户认证、权限管理、分页、缓存)，甚至还自带了管理后台 Admin，适合快速开发功能完善的企业级网站。

URL 映射器：用于根据请求 URL 将 HTTP 请求重定向到相应的视图。

View：视图是一个请求处理函数，它接收 HTTP 请求并返回 HTTP 响应。

Models：模型是定义应用程序数据结构的 Python 对象，并提供在数据库中管理（添加，修改，删除）和查询记录的机制。Django 的 Model 层自带数据库 ORM 组件。

Django 应用开发

使用 PyCharm 新建项目

创建 Django 项目 

在 PyCharm 中打开 Terminal 执行：

django-admin startproject mysite

 

创建 dm 应用 

在 Django 中，项目是一个网站使用的配置和应用的集合。项目可以包含很多个应用。

cd mysite

python manage.py startapp dm

创建视图 
编辑 dm/views.py 文件。
from django.http import HttpResponsedef index(request):
return HttpResponse("Hello, 达梦数据库。 ")
在 dm 目录里新建 urls.py 文件
在 dm 目录里新建 urls.py 文件。from . import views
from django.urls import path
urlpatterns = [path('', views.index, name='index'),
]

 

在根 URLconf 文件中指定创建的 dm.urls 模块
在 mysite/urls.py 文件的 urlpatterns 列表里插入一个 include()。
from django.contrib import admin
from django.urls import include, pathurlpatterns = [path('dm/', include('dm.urls')),path('admin/', admin.site.urls),
]
启动 Django 自带的简易服务器，验证是否正常工作

在 PyCharm 中打开 Terminal 执行：

python manage.py runserver

达梦社区地址：达梦数据库 - 新一代大型通用关系型数据库 | 达梦在线服务平台https://eco.dameng.com/

http://www.lryc.cn/news/607028.html

相关文章：

  • VUE2 学习笔记15 代理
  • Leetcode 10 java
  • Go语言中的盲点：竞态检测和互斥锁的错觉
  • 常见的网络攻击以及预防
  • C++ 构造函数语义学
  • Context API
  • 【AI论文】具备测试时扩散能力的深度研究者
  • win11怎么看本机ip地址？怎么查看代理端口？
  • leetcode 118. 杨辉三角 简单
  • 【C#学习Day14笔记】泛型、集合(数组列表Arraylist、列表list)与字典
  • 基于单片机汽车少儿安全预警系统
  • 118. 杨辉三角
  • 数据结构：在链表中查找（Searching in a Linked List）
  • [ java 网络 ] TPC与UDP协议
  • NTC热敏电阻的原理及应用
  • 8.1 开始新的学习历程
  • 应急响应（windows工具版）
  • Java文件读写I/O操作教程
  • Mysql group by
  • 【C++篇】C++11入门：踏入C++新世界的大门
  • 国内用户如何用手机进行YouTube直播？
  • 『React』 组件通信全攻略
  • 如何从头开始搭建属于自己的家用nas实现内网穿透访问
  • 提升文档管理：推荐一键Docker部署的全文索引搜索引擎工具
  • 如何将联系人从三星手机转移到 iPhone
  • RabbitMQ-镜像队列（Mirrored Queues）
  • 测试平台如何重塑CI/CD流程中的质量协作新范式
  • 什么是CI/CD？
  • 层次聚类：无需“猜”K值，如何让数据自己画出“家族图谱”？
  • HQChart实战教程58：K线主图仿TradingView实现