完成我的场景后如何删除 table？
How can i drop table after finish my scenario?
完成我的方案后，我遇到了删除 table 的问题
当我尝试使用 db.drop_table(Routes, if_exists=False, with_all_data=False) or db.drop_table(Routes, if_exists=True, with_all_data=True) 删除时
我得到一个错误
pony.orm.core.TransactionError: @db_session-decorated drop_table() 带有 ddl 选项的函数不能在另一个 db_sessio 内部调用
我尝试使用 Routes.delete() 删除，但我得到了相同的
我阅读了更多文档，但我不知道如何删除此 table
我的完整代码
import logging
import random
import requests
import vk_api
from pony.orm import db_session, rollback
from vk_api.bot_longpoll import VkBotLongPoll, VkBotEventType
import handlers
from models import UserState, Registration, Routes, db
try:
import settings
except ImportError:
exit('DO CP settings.py.default setting.py and set token')
log = logging.getLogger('bot')
def configure_logging():
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(logging.Formatter('%(levelname)s %(message)s'))
stream_handler.setLevel(logging.INFO)
log.addHandler(stream_handler)
file_handler = logging.FileHandler('bot.log')
file_handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s'))
file_handler.setLevel(logging.DEBUG)
log.addHandler(file_handler)
class Bot:
"""
Echo bot for vk.com
Use python 3.7
"""
def __init__(self, group_id, token):
"""
:param group_id:group_id from group vk.com
:param token:secret key
"""
self.group_id = group_id
self.token = token
self.vk = vk_api.VkApi(token=self.token)
self.long_poller = VkBotLongPoll(self.vk, self.group_id)
self.api = self.vk.get_api()
def run(self):
"""run bot"""
for event in self.long_poller.listen():
try:
print(event)
self.on_event(event)
except Exception:
log.exception('ошибка в обработке события')
@db_session
def on_event(self, event):
"""
sends the message back, if message text
:param event: VKBotMessageEvent object
:return: None
"""
if event.type != VkBotEventType.MESSAGE_NEW:
log.info('пока не умеем работать с таким типом события %s', event.type)
return
user_id = event.message.peer_id
text = event.message.text
state = UserState.get(user_id=str(user_id))
if state is not None:
self.continue_scenario(text=text, state=state, user_id=user_id)
else:
# search intent, the user is found outside of the scenario
for intent in settings.INTENTS:
log.debug(f'User gets {intent}')
if any(token in text.lower() for token in intent['tokens']):
# run intent
if intent['answer']:
self.send_text(text_to_send=intent['answer'], user_id=user_id)
else:
self.start_scenario(scenario_name=intent['scenario'], user_id=user_id, text=text)
break
else:
self.send_text(text_to_send=settings.DEFAULT_ANSWER, user_id=user_id)
# функция отправки текста
def send_text(self, text_to_send, user_id):
self.api.messages.send(
message=text_to_send,
random_id=random.randint(0, 2 ** 20),
peer_id=user_id)
# функция отправки изображения
def send_image(self, image, user_id):
upload_url = self.api.photos.getMessagesUploadServer()['upload_url']
upload_data = requests.post(url=upload_url, files={'photo': ('image.png', image, 'image/png')}).json()
image_data = self.api.photos.saveMessagesPhoto(**upload_data)
owner_id = image_data[0]['owner_id']
media_id = image_data[0]['id']
attachment = f'photo{owner_id}_{media_id}'
self.api.messages.send(
attachment=attachment,
random_id=random.randint(0, 2 ** 20),
peer_id=user_id)
# функция отправки маршрутов
def send_routes(self, text_to_send, user_id):
self.api.messages.send(
message=text_to_send,
random_id=random.randint(0, 2 ** 20),
peer_id=user_id)
# шаг отправки
def send_step(self, step, user_id, text, context):
if 'text' in step:
self.send_text(text_to_send=step['text'].format(**context), user_id=user_id)
if 'image' in step:
handler = getattr(handlers, step['image'])
image = handler(text=text, context=context)
self.send_image(image=image, user_id=user_id)
if 'choice' in step:
handler = getattr(handlers, step['choice'])
choices = handler(text=text, context=context)
for i, choice in enumerate(choices, start=1):
city_from = context['city_from']
city_to = context['city_to']
flight_time = choice[city_from][city_to][0]
flight_board = choice[city_from][city_to][1]
flight_date = choice[city_from][city_to][2]
number_choice = str(i) + ') '
Routes(number_choice=number_choice, flight_time=flight_time, city_from=city_from, city_to=city_to,
flight_board=flight_board, flight_date=flight_date)
select = db.select('number_choice,flight_time,city_from,city_to,flight_board,flight_date FROM Routes')
text_to_send = ' '.join(select[0]) + '\n' + ' '.join(select[1]) + '\n' + ' '.join(
select[2]) + '\n' + ' '.join(
select[3]) + '\n' + ' '.join(select[4]) + '\n'
self.send_routes(text_to_send=text_to_send, user_id=user_id)
# начало сценария
def start_scenario(self, scenario_name, user_id, text):
scenario = settings.SCENARIOS[scenario_name]
first_step = scenario['first_step']
step = scenario['steps'][first_step]
self.send_step(step=step, user_id=user_id, text=text, context={})
UserState(user_id=str(user_id), scenario_name=scenario_name, step_name=first_step, context={})
# конец сценария
def continue_scenario(self, text, state, user_id):
steps = settings.SCENARIOS[state.scenario_name]['steps']
step = steps[state.step_name]
handler = getattr(handlers, step['handler'])
if handler(text=text, context=state.context):
# next_step
if step['handler'] != 'handler_answer':
self.walking_through_scenario(state=state, step=step, steps=steps, text=text, user_id=user_id)
else:
if state.context['answer'] == 'да':
self.walking_through_scenario(state=state, step=step, steps=steps, text=text, user_id=user_id)
else:
next_step = step['failure_step']
state.delete()
self.send_step(step=step, user_id=user_id, text=text, context={})
UserState(user_id=str(user_id), scenario_name=state.scenario_name, step_name=next_step, context={})
else:
# retry_current_step
text_to_send = step['failure_text'].format(**state.context)
self.send_text(text_to_send=text_to_send, user_id=user_id)
# движения по сценарию
def walking_through_scenario(self, state, step, steps, text, user_id):
next_step = steps[step['next_step']]
self.send_step(step=next_step, user_id=user_id, text=text, context=state.context)
if next_step['next_step']:
# switch to next step
state.step_name = step['next_step']
else:
# finish scenario and delete user states
log.info(
'Зарегистрирован пользователь ФИО {name} следует из {city_from} в {city_to} телефон для контактов {phone} '.format(
**state.context))
Registration(name=state.context['name'], city_from=state.context['city_from'],
city_to=state.context['city_to'],
phone=state.context['phone'])
state.delete()
if __name__ == '__main__':
configure_logging()
bot = Bot(group_id=settings.GROUP_ID, token=settings.TOKEN)
bot.run()
它帮了我这个命令 Routes.select(lambda p: p.id > 0 ).delete(bulk=True)
完成我的方案后，我遇到了删除 table 的问题
当我尝试使用 db.drop_table(Routes, if_exists=False, with_all_data=False) or db.drop_table(Routes, if_exists=True, with_all_data=True) 删除时
我得到一个错误
pony.orm.core.TransactionError: @db_session-decorated drop_table() 带有 ddl 选项的函数不能在另一个 db_sessio 内部调用
我尝试使用 Routes.delete() 删除，但我得到了相同的
我阅读了更多文档，但我不知道如何删除此 table
我的完整代码
import logging
import random
import requests
import vk_api
from pony.orm import db_session, rollback
from vk_api.bot_longpoll import VkBotLongPoll, VkBotEventType
import handlers
from models import UserState, Registration, Routes, db
try:
import settings
except ImportError:
exit('DO CP settings.py.default setting.py and set token')
log = logging.getLogger('bot')
def configure_logging():
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(logging.Formatter('%(levelname)s %(message)s'))
stream_handler.setLevel(logging.INFO)
log.addHandler(stream_handler)
file_handler = logging.FileHandler('bot.log')
file_handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s'))
file_handler.setLevel(logging.DEBUG)
log.addHandler(file_handler)
class Bot:
"""
Echo bot for vk.com
Use python 3.7
"""
def __init__(self, group_id, token):
"""
:param group_id:group_id from group vk.com
:param token:secret key
"""
self.group_id = group_id
self.token = token
self.vk = vk_api.VkApi(token=self.token)
self.long_poller = VkBotLongPoll(self.vk, self.group_id)
self.api = self.vk.get_api()
def run(self):
"""run bot"""
for event in self.long_poller.listen():
try:
print(event)
self.on_event(event)
except Exception:
log.exception('ошибка в обработке события')
@db_session
def on_event(self, event):
"""
sends the message back, if message text
:param event: VKBotMessageEvent object
:return: None
"""
if event.type != VkBotEventType.MESSAGE_NEW:
log.info('пока не умеем работать с таким типом события %s', event.type)
return
user_id = event.message.peer_id
text = event.message.text
state = UserState.get(user_id=str(user_id))
if state is not None:
self.continue_scenario(text=text, state=state, user_id=user_id)
else:
# search intent, the user is found outside of the scenario
for intent in settings.INTENTS:
log.debug(f'User gets {intent}')
if any(token in text.lower() for token in intent['tokens']):
# run intent
if intent['answer']:
self.send_text(text_to_send=intent['answer'], user_id=user_id)
else:
self.start_scenario(scenario_name=intent['scenario'], user_id=user_id, text=text)
break
else:
self.send_text(text_to_send=settings.DEFAULT_ANSWER, user_id=user_id)
# функция отправки текста
def send_text(self, text_to_send, user_id):
self.api.messages.send(
message=text_to_send,
random_id=random.randint(0, 2 ** 20),
peer_id=user_id)
# функция отправки изображения
def send_image(self, image, user_id):
upload_url = self.api.photos.getMessagesUploadServer()['upload_url']
upload_data = requests.post(url=upload_url, files={'photo': ('image.png', image, 'image/png')}).json()
image_data = self.api.photos.saveMessagesPhoto(**upload_data)
owner_id = image_data[0]['owner_id']
media_id = image_data[0]['id']
attachment = f'photo{owner_id}_{media_id}'
self.api.messages.send(
attachment=attachment,
random_id=random.randint(0, 2 ** 20),
peer_id=user_id)
# функция отправки маршрутов
def send_routes(self, text_to_send, user_id):
self.api.messages.send(
message=text_to_send,
random_id=random.randint(0, 2 ** 20),
peer_id=user_id)
# шаг отправки
def send_step(self, step, user_id, text, context):
if 'text' in step:
self.send_text(text_to_send=step['text'].format(**context), user_id=user_id)
if 'image' in step:
handler = getattr(handlers, step['image'])
image = handler(text=text, context=context)
self.send_image(image=image, user_id=user_id)
if 'choice' in step:
handler = getattr(handlers, step['choice'])
choices = handler(text=text, context=context)
for i, choice in enumerate(choices, start=1):
city_from = context['city_from']
city_to = context['city_to']
flight_time = choice[city_from][city_to][0]
flight_board = choice[city_from][city_to][1]
flight_date = choice[city_from][city_to][2]
number_choice = str(i) + ') '
Routes(number_choice=number_choice, flight_time=flight_time, city_from=city_from, city_to=city_to,
flight_board=flight_board, flight_date=flight_date)
select = db.select('number_choice,flight_time,city_from,city_to,flight_board,flight_date FROM Routes')
text_to_send = ' '.join(select[0]) + '\n' + ' '.join(select[1]) + '\n' + ' '.join(
select[2]) + '\n' + ' '.join(
select[3]) + '\n' + ' '.join(select[4]) + '\n'
self.send_routes(text_to_send=text_to_send, user_id=user_id)
# начало сценария
def start_scenario(self, scenario_name, user_id, text):
scenario = settings.SCENARIOS[scenario_name]
first_step = scenario['first_step']
step = scenario['steps'][first_step]
self.send_step(step=step, user_id=user_id, text=text, context={})
UserState(user_id=str(user_id), scenario_name=scenario_name, step_name=first_step, context={})
# конец сценария
def continue_scenario(self, text, state, user_id):
steps = settings.SCENARIOS[state.scenario_name]['steps']
step = steps[state.step_name]
handler = getattr(handlers, step['handler'])
if handler(text=text, context=state.context):
# next_step
if step['handler'] != 'handler_answer':
self.walking_through_scenario(state=state, step=step, steps=steps, text=text, user_id=user_id)
else:
if state.context['answer'] == 'да':
self.walking_through_scenario(state=state, step=step, steps=steps, text=text, user_id=user_id)
else:
next_step = step['failure_step']
state.delete()
self.send_step(step=step, user_id=user_id, text=text, context={})
UserState(user_id=str(user_id), scenario_name=state.scenario_name, step_name=next_step, context={})
else:
# retry_current_step
text_to_send = step['failure_text'].format(**state.context)
self.send_text(text_to_send=text_to_send, user_id=user_id)
# движения по сценарию
def walking_through_scenario(self, state, step, steps, text, user_id):
next_step = steps[step['next_step']]
self.send_step(step=next_step, user_id=user_id, text=text, context=state.context)
if next_step['next_step']:
# switch to next step
state.step_name = step['next_step']
else:
# finish scenario and delete user states
log.info(
'Зарегистрирован пользователь ФИО {name} следует из {city_from} в {city_to} телефон для контактов {phone} '.format(
**state.context))
Registration(name=state.context['name'], city_from=state.context['city_from'],
city_to=state.context['city_to'],
phone=state.context['phone'])
state.delete()
if __name__ == '__main__':
configure_logging()
bot = Bot(group_id=settings.GROUP_ID, token=settings.TOKEN)
bot.run()
它帮了我这个命令 Routes.select(lambda p: p.id > 0 ).delete(bulk=True)