python - 使用 Session.Merg() 时 SQLAlchemy 崩溃脚本
问题描述
我有一个脚本,它使用 amazon api 根据 upc 查找项目,然后将它们保存到 MySql 数据库。由于某种原因,当我使用session.merge
它时,脚本会随机崩溃。使用session.add
有效,但是当数据库中有重复时,提交时的脚本错误。
此外,当脚本崩溃时,它不会输出任何类型的错误。
session.merge 是否有原因使脚本崩溃或 session.add 可以更新 session.commit 上的重复记录的方式?
谢谢,
代码:
upcs = <List of UPC Codes>
# Breaking the list into groups of 10 for Amazon API
group = 10
sub_upcs = [upcs[i:i+group] for i in range(0, len(upcs), group)]
for sub_upc in sub_upcs:
try:
products = amazon.lookup(ItemId=','.join(sub_upc), IdType='UPC', SearchIndex='All')
except AsinNotFound as e:
print("UPC Not Found")
except Exception as e:
print(e)
if type(products) is list:
for product in products:
am_item = Item(product)
session.merge(am_item.db_item)
else:
am_item = Item(products)
session.merge(am_item.db_item)
try:
session.commit()
print("Session Written")
except Exception as e:
print(e)
物品类别:
class AmazonItem(Base):
__tablename__ = 'amazon_products'
timestamp = Column(TIMESTAMP)
itemid = Column(String, primary_key=True, index=True, unique=True)
name = Column(String)
upc = Column(String, index=True)
msrp = Column(Float)
price = Column(Float)
brandname = Column(String)
modelnumber = Column(String)
url = Column(String)
def __init__(self, item):
self.timestamp = datetime.datetime.fromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S')
self.itemid = item.asin
self.name = item.title
self.upc = item.upc
self.msrp = item.price_and_currency[0]
self.price = item.price_and_currency[0]
self.brandname = ""
self.modelnumber = ""
self.url = item.offer_url
class Item():
item = None
db_item = None
store = ""
itemid = ""
name = ""
upc = ""
msrp = 0
price = 0
brandname = ""
modelnumber = ""
def __init__(self, item):
self.timestamp = datetime.datetime.fromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S')
self.item = item
self.db_item = AmazonItem(item)
self.store = "Amazon"
self.itemid = item.asin
self.name = item.title
self.upc = item.upc
self.msrp = item.price_and_currency[0]
self.price = item.price_and_currency[0]
self.brandname = ""
self.modelnumber = ""
self.url = item.offer_url
self.asin = item.asin
if self.msrp == 0 or self.msrp is None:
self.msrp = self.price
if self.price == 0 or self.price is None:
self.price = self.msrp
def __eq__(self, other):
return self.itemid == other.itemid
def __hash__(self):
return hash(('itemid', self.itemid))
def writetocsv(self, file):
csv_writer = CSVWriter(file)
csv_writer.addrow(self.getrow())
csv_writer.write()
解决方案
推荐阅读
- aem-touch-ui - 隐藏/显示功能不适用于 AEM 中的文件上传资源类型
- mockito - 使用 mockito 测试异步 Vertx 和 RxJava 方法
- spring-boot - 带有会话范围 bean 的 uaa 应用程序的 Jhipster
- android - 版本冲突错误而不添加错误显示的任何依赖项?
- javascript - 使用蒙特卡罗模拟限制计算 π
- wordpress - 如何过滤与当前帖子具有相同分类术语的自定义帖子?
- google-app-engine - Google App Engine Standard Java 8 - 如何为实例添加健康检查?
- mysql - SQL查询多对多关系
- vb.net - 一次替换多个字符串
- python - Python - 如何使用 Open-cv 或 PIL 将 24 位 PNG 图像转换为 32 位