如何将feedparser日期通过asyncpg存入PostgreSQL的timestamptz字段？

问题背景 #

尝试通过asyncpg将feedparser获取的RSS数据存入PostgreSQL数据库时，存储类型为timestamptz的pubdate字段出现类型不匹配错误。PostgreSQL表test结构如下：

+---------------------+----------------------+
| feed_item_id (uuid) | pubdate(timestamptz) |
+---------------------+----------------------+
|                 ... |                  ... |
+---------------------+----------------------+

原始代码 #

用于加载RSS数据并保存的代码：

def md5(text):
    import hashlib
    return hashlib.md5(text.encode('utf-8')).hexdigest()

def fetch():
    import feedparser
    data = feedparser.parse('https://cointelegraph.com/rss')
    return data

async def insert(rows):
    import asyncpg
    async with asyncpg.create_pool(user='postgres', database='postgres') as pool:
        async with pool.acquire() as conn:
            results = await conn.executemany('INSERT INTO test (feed_item_id, pubdate) VALUES($1, $2)', rows)
            print(results)

async def main():
    data = fetch()
    first_entry = data.entries[0]
    await insert([(md5(first_entry.guid), first_entry.published)])

import asyncio
asyncio.run(main())

错误情况1：使用published字段 #

执行后报错，提示字符串类型不符合timestamptz要求：

Traceback (most recent call last):
  File "asyncpg/protocol/prepared_stmt.pyx", line 168, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
  File "asyncpg/protocol/codecs/base.pyx", line 206, in asyncpg.protocol.protocol.Codec.encode
  File "asyncpg/protocol/codecs/base.pyx", line 111, in asyncpg.protocol.protocol.Codec.encode_scalar
  File "asyncpg/pgproto/./codecs/datetime.pyx", line 208, in asyncpg.pgproto.pgproto.timestamptz_encode
TypeError: expected a datetime.date or datetime.datetime instance, got 'str'

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/opt/homebrew/Cellar/python@3.9/3.9.13_1/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/opt/homebrew/Cellar/python@3.9/3.9.13_1/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/base_events.py", line 647, in run_until_complete
    return future.result()
  File "<stdin>", line 4, in main
  File "<stdin>", line 5, in insert
  File "/Users/vr/.local/share/virtualenvs/python-load-feed-items-uopsj7-P/lib/python3.9/site-packages/asyncpg/connection.py", line 358, in executemany
    return await self._executemany(command, args, timeout)
  File "/Users/vr/.local/share/virtualenvs/python-load-feed-items-uopsj7-P/lib/python3.9/site-packages/asyncpg/connection.py", line 1697, in _executemany
    result, _ = await self._do_execute(query, executor, timeout)
  File "/Users/vr/.local/share/virtualenvs/python-load-feed-items-uopsj7-P/lib/python3.9/site-packages/asyncpg/connection.py", line 1731, in _do_execute
    result = await executor(stmt, None)
  File "asyncpg/protocol/protocol.pyx", line 254, in bind_execute_many
  File "asyncpg/protocol/coreproto.pyx", line 958, in asyncpg.protocol.protocol.CoreProtocol._bind_execute_many_more
  File "asyncpg/protocol/protocol.pyx", line 220, in genexpr
  File "asyncpg/protocol/prepared_stmt.pyx", line 197, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
asyncpg.exceptions.DataError: invalid input for query argument $2 in element #0 of executemany() sequence: 'Thu, 04 Aug 2022 03:29:19 +0100' (expected a datetime.date or datetime.datetime instance, got 'str')

错误情况2：使用published_parsed字段 #

改用published_parsed后，提示struct_time类型仍不匹配：

async def main():
    ...
    await insert([(md5(first_entry.guid), first_entry.published_parsed)])

asyncio.run(main())

错误信息：

Traceback (most recent call last):
  File "asyncpg/protocol/prepared_stmt.pyx", line 168, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
  File "asyncpg/protocol/codecs/base.pyx", line 206, in asyncpg.protocol.protocol.Codec.encode
  File "asyncpg/protocol/codecs/base.pyx", line 111, in asyncpg.protocol.protocol.Codec.encode_scalar
  File "asyncpg/pgproto/./codecs/datetime.pyx", line 208, in asyncpg.pgproto.pgproto.timestamptz_encode
