如果我有例外,我该如何重试?
How can I redo a try If I had an except?
我正在编写一个 scraper,它使用 gspread 在 Google 表格中读写。
在代码的“写入”部分我不得不添加一个try-except
因为写入的配额限制导致APIError
,所以当except被执行它有等待100秒,然后继续。问题是它忽略了导致异常的项目,但它应该重复那个项目
def process_cpf_list(self):
# SKIP OVER COLUMN HEADING IN THE SPREADSHEET
cpfs = self.sheet.col_values(self.cpf_col)[1:]
bot_url = BOT()
for row, cpf in enumerate(cpfs):
nome, idade, beneficio, concessao, salario, bancos, bancocard, consig, card = bot_url.search_cpfs(cpf)
# UPDATE THE SHEET
print("Atualizando...")
try:
row = row + 2
self.sheet.update_cell(row, self.nome_col, nome)
self.sheet.update_cell(row, self.age_col, idade)
self.sheet.update_cell(row, self.beneficio_col, beneficio)
self.sheet.update_cell(row, self.concessao_col, concessao)
self.sheet.update_cell(row, self.salario_col, salario)
self.sheet.update_cell(row, self.bancos_col, bancos)
self.sheet.update_cell(row, self.bancocard_col, bancocard)
self.sheet.update_cell(row, self.consig_col, consig)
self.sheet.update_cell(row, self.card_col, card)
print('Cliente atualizado!')
except APIError:
print('Esperando para atualizar...')
time.sleep(100)
continue
cpf_updater = CpfSearch('TESTE')
cpf_updater.process_cpf_list()
您可以将 Try: 中的内容放到一个单独的函数中。如果该函数抛出错误,它将自动在您的 except: 块中捕获。在那里你可以回忆起所说的功能。
while True:
try:
# do some stuff
break # we didn't hit the exception, exit the loop
except APIError:
# handle the exception...
在这种情况下不应重复该项目。当您使用 for 循环进行迭代时,一旦物品被拾取,它就不会被再次拾取。
您可以添加一个 while 循环,它会无限尝试更新,除非更新成功:
for row, cpf in enumerate(cpfs):
nome, idade, beneficio, concessao, salario, bancos, bancocard, consig, card = bot_url.search_cpfs(cpf)
# UPDATE THE SHEET
print("Atualizando...")
max_retries = 3
row = row + 2
while max_retries:
try:
self.sheet.update_cell(row, self.nome_col, nome)
self.sheet.update_cell(row, self.age_col, idade)
self.sheet.update_cell(row, self.beneficio_col, beneficio)
self.sheet.update_cell(row, self.concessao_col, concessao)
self.sheet.update_cell(row, self.salario_col, salario)
self.sheet.update_cell(row, self.bancos_col, bancos)
self.sheet.update_cell(row, self.bancocard_col, bancocard)
self.sheet.update_cell(row, self.consig_col, consig)
self.sheet.update_cell(row, self.card_col, card)
print('Cliente atualizado!')
break
except APIError:
max_retries -= 1
print('Esperando para atualizar...')
time.sleep(100)
编辑:
我添加了一个简单的重试机制,它会在跳过该行之前尝试 3 次。
你为什么不直接使用 while
循环?
for row, cpf in enumerate(cpfs):
nome, idade, beneficio, concessao, salario, bancos, bancocard, consig, card = bot_url.search_cpfs(cpf)
# UPDATE THE SHEET
print("Atualizando...")
while True:
try:
row = row + 2
self.sheet.update_cell(row, self.nome_col, nome)
self.sheet.update_cell(row, self.age_col, idade)
self.sheet.update_cell(row, self.beneficio_col, beneficio)
self.sheet.update_cell(row, self.concessao_col, concessao)
self.sheet.update_cell(row, self.salario_col, salario)
self.sheet.update_cell(row, self.bancos_col, bancos)
self.sheet.update_cell(row, self.bancocard_col, bancocard)
self.sheet.update_cell(row, self.consig_col, consig)
self.sheet.update_cell(row, self.card_col, card)
print('Cliente atualizado!')
break
except APIError:
print('Esperando para atualizar...')
time.sleep(100)
另一种选择是使用 retrying 库,它可以在方法级别为您处理重试。
这里的一个好处是可以访问许多经过测试的功能,例如指数 back-offs/retrying N 次等...而无需自己编写。
一个缺点是,如果您处于重要的情况下,您将添加一个新的依赖项。它可能比您需要的更多 - 在这种情况下,使用 while 块并永远重试要简单得多。
例如,您可以将不可靠的逻辑移动到单独的方法中,并将其包装在 @retry
注释中,如下所示:
from retrying import retry
def process_cpf_list(self):
...
for row, cpf in enumerate(cpfs):
update_cells(...)
@retry(wait_fixed=100000) # Retry after 100 seconds
def update_cells(...):
self.sheet.update_cell(row, self.nome_col, nome)
etc...
cpf_updater = CpfSearch('TESTE')
cpf_updater.process_cpf_list()
我正在编写一个 scraper,它使用 gspread 在 Google 表格中读写。
在代码的“写入”部分我不得不添加一个try-except
因为写入的配额限制导致APIError
,所以当except被执行它有等待100秒,然后继续。问题是它忽略了导致异常的项目,但它应该重复那个项目
def process_cpf_list(self):
# SKIP OVER COLUMN HEADING IN THE SPREADSHEET
cpfs = self.sheet.col_values(self.cpf_col)[1:]
bot_url = BOT()
for row, cpf in enumerate(cpfs):
nome, idade, beneficio, concessao, salario, bancos, bancocard, consig, card = bot_url.search_cpfs(cpf)
# UPDATE THE SHEET
print("Atualizando...")
try:
row = row + 2
self.sheet.update_cell(row, self.nome_col, nome)
self.sheet.update_cell(row, self.age_col, idade)
self.sheet.update_cell(row, self.beneficio_col, beneficio)
self.sheet.update_cell(row, self.concessao_col, concessao)
self.sheet.update_cell(row, self.salario_col, salario)
self.sheet.update_cell(row, self.bancos_col, bancos)
self.sheet.update_cell(row, self.bancocard_col, bancocard)
self.sheet.update_cell(row, self.consig_col, consig)
self.sheet.update_cell(row, self.card_col, card)
print('Cliente atualizado!')
except APIError:
print('Esperando para atualizar...')
time.sleep(100)
continue
cpf_updater = CpfSearch('TESTE')
cpf_updater.process_cpf_list()
您可以将 Try: 中的内容放到一个单独的函数中。如果该函数抛出错误,它将自动在您的 except: 块中捕获。在那里你可以回忆起所说的功能。
while True:
try:
# do some stuff
break # we didn't hit the exception, exit the loop
except APIError:
# handle the exception...
在这种情况下不应重复该项目。当您使用 for 循环进行迭代时,一旦物品被拾取,它就不会被再次拾取。
您可以添加一个 while 循环,它会无限尝试更新,除非更新成功:
for row, cpf in enumerate(cpfs):
nome, idade, beneficio, concessao, salario, bancos, bancocard, consig, card = bot_url.search_cpfs(cpf)
# UPDATE THE SHEET
print("Atualizando...")
max_retries = 3
row = row + 2
while max_retries:
try:
self.sheet.update_cell(row, self.nome_col, nome)
self.sheet.update_cell(row, self.age_col, idade)
self.sheet.update_cell(row, self.beneficio_col, beneficio)
self.sheet.update_cell(row, self.concessao_col, concessao)
self.sheet.update_cell(row, self.salario_col, salario)
self.sheet.update_cell(row, self.bancos_col, bancos)
self.sheet.update_cell(row, self.bancocard_col, bancocard)
self.sheet.update_cell(row, self.consig_col, consig)
self.sheet.update_cell(row, self.card_col, card)
print('Cliente atualizado!')
break
except APIError:
max_retries -= 1
print('Esperando para atualizar...')
time.sleep(100)
编辑:
我添加了一个简单的重试机制,它会在跳过该行之前尝试 3 次。
你为什么不直接使用 while
循环?
for row, cpf in enumerate(cpfs):
nome, idade, beneficio, concessao, salario, bancos, bancocard, consig, card = bot_url.search_cpfs(cpf)
# UPDATE THE SHEET
print("Atualizando...")
while True:
try:
row = row + 2
self.sheet.update_cell(row, self.nome_col, nome)
self.sheet.update_cell(row, self.age_col, idade)
self.sheet.update_cell(row, self.beneficio_col, beneficio)
self.sheet.update_cell(row, self.concessao_col, concessao)
self.sheet.update_cell(row, self.salario_col, salario)
self.sheet.update_cell(row, self.bancos_col, bancos)
self.sheet.update_cell(row, self.bancocard_col, bancocard)
self.sheet.update_cell(row, self.consig_col, consig)
self.sheet.update_cell(row, self.card_col, card)
print('Cliente atualizado!')
break
except APIError:
print('Esperando para atualizar...')
time.sleep(100)
另一种选择是使用 retrying 库,它可以在方法级别为您处理重试。
这里的一个好处是可以访问许多经过测试的功能,例如指数 back-offs/retrying N 次等...而无需自己编写。
一个缺点是,如果您处于重要的情况下,您将添加一个新的依赖项。它可能比您需要的更多 - 在这种情况下,使用 while 块并永远重试要简单得多。
例如,您可以将不可靠的逻辑移动到单独的方法中,并将其包装在 @retry
注释中,如下所示:
from retrying import retry
def process_cpf_list(self):
...
for row, cpf in enumerate(cpfs):
update_cells(...)
@retry(wait_fixed=100000) # Retry after 100 seconds
def update_cells(...):
self.sheet.update_cell(row, self.nome_col, nome)
etc...
cpf_updater = CpfSearch('TESTE')
cpf_updater.process_cpf_list()