我一直在使用psycopg2管理PostgreSQL数据库中的项目。最近有人建议我可以在代码中使用asyncio和asyncpg来改进数据库事务。我查看了堆栈溢出,并阅读了文档中的示例。我已经能够创建表和插入记录,但是我还没有得到我想要的执行反馈。
例如,在我的代码中,我可以在插入记录之前验证一个表是否存在或不存在。
def table_exists(self, verify_table_existence, name):
'''Verifies the existence of a table within the PostgreSQL database'''
try:
self.cursor.execute(verify_table_existence, name)
answer = self.cursor.fetchone()[0]
if answer == True:
print('The table - {} - exists'.format(name))
return True
else:
print ('The table - {} - does NOT exist'.format(name))
return False
except Exception as error:
logger.info('An error has occurred while trying to verify the existence of the table {}'.format(name))
logger.info('Error message: {}').format(error)
sys.exit(1)
使用asyncpg,我无法获得相同的反馈。我该怎么做?
import asyncpg
import asyncio
async def main():
conn = await asyncpg.connect('postgresql://postgres:mypassword@localhost:5432/mydatabase')
answer = await conn.fetch('''
SELECT EXISTS (
SELECT 1
FROM pg_tables
WHERE schemaname = 'public'
AND tablename = 'test01'
); ''')
await conn.close()
#####################
# the fetch returns
# [<Record exists=True>]
# but prints 'The table does NOT exist'
#####################
if answer == True:
print('The table exists')
else:
print('The table does NOT exist')
asyncio.get_event_loop().run_until_complete(main())
最佳答案
您在psycopg2中使用了fetchone()[0]
,但在asyncpg中使用了fetch(...)
。前者将检索第一行的第一列,而后者将检索整个行列表。作为一个列表,它不等于True
。
要从一行中获取单个值,请使用类似answer = await conn.fetchval(...)
的方法。