SQLAlchemy 框架

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

使用 SQLAlchemy 来访问达梦数据库

SQLAlchemy 使用方言 Dialect 与各种类型的数据库进行通信,每种数据库都有对应的 DBAPI,所有方言都要求安装对应的 DBAPI 驱动程序。SQLAlchemy 根据配置的方言 Dialect 不同调用不同的数据库 DBAPI,从而实现对数据库的操作。

达梦数据库的方言是 sqlalchemy_dm,使用 dmPython 作为默认的 DBAPI。因此使用 SQLAlchemy 访问达梦数据库时需要先安装 dmPython 和 sqlalchemy_dm,此外还需要安装 SQLAlchemy。

安装完 DM 数据库软件后,在安装路径下的 drivers 目录下的 python 目录中,可以找到 dmPython 和sqlalchemy 方言的驱动源码,由于提供的是源码,需要自己编译安装,详细的编译安装方法参考 Python应用程序开发

开发步骤:

安装 SQLAlchemy

pip install SQLAlchemy==1.3.23

安装SQLAlchemy

注意

注意:安装 SQLAlchemy 时需指定版本,SQLAlchemy1.4 版本及以上暂不支持。

编译安装 dmPython 驱动

cd D:\dmdbms\drivers\python\dmPython
python setup.py install

编译安装

编译安装达梦数据库的 sqlalchemy 方言

cd D:\dmdbms\drivers\python\sqlalchemy
python setup.py install

方言

使用 pip list 命令查看是否安装成功

安装成功

查看搜索路径并将 dpi 目录文件拷贝到搜索路径下

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

注意

Windows 环境需要执行此操作,Linux 环境设置 LD_LIBRARY_PATH 环境变量即可。

进入 python 解释器查看搜索路径

python
import sys
sys.path

查看搜索路径1

C:\\Users\\Admin\\AppData\\Local\\Programs\\Python\\Python39\\lib\\site-packages\\dmpython-2.3-py3.9-win-amd64.egg

查看搜索路径2

基本操作示例

该代码实现增、删、改、查四个基本操作。

先在数据库中创建 PRODUCT 表,然后创建 Product.py 文件和 dm_ sqlalchemy.py 文件

登录数据库创建表


CREATE TABLE "PRODUCT"
(
"PRODUCTID" INT IDENTITY(1, 1) PRIMARY KEY,
"NAME" VARCHAR(100) NOT NULL,
"AUTHOR" VARCHAR(25) NOT NULL,
"PUBLISHER" VARCHAR(50) NOT NULL,
"PUBLISHTIME" DATE NOT NULL,
"PRODUCT_SUBCATEGORYID" INT,
"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,
"PHOTO" IMAGE,
"TYPE" VARCHAR(5),
"PAPERTOTAL" INT,
"WORDTOTAL" INT,
"SELLSTARTTIME" DATE NOT NULL,
"SELLENDTIME" DATE
);

创建 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: SYSDBA123@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):
    # 创建Session
    session = 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):
    # 创建Session
    session = 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):
    # 创建Session
    session = DBSession()
    product = session.query(Product).filter(Product.NAME == '水浒传').one()
    product.NAME = '水浒'
    session.commit()
    print('更新成功')
    session.close()

def fun_delete(DBSession):
    # 创建Session
    session = DBSession()
    session.query(Product).filter(Product.NAME == '水浒').delete()
    session.commit()
    print('删除成功')
    session.close()

if __name__ == '__main__':
main()

运行结果如下:

运行结果

微信扫码
分享文档
扫一扫
联系客服