1. 监测数据采集物联网应用开发步骤(4)

Sqlite3数据库读写操作开发、异常信息统一处理类开发

本章节需要调用sqlite3及mysql-connector

安装sqlite3

Pip3 install sqlite3

安装mysql-connector

pip3 install mysql-connector

验证是否安装成功，python中运行下列代码无异常则安装成功：

import sqlite3
import mysql.connector

在项目开发过程中各类异常信息try...except处理，该处理类将try ...except异常信息统一处理并日志文件输出，便于快速定位错误信息及代码位置；

异常信息日志打印格式

yyyy-MM-dd H:mm:ss=>异常的类名=>该类中调用出错的函数名==>其他异常信息

参考如下图：

创建异常信息统一处理类com.zxy.z_debug.py

调用方式参考：

from com.zxy.z_debug import z_debug
class ClassName(z_debug):

com.zxy.z_debug.py类

#! python3
# -*- coding: utf-8 -
'''
Created on 2023年08月28日
@author: zxyong 13738196011
'''

import inspect,datetime
from com.zxy.adminlog.UsAdmin_Log import UsAdmin_Log
from com.zxy.common import Com_Para

#监测数据采集物联网应用--异常信息统一处理
class z_debug(object):

    def __init__(self):
        pass
    
    def debug_in(self,inputMsg):
        excStr = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")+"=>"+str(self)+"=>"+self.__class__.__name__+"."+inspect.stack()[1][3]+"==>"+inputMsg
        uL = UsAdmin_Log(Com_Para.ApplicationPath, excStr)
        uL.SaveFileDaySub("exception")
        print(excStr)
    
    def debug_info(self):
        excStr = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")+"=>"+str(self)+"=>"+self.__class__.__name__+"."+inspect.stack()[1][3]+"==>"
        uL = UsAdmin_Log(Com_Para.ApplicationPath, excStr)
        uL.SaveFileDaySub("exception_debug")
        return excStr

在多个线程中使用访问SQLite数据库，为避免多个线程之间的竞争。需要添加数据库线程锁机制以确保线程安全，若使用mysql数据库无需数据库线程锁机制。

在com.zxy.common.Com_Para.py中添加如下内容

#数据库线程锁机制
Dblock1 = threading.Lock()
Dblock2 = threading.Lock()
Dblock3 = threading.Lock()
Dblock4 = threading.Lock()
#静态数据库配置文件路径
ACTIONPATH = "dbconfig.properties"
driverClassName = ""
url = ""
username = ""
password = ""
#是否桌面数据库 0:Mysql 1:sqlite3
IL_DB = 1
#数据库连接池初始数量 1:Sqlite3
incrementalConnections = 2
#数据库连接池最大数量
maxConnections = 10
port = -1
UserDebug = "0"

创建静态配置库数据库配置文件读写类com.zxy.common.DbConfigSelf.py

#! python3
# -*- coding: utf-8 -
'''
Created on 2023年08月28日
@author: zxyong 13738196011
'''

import configparser
from com.zxy.common import Com_Para
from com.zxy.z_debug import z_debug

#监测数据采集物联网应用--静态配置库数据库配置文件读写
class DbConfigSelf(z_debug):
    
    def __init__(self):
        pass
    
    @staticmethod
    def GetDbConfigSelfNew():
        temPath = Com_Para.ApplicationPath + Com_Para.zxyPath + Com_Para.ACTIONPATH
        temProp = configparser.ConfigParser()
        try:
            temProp.read(temPath)
            Com_Para.driverClassName = temProp.get("DB", "dataSource.driverClassName")            
            if Com_Para.driverClassName == "org.sqlite.JDBC" :
                Com_Para.incrementalConnections = 1
                Com_Para.maxConnections = 1
                Com_Para.IL_DB = 1
                temStrRem = temProp.get("DB","dataSource.url").replace("@",Com_Para.ApplicationPath + Com_Para.zxyPath)
                Com_Para.url = temStrRem
            else:
                Com_Para.url = temProp.get("DB", "dataSource.url")
            Com_Para.username = temProp.get("DB", "dataSource.username")
            Com_Para.password = temProp.get("DB", "dataSource.password")
            Com_Para.port = temProp.getint("DB", "server.port")
            Com_Para.UserDebug = temProp.getint("DB", "userDebug")
        except Exception as e:
            print("GetDbConfigSelfNew:"+repr(e)+"=>"+str(e.__traceback__.tb_lineno))
        finally:
            Pass

创建静态配置库数据库配置文件dbconfig.properties

配置文件内容：

[DB]
dataSource.driverClassName=org.sqlite.JDBC
dataSource.url=@center_data.db
dataSource.username=
dataSource.password=

server.port=9000
userDebug=0

1. 监测数据采集物联网应用开发步骤(5.2)