TypeError: expected a datetime.date or datetime.datetime instance, got 'struct_time'

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/opt/homebrew/Cellar/python@3.9/3.9.13_1/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/opt/homebrew/Cellar/python@3.9/3.9.13_1/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/base_events.py", line 647, in run_until_complete
    return future.result()
  File "<stdin>", line 4, in main
  File "<stdin>", line 5, in insert
  File "/Users/vr/.local/share/virtualenvs/python-load-feed-items-uopsj7-P/lib/python3.9/site-packages/asyncpg/connection.py", line 358, in executemany
    return await self._executemany(command, args, timeout)
  File "/Users/vr/.local/share/virtualenvs/python-load-feed-items-uopsj7-P/lib/python3.9/site-packages/asyncpg/connection.py", line 1697, in _executemany
    result, _ = await self._do_execute(query, executor, timeout)
  File "/Users/vr/.local/share/virtualenvs/python-load-feed-items-uopsj7-P/lib/python3.9/site-packages/asyncpg/connection.py", line 1731, in _do_execute
    result = await executor(stmt, None)
  File "asyncpg/protocol/protocol.pyx", line 254, in bind_execute_many
  File "asyncpg/protocol/coreproto.pyx", line 958, in asyncpg.protocol.protocol.CoreProtocol._bind_execute_many_more
  File "asyncpg/protocol/protocol.pyx", line 220, in genexpr
  File "asyncpg/protocol/prepared_stmt.pyx", line 197, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
asyncpg.exceptions.DataError: invalid input for query argument $2 in element #0 of executemany() sequence: time.struct_time(tm_year=2022, tm_mon=8,... (expected a datetime.date or datetime.datetime instance, got 'struct_time')

asyncpg要求存入timestamptz字段的必须是Python标准库中的datetime.datetime实例，只需将feedparser返回的日期格式转换为该类型即可，以下是两种可行方法：

方法1：将published_parsed转换为datetime #

published_parsed返回的time.struct_time对象可以直接通过datetime.datetime构造函数转换：

import datetime

async def main():
    data = fetch()
    first_entry = data.entries[0]
    # 提取struct_time的前6个元素（年、月、日、时、分、秒）构造datetime
    pub_datetime = datetime.datetime(*first_entry.published_parsed[:6])
    await insert([(md5(first_entry.guid), pub_datetime)])

注：published_parsed已经是UTC时间，转换后asyncpg会自动适配PostgreSQL的timestamptz时区处理。

方法2：解析published字符串为datetime #

如果需要直接处理日期字符串，可使用python-dateutil库的解析工具（先执行pip install python-dateutil安装）：

from dateutil import parser

async def main():
    data = fetch()
    first_entry = data.entries[0]
    # 自动识别字符串中的时区信息，生成带时区的datetime对象
    pub_datetime = parser.parse(first_entry.published)
    await insert([(md5(first_entry.guid), pub_datetime)])

这种方法无需手动处理格式，兼容性更强。

完整修正代码示例 #

以下是采用方法1的完整可运行代码：

import hashlib
import feedparser
import asyncpg
import asyncio
import datetime

def md5(text):
    return hashlib.md5(text.encode('utf-8')).hexdigest()

def fetch():
    data = feedparser.parse('https://cointelegraph.com/rss')
    return data

async def insert(rows):
    async with asyncpg.create_pool(user='postgres', database='postgres') as pool:
        async with pool.acquire() as conn:
            results = await conn.executemany('INSERT INTO test (feed_item_id, pubdate) VALUES($1, $2)', rows)
            print(results)

async def main():
    data = fetch()
    first_entry = data.entries[0]
    pub_datetime = datetime.datetime(*first_entry.published_parsed[:6])
    await insert([(md5(first_entry.guid), pub_datetime)])

asyncio.run(main())

内容的提问来源于stack exchange，提问作者PirateApp