可以使用sqlalchemy将值插入tsrange列中

Question

我是Python和Sqlalchemy的初学者。

我现在有这个表

import sqlalchemy
from sqlalchemy import BIGINT
from sqlalchemy.dialects.postgresql import UUID, TSRANGE

from application import metadata
from models.Staff import _staff_table_name

StaffAvailability = sqlalchemy.Table(
    "staff_availability",
    metadata,
    sqlalchemy.Column(
        "id",
        BIGINT,
        autoincrement=True,
        primary_key=True
    ),
    sqlalchemy.Column(
        "staff_id",
        UUID,
        sqlalchemy.ForeignKey(f"{_staff_table_name}.staff_id", ondelete="CASCADE"),
        nullable=False
    ),
    sqlalchemy.Column("during", TSRANGE(), nullable=True),
    sqlalchemy.Column("created_at", sqlalchemy.DateTime(timezone=True), nullable=True, server_default="now()"),
)

，如果我尝试将任何值插入此处，它将返回错误，

    staff_availability = StaffAvailability.insert()
    avail_id = await database.execute(staff_availability, values={
        "staff_id": kwargs.get("staff_id"),
        "during": DateTimeRange(datetime(2021, 3, 22, 10, 0, 0), datetime(2021, 3, 25, 10, 0, 0))
    })

我从psycopg2导入dateTimerange

from psycopg2.extras import DateTimeRange

我不知道为什么这不起作用。我还尝试了其他方法使用text函数

"during": text("'[2022-10-10 10:00, 2020-10-10 12:00)'")

，其中我从sqlalchemy导入text函数，

from sqlalchemy import text

我也尝试导入text> text text 这样的

from sqlalchemy.sql import text

功能 它会生成的错误，我得到此错误

 result = await executor(stmt, None)
  File "asyncpg/protocol/protocol.pyx", line 183, in bind_execute
  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: DateTimeRange(datetime.datetime(2021, 3,... (list, tuple or Range object expected (got type <class 'psycopg2._range.DateTimeRange'>))

当我尝试使用dateTimerange尝试使用text> text函数插入值时，

File "/Users/koushik/Documents/projects/colorus81/env/lib/python3.8/site-packages/fastapi/encoders.py", line 144, in jsonable_encoder
    raise ValueError(errors)
ValueError: [TypeError("'Range' object is not iterable"), TypeError('vars() argument must have __dict__ attribute')]

Answer 1

尝试各种方式后。这就是我解决的方式。

这个人如何运作很奇怪。

from sqlalchemy.sql import text

   from_time_str = from_time.strftime("%Y-%m-%d %H:%M:%S")
    to_time_str = to_time.strftime("%Y-%m-%d %H:%M:%S")
    stmt = text(f"insert into staff_availability(staff_id, during) values (:staff_id, '[{from_time_str},{to_time_str})') returning id")
    stmt = stmt.bindparams(staff_id=kwargs.get('staff_id'))

但是，

from_time_str = from_time.strftime("%Y-%m-%d %H:%M:%S")
    to_time_str = to_time.strftime("%Y-%m-%d %H:%M:%S")
    stmt = text(
        f"insert into staff_availability(staff_id, during) "
        f"values (:staff_id, :during) returning id")
    stmt = stmt.bindparams(staff_id=kwargs.get('staff_id'), during=f'[{from_time_str},{to_time_str})')

如果我在值中将放入bind param中，则该错误不起作用并返回此错误，这将不起作用。

File "/Users/koushik/Documents/projects/colorus81/env/lib/python3.8/site-packages/asyncpg/connection.py", line 1684, in __execute
    return await self._do_execute(
  File "/Users/koushik/Documents/projects/colorus81/env/lib/python3.8/site-packages/asyncpg/connection.py", line 1731, in _do_execute
    result = await executor(stmt, None)
  File "asyncpg/protocol/protocol.pyx", line 183, in bind_execute
  File "asyncpg/protocol/prepared_stmt.pyx", line 197, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
asyncpg.exceptions.DataError: invalid input for query argument $1: '[2022-10-10 10:00:00,2022-10-10 14:00:0... (list, tuple or Range object expected (got type <class 'str'>))

Answer 2

尝试使用asyncpg驱动程序的类型range：

from asyncpg.types import Range

during = Range(datetime(2021, 3, 22, 10, 0, 0), datetime(2021, 3, 25, 10, 0, 0))