python - 迭代 SQLalchemy 中的抓取列表
问题描述
我正在尝试将部分 JIRA 信息刮到列表中。这可以正常工作。但是在我尝试将此信息添加到我的 SQLalchemy 中之后，它不会添加或提交任何内容。它从何而来？在调试时，我在变量字段中看到了正确的值。
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Date, Boolean, PrimaryKeyConstraint, UniqueConstraint
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.sql import select
from Dataextractie import testplanImport, testExecutionImport, projectversionImport
from jira_ext import JiraExt
import json
from datetime import datetime
#from Databasemodel import Requirements, TestSet, Test, ProjectVersion, TestPlan, Testexecutie, TestRun
jiraServer = JiraExt(snip)
Base = declarative_base()
#class Requirements(Base):
# __tablename__ = "requirements"
#
# ReqID = Column('reqid', Integer, primary_key=True)
# Tags = Column('tags', String)
# Desc = Column('desc', String)
#class TestSet(Base):
# __tablename__ = "testset"
# TestSetID = Column('testsetid', Integer, primary_key=True)
# TestSetName = Column('testsetname', String)
# TestSetDesc = Column('testsetdesc', String)
# Testid = Column('testid', String, ForeignKey('test.testid'))
class ProjectVersion(Base):
__tablename__ = "projectversion"
#__table_args__ = (PrimaryKeyConstraint('productrelease.releaseid', 'productrelease.productid'),)
Projectid = Column('projectid', String, primary_key=True)
Desc = Column('desc', String)
Tags = Column('tags', String)
#ForeignKeyConstraint(['productid'], ['products.productid'])
class TestPlan(Base):
__tablename__ = "testplan"
#__table_args__ = (PrimaryKeyConstraint('testplan.runid', 'testplan.reqid'),)
PlanID = Column('planid', String, primary_key=True)
ProjectID = Column('projectid', String, ForeignKey('projectversion.projectid'))
Naam = Column('naam', String)
#ForeignKeyConstraint(['reqid', 'runid'], ['requirements.reqid', 'testrun.runid'])
class Testexecutie(Base):
__tablename__ = "testexecutie"
#__table_args__ = (PrimaryKeyConstraint('testexecutie.runid', 'testexecutie.testcadeid'),)
Planid = Column('planid', String, ForeignKey("testplan.planid"), primary_key=True)
Testdesc = Column('testdesc', String)
ExecutieID = Column('executieid', String, primary_key=True)
Status = Column('status', String)
__table_args__ = (UniqueConstraint('executieid'),)
class Test(Base):
__tablename__ = "test"
#__table_args__ = (PrimaryKeyConstraint('testcase.testcaseid'),)
TestID = Column('testid', String, primary_key=True)
Casename = Column('casename', String)
Executieid = Column('ExecutieID', String, ForeignKey('testexecutie.executieid'), primary_key=True)
Result = Column('result', String)
#class TestRun(Base):
# __tablename__ = "testrun"
# #__table_args__ = (PrimaryKeyConstraint('runid', 'releaseid'),)
#
# RunID = Column('runid', String, primary_key=True)
# ExecutieID = Column('executieid', String, ForeignKey('testexecutie.executieid'))
# TestID = Column('testid', String, ForeignKey('test.testid'))
# Tags = Column('tags', String)
# Tsttimestart = Column('testtimestart', Date, nullable=False)
# Tsttimeend = Column('testtimeend', Date, nullable=False)
# Testresult = Column('testresult', String, nullable=False)
# Testoutput = Column('testoutput', String)
#ForeignKeyConstraint(['runid', 'releaseid'], ['testexecutie.runid', 'productrelease.releaseid'])
engine = create_engine('snip', echo=True)
Base.metadata.create_all(bind=engine)
Session = sessionmaker(bind=engine)
session = Session()
############################################## Session Open, Klaar voor invoer
Dataextraction = testExecutionImport("10100")
counter = 0
for identifier in Dataextraction[5]:
projectversion = ProjectVersion()
testplan = TestPlan()
projectversion.Projectid = "10100"
testplan.ProjectID = "10100"
testplan.PlanID = identifier
#testplan.Naam = identifier[0]["summary"]
session.add(projectversion, testplan)
counter +=1
session.commit
for identifier in Dataextraction[6]:
counter = 0
testexecutie = Testexecutie()
testexecutie.ExecutieID = identifier
testexecutie.Planid = Dataextraction[5][counter]
session.add(testexecutie)
counter +=1
session.commit
for identifier in Dataextraction[4]:
counter = 0
test = Test()
test.TestID = identifier
test.Executieid = Dataextraction[2][counter]
test.Result = Dataextraction[3][counter]
session.add(test)
counter +=1
session.commit
session.close()
如果它是我的刮刀，这是代码：
from jira_ext import JiraExt
import json
from datetime import datetime
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Date, Boolean, PrimaryKeyConstraint, UniqueConstraint
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from collections import namedtuple
jiraServer = JiraExt(url="Snip")
def testExecutionImport(projectplan):
testplanid = jiraServer.projectGetTestPlans(projectplan)
listtestplanid = []
listtestid = []
liststatus = []
listexecutieid = []
listindvexecutieid = []
listtestplannaam = []
listprojectid = []
for issue in testplanid["issues"]:
print(issue["key"])
listtestplannaam.append(issue["fields"])
listprojectid.append(projectplan)
#listexecutieid.append(issue["id"])
executionsintestplan = jiraServer.testPlanGetAllTestExecutions(issue["key"])
for key in executionsintestplan:
#for i in range(executionsintestplan):
importapi = jiraServer.testExecutionGetTests(key["key"])
currentexecutieid = key["key"]
listindvexecutieid = key["key"]
listtestplanid.append(issue["key"])
for key in importapi:
#testid = jiraServer.testExecutionGetTests(executionsintestplan)
#executieid = jiraServer.testExecutionGetTests(executionsintestplan)
#status = jiraServer.testExecutionGetTests(executionsintestplan)
listtestid.append(key["key"])
listexecutieid.append(currentexecutieid)
liststatus.append(key["status"])
#values["projectplanid"]=projectplan
# values["testid"] = testid
# values["executieid"] = executieid
# values["status"] = status
# values["planid"] = testplanid
return listtestplannaam, listprojectid, listexecutieid, liststatus, listtestid, listtestplanid, listindvexecutieid
解决方案
您实际上并没有在调用session.commit-您只是在引用该方法：
session.commit
.. 应该：
session.commit()