Python脚本两次写入表

Question

我正在尝试使这种情况自动化。我有 2 个 .sql 文件（add1.sql 和 add2.sql），每个文件都有 1 个插入脚本。

我的目标是通过执行 add1.sql 中的行将一条记录写入 table1，并通过执行 add2.sql 中的行将一条记录写入 cm.cl，等待大约 5 分钟，这样后端服务就会运行。该服务从 DB1 写入到 DB2。然后我连接到 DB2 以查看来自 DB1 的记录是否与写入到 DB2 的记录相匹配。根据没有结果，会发送一封电子邮件。

下面是我的代码。除了两次写入DB1 之外，一切正常。所以，基本上插入了 4 条记录而不是 2 条。知道为什么它会写入 4 条记录吗？

import pypyodbc as pyodbc
import smtplib
sender = 'abc@abc.com'
receivers = ['abc@abc.com','xyz@abc.com']
import unittest
import time

class TestDB1(unittest.TestCase):


    def testing_master(self):

           Master_Conn = 'Driver=
 {SQLServer};Server=server\servername;Database=database;UID=userid;PWD=password'
           Master_db = pyodbc.connect(Master_Conn)
           Master_Cursor = Master_db.cursor()
           try:
               #Open, read and execute add_shell.sql
               file = open('C:\\aaa\\add1.sql', 'r')
               line = file.read()
               lines = line.replace('\n', ' ')
               file1 = open('C:\\aaa\\add2.sql', 'r')
               line1=file1.read()
               lines1=line1.replace('\n', ' ')
               Master_Cursor.execute(lines)
               time.sleep(1)
               Master_Cursor.execute(lines1)
               Master_db.commit()
               file.close()
               file1.close()

               #Get python object for latest record inserted in DB1
               Master_CID=Master_Cursor.execute("select col1 from tablename1 order by sequenceid desc").fetchone()
               #convert tuple to srting [0] gives first tuple element. 
               Master_CID_str=str(Master_CID[0])
               #Get GUID by stripping first 2 chars and last char.
               Master_CID_str=Master_CID_str[2:len(Master_CID_str)-1]
               Master_CLID=Master_Cursor.execute("select col2 from tablename2 order by sequenceid desc").fetchone()
               Master_CLID_str=str(Master_CLID[0])
               Master_CLID_str=Master_CLID_str[2:len(Master_CLID_str) - 1]



           # Wait for service that transfers data from one db to another DB to run
               time.sleep(310)
           finally:
                Master_Cursor.close()
                Master_db.close()

           return Master_CID,Master_CID_str,Master_CLID,Master_CLID_str


    def testing_int_instance(self):
           #unpacking return value of tuple from testing_master() function
           Master_CID,Master_CID_str,Master_CLID,Master_CLID_str=self.testing_master()
           print ("printing from testing_int_instance {0}".format(Master_CID))
           Int_Instance_Conn = 'Driver={SQL Server};Server=server2\servername2;Database=database2;UID=uid;PWD=password;'
           Int_db = pyodbc.connect(Int_Instance_Conn)
           Int_Cursor = Int_db.cursor()
           #return Int_db, Int_Cursor
           #execute select from  db where col matches that of one inserted in master db.
           Int_Instance_CID=Int_Cursor.execute("select col1 from table1 where cartridgemodelid = '%s'" %(Master_CID_str)).fetchone()
           print(Int_Instance_CID)
           smtpObj = smtplib.SMTP('22.101.1.333', 25)
           if (Master_CID==Int_Instance_CID):
               print("Matched")
               content="This email confirms successful  data transfer from Master to  Instance for col1: \n"
               message = "\r\n".join(["From:" + sender,"To:" + str(receivers[:]),"Subject: Test Result","",content +Master_CID_str])
               #smtpObj = smtplib.SMTP('22.101.2.222', 25)
               smtpObj.sendmail(sender, receivers, message)
           elif (Master_CID!=Int_Instance_CID):
               print("no match")
               content = "This email confirms failure of  data transfer from DB1 to DB2 for COL1: \n"
               message = "\r\n".join(["From:" + sender, "To:" + str(receivers[:]), "Subject: Test Result", "",content +Master_CID_str])
               smtpObj.sendmail(sender, receivers, message)
           Int_Instance_CLID=Int_Cursor.execute("select COL2 from table2 where col= '%s'" %(Master_CLID_str)).fetchone()
           print (Int_Instance_CLID)
           if (Master_CLID == Int_Instance_CLID):
               print ("Printing int_instance CLID {0}".format(Int_Instance_CLID))
               content = "This email confirms successful data transfer from DB1 to DB2 for COL: \n"
               message = "\r\n".join(
                  ["From:" + sender, "To:" + str(receivers[:]), "Subject: Test Result", "", content + Master_CLID_str])
               #smtpObj = smtplib.SMTP('22.101.2.222', 25)
               smtpObj.sendmail(sender, receivers, message)
               print ("Ids Matched")
           elif (Master_CLID != Int_Instance_CLID):
DB1 to DB2 for COL: \n"
               message = "\r\n".join(
                  ["From:" + sender, "To:" + str(receivers[:]), "Subject: Test Result", "", content + Master_CLID_str])
               #smtpObj = smtplib.SMTP('22.101.2.222', 25)
               smtpObj.sendmail(sender, receivers, message)
           smtpObj.quit()


           Int_db.close()

如果 name == 'ma​​in'： unittest.main()

add1.sql 是：

DECLARE @Name VARCHAR(2000)
DECLARE @PartNumber VARCHAR(2000)
SELECT @Name='test'+convert(varchar,getdate(),108)
SELECT @PartNumber='17_00001_'+convert(varchar,getdate(),108)
DECLARE @XML XML
DECLARE @FileName VARCHAR(1000)
DECLARE @Id UNIQUEIDENTIFIER
SELECT @Id = NEWID()
SELECT @FileName = 'test.xml'
SELECT @XML='<model>
   <xml tags go here>
BEGIN
  INSERT INTO table1
         (ID,Name,Type,Desc,Number,Revision,Model,status,Modifiedby,Modifiedon)
   VALUES(@Id,@Name,'xyz','',@partnumber,'01',@XML,'A','453454-4545-4545-4543-345342343',GETUTCDATE())

add2.sql 是：

DECLARE @XML XML
DECLARE @CM_Name VARCHAR(2000)
DECLARE @FileName VARCHAR(1000)
DECLARE @PartNumber VARCHAR(2000)
DECLARE @Id UNIQUEIDENTIFIER
SELECT @Id=NEWID()
DECLARE @Name VARCHAR(2000)
DECLARE @CMId VARCHAR(2000)
DECLARE @CM_PartName VARCHAR(2000)
DECLARE @CM_Partnumber VARCHAR(2000)
SELECT @Name='test'+convert(varchar,getdate(),108)
SELECT @PartNumber='test'+convert(varchar,getdate(),108)
DECLARE @RowCount INT
DECLARE @Message VARCHAR(100);

SELECT @FileName = 'test.xml' 
SELECT @CMId = CM.CMID,
    @CM_Name = CM.CMName,
    @CM_PN  = CM.PN
    FROM cm.Model CM
    WHERE CM.MName LIKE 'test%'
    ORDER BY CM.ModifiedBy DESC 
SELECT @XML='<Layout>
   other xml tags...

BEGIN
  INSERT INTO cm.CL(ID, ModelID, Layout, Description, PN, Revision, CLayout, Status, ModifiedBy, ModifiedOn)
    SELECT TOP 1 @Id, @CMId, @Name, '', @PartNumber, '01', @XML, 'A', '453454-345-4534-4534-4534543545', GETUTCDATE()
    FROM cm.table1 CM
    WHERE CM.Name=@CM_Name
    AND CM.Partnumber=@CM_Partnumber

Answer 1

目前，您正在呼叫test_master() 两次！首先作为您的命名方法，然后在您解压缩返回值时使用第二种方法。下面是在 Class 对象之外定义的方法的演示。如果按原样调用，testing_master 将运行两次。

考虑使用上下文管理器来读取使用with() 的 .sql 脚本，该脚本处理如下所示的打开和关闭 i/o 操作：

# FIRST CALL
def testing_master():
    #...SAME CODE...
    try:
        with open('C:\\aaa\\add1.sql', 'r') as file:
            lines = file.read().replace('\n', ' ')    
        Master_Cursor.execute(lines)
        Master_db.commit()

        time.sleep(1)

        with open('C:\\aaa\\add2.sql', 'r') as file1:
            lines1 = file1.read().replace('\n', ' ')    
        Master_Cursor.execute(lines1)
        Master_db.commit()

    #...SAME CODE...  
    return Master_CID, Master_CID_str, Master_CLID, Master_CLID_str


def testing_int_instance():
    # SECOND CALL
    Master_CID, Master_CID_str, Master_CLID, Master_CLID_str = testing_master()    
    #...SAME CODE...

if __name__ == "__main__": 
    testing_master()
    testing_int_instance()

注释掉 time(310) 似乎可行，但正如您提到的后台 Windows 服务无法有效运行，因此会中断数据库传输。

要解决此问题，请考虑在第一个方法的末尾调用第二个方法，将值作为参数传递，不带任何return 并删除解包行。然后，在主全局环境中，只运行testing_master()。当然，在 Class 定义中使用 self 进行限定。

def testing_master():
    #...SAME CODE...
    testing_int_instance(Master_CID, Master_CID_str, Master_CLID, Master_CLID_str)

def testing_int_instance(Master_CID, Master_CID_str, Master_CLID, Master_CLID_str):
    #...SKIP UNPACK LINE 
    #...CONTINUE WITH SAME CODE...

if __name__ == "__main__": 
    testing_master()

由于您的unittest，请考虑对原始设置稍作调整，您可以使用self 限定每个变量：

def testing_master():
    ...
    self.Master_CID=Master_Cursor.execute("select col1 from tablename1 order by sequenceid desc").fetchone()
    self.Master_CID_str=str(Master_CID[0])
    self.Master_CID_str=Master_CID_str[2:len(Master_CID_str)-1]
    self.Master_CLID=Master_Cursor.execute("select col2 from tablename2 order by sequenceid desc").fetchone()
    self.Master_CLID_str=str(Master_CLID[0])
    self.Master_CLID_str=Master_CLID_str[2:len(Master_CLID_str) - 1]

def testing_int_instance(self):
    # NO UNPACK LINE
    # ADD self. TO EVERY Master_* VARIABLE
    ...