Skip to content

10. 练习 [IMPOTS] 与 MySQL

10.1. 将文本文件导入表中 MySQL

接下来的脚本将导入以下文本文件中的数据：

12620:13190:15640:24740:31810:39970:48360:55790:92970:127860:151250:172040:195000:0
0:0.05:0.1:0.15:0.2:0.25:0.3:0.35:0.4:0.45:0.5:0.55:0.6:0.65
0:631:1290.5:272.5:3309.5:4900:6898.5:9316.5:12106:16754.5:23147.5:30710:39312:49062

导入到以下数据库的 [impots] 表中：

 

将使用用户名 (root,"") 连接到数据库 [dbimpots]。

程序（impotstxt2mysql）

我们将采用以下架构：

我们将编写的脚本 [console] 将使用类 [ImpotsFile] 来访问文本文件中的数据。对数据库的写入操作将使用之前介绍的方法。

脚本代码如下：


# -*- coding=utf-8 -*-

# 导入“税种”类模块
from impots import *
import re

# --------------------------------------------------------------------------
def copyToMysql(limites,coeffR,coeffN,HOTE,USER,PWD,BASE,TABLE):
    # 复制 3 个数值限额表：coeffR、coeffN
    # 到 mysql 数据库 TABLE 表中 BASE
    # 该 MySQL 数据库位于 HOTE 服务器上
    # 用户由 USER 和 PWD 标识

    # 将查询 SQL 放入列表中
    # 删除该表
    requetes=["drop table %s" % (TABLE)]
    # 创建表
    requete="create table %s (limites decimal(10,2), coeffR decimal(6,2), coeffN decimal(10,2))" % (TABLE)
    requetes.append(requete)
    # 数据填充
    for i in range(len(limites)):
        # 插入查询
        requetes.append("insert into %s (limites,coeffR,coeffN) values (%s,%s,%s)" % (TABLE,limites[i],coeffR[i],coeffN[i]))
    # 执行命令 SQL
    return executerCommandes(HOTE,USER,PWD,BASE,requetes,False,False)
    

def executerCommandes(HOTE,ID,PWD,BASE,requetes,suivi=False,arret=True):
    # 使用连接 (HOTE,ID,PWD,BASE)
    # 通过该连接执行请求列表中包含的 SQL 命令
    # 如果跟踪=True，则每次执行 SQL 命令时，都会显示其成功或失败的状态
    # 如果停止=False，则函数在遇到第一个错误时停止；否则执行所有 SQL 命令
    # 该函数返回一个列表（错误数量、错误1、错误2、...）

    # 连接
    try:
        connexion=MySQLdb.connect(host=HOTE,user=ID,passwd=PWD,db=BASE)
    except MySQLdb.OperationalError,erreur:
        return [1,"Erreur lors de la connexion a MySQL sous l'identite (%s,%s,%s,%s) : %s" % (HOTE, ID, PWD, BASE, erreur)]
      
    # 请求游标
    curseur=connexion.cursor()
    # 执行 SQL 列表中包含的查询
    # 正在执行它们——起初没有错误
    erreurs=[0]
    for i in range(len(requetes)):
        # 将查询放入局部变量
        requete=requetes[i]
        # 是否为空查询？
        if re.match(r"^\s*$",requete):
            continue
        # 执行查询 i
        erreur=""
        try:
            curseur.execute(requete)
        except Exception, erreur:
            pass
        # 是否发生错误？
        if erreur:
            # 又一个错误
            erreurs[0]+=1
            # 错误消息
            msg="%s : Erreur (%s)" % (requete[0:len(requete)-1],erreur)
            erreurs.append(msg)
            # 是否显示屏幕？
            if suivi:
                print msg
            # 停止吗？
            if arret:
                return erreurs
        else:
            if suivi: 
                # 显示不带换行符的查询
                print "%s : Execution reussie" % (cutNewLineChar(requete))
                # 关于已执行查询结果的信息
                afficherInfos(curseur)
    # 关闭连接
    try:
        connexion.commit()
        connexion.close()
    except MySQLdb.OperationalError,erreur:
        # 又一个错误
        erreurs[0]+=1
        # 错误消息
        msg="%s : Erreur (%s)" % (requete,erreur)
        erreurs.append(msg)

    # 返回
    return erreurs


# ------------------------------------------------ 主程序
# 用户身份
USER="root"
PWD=""
# DBMS的主机
HOTE="localhost"
# 数据库标识
BASE="dbimpots"
# 数据表标识
TABLE="impots"
# 数据文件
IMPOTS="impots.txt"

# [dao] 层实例
try:
    dao=ImpotsFile(IMPOTS)
except (IOError, ImpotsError) as infos:
    print ("Une erreur s'est produite : {0}".format(infos))
    sys.exit()

# 将检索到的数据导入 MySQL 表
erreurs=copyToMysql(dao.limites,dao.coeffR,dao.coeffN,HOTE,USER,PWD,BASE,TABLE)
if erreurs[0]:
    for i in range(1,len(erreurs)):
        print erreurs[i]
else:
    print "Transfert opere"
# 结束
sys.exit()

注：

  • 第 106-110 行：实例化第 8.1 节中介绍的 [ImpotsFile] 类；
  • 第 113 行：将数组 limites、coeffR、coeffN 传输到数据库 MySQL 中；
  • 第8行：函数copyToMysql执行此转移。函数copyToMysql创建待执行的查询表，并由函数executerCommandes（第25行）执行这些查询；
  • 第28-89行： ，函数executerCommandes即前文第9.7节中已介绍过的函数，但有一处不同：查询不再位于文本文件中，而是位于列表中；
结果

文本文件 impots.txt：

12620:13190:15640:24740:31810:39970:48360:55790:92970:127860:151250:172040:195000:0
0:0.05:0.1:0.15:0.2:0.25:0.3:0.35:0.4:0.45:0.5:0.55:0.6:0.65
0:631:1290.5:272.5:3309.5:4900:6898.5:9316.5:12106:16754.5:23147.5:30710:39312:49062

屏幕结果：

Transfert opéré

使用 phpMyAdmin 进行验证：

 

10.2. 税款计算程序

现在，计算税款所需的数据已存入数据库，我们可以编写税款计算脚本了。我们再次采用三层架构：

新的 [dao] 层将连接到 SGBD 和 MySQL，并由类 [ImpotsMySQL] 实现。 它将为 [metier] 层提供与之前相同的接口，该接口由唯一的方法 getData 组成，该方法返回元组 (limites, coeffR, coeffN)。 因此，[metier] 层相对于上一版本不会发生变化。

10.3. 类 [ImpotsMySQL]

[dao] 层现由以下 [ImpotsMySQL] 类实现（文件 impots.py）：


class ImpotsMySQL:

    # 构造函数
    def __init__(self,HOTE,USER,PWD,BASE,TABLE):
        # 初始化边界属性，coeffR，coeffN
        # 用于计算税款所需的数据已放入表 mysqL TABLE
        # 属于数据库 BASE。该表的结构如下
        # 范围 decimal(10,2)，coeffR decimal(10,2)，coeffN decimal(10,2)
        # 通过 (USER,PWD) 身份连接到 HOTE 机器上的 MySQL 数据库
        # 若发生错误则抛出异常
        
        # 连接 MySQL 数据库
        connexion=MySQLdb.connect(host=HOTE,user=USER,passwd=PWD,db=BASE)
        # 请求游标
        curseur=connexion.cursor()
        
        # 批量读取表 TABLE
        requete="select limites,coeffR,coeffN from %s" % (TABLE)
        # 在连接 [connexion] 的数据库 [base] 上执行查询 [requete]
        curseur.execute(requete)
        # 处理查询结果
        ligne=curseur.fetchone()
        self.limites=[]
        self.coeffR=[]
        self.coeffN=[]
        while(ligne):
            # 当前行
            self.limites.append(ligne[0])
            self.coeffR.append(ligne[1])
            self.coeffN.append(ligne[2])
            # 下一行
            ligne=curseur.fetchone()
        # 注销
        connexion.close()

    def getData(self):
        return (self.limites, self.coeffR, self.coeffN)

注：

  • 第 18 行：查询 SQL SELECT，该查询请求 MySQL 数据库中的数据。 随后，SELECT的查询结果行由[curseur.fetchone]逐行处理（第22行和第32行），以生成表格limites、coeffR、 coeffN（第28-30行）；
  • [dao] 层接口中的 getData 方法。

10.4. 控制台脚本

控制台脚本（impots_04）的代码如下：


# -*- coding=utf-8 -*-

# 导入 Impots* 类模块
from impots import *

# ------------------------------------------------ 主程序
# 用户身份
USER="root"
PWD=""
# 数据库管理系统的主机
HOTE="localhost"
# 数据库标识
BASE="dbimpots"
# 数据表标识
TABLE="impots"
# 输入文件
DATA="data.txt"
# 输出文件
RESULTATS="resultats.txt"

# [metier] 层实例
try:
    metier=ImpotsMetier(ImpotsMySQL(HOTE,USER,PWD,BASE,TABLE))
except (IOError, ImpotsError) as infos:
    print ("Une erreur s'est produite : {0}".format(infos))
    sys.exit()

# 计算税款所需的数据已放入文件 IMPOTS
# 每张表格占一行，格式为
# val1:val2:val3,...

# 读取数据
try:
    data=open(DATA,"r")
except:
    print "Impossible d'ouvrir en lecture le fichier des donnees [DATA]"
    sys.exit()

# 打开结果文件
try:
    resultats=open(RESULTATS,"w")
except:  
    print "Impossible de creer le fichier des résultats [RESULTATS]"
    sys.exit()

# 实用工具
u=Utilitaires()

# 处理数据文件中的当前行
ligne=data.readline()
while(ligne != ''):
    # 移除可能存在的换行符
    ligne=u.cutNewLineChar(ligne)
    # 提取构成该行的 3 个字段：已婚:子女:工资
    (marie,enfants,salaire)=ligne.split(",")
    enfants=int(enfants)
    salaire=int(salaire)
    # 计算税额
    impot=metier.calculer(marie,enfants,salaire)
    # 写入计算结果
    resultats.write("{0}:{1}:{2}:{3}\n".format(marie,enfants,salaire,impot))
    # 读取新行
    ligne=data.readline()
# 关闭文件
data.close()
resultats.close()

注：

  • 第 23 行：实例化 [dao] 和 [metier] 层；
  • 其余代码已知。
结果

与本练习的前几个版本结果相同。

https://stahe.github.io
本教程由 Serge Tahé 编写，根据 知识共享 署名-非商业性使用- ShareAlike 3.0 Unported.