在SqlAlchemy中，选择一个未测试的PostgreSQL数组并不能像预期的那样工作

Question

我做了一个SQL查询，它在表中复制一行（实际上是删除并插入新行），同时按给定的时间戳拆分其原始TSRANGE字段。 据我测试，它工作得很好。下面是一个演示来说明我的意思和here is a fiddle为了方便起见：

-- demo initialisation
CREATE TABLE random_table (
    uid VARCHAR(36) PRIMARY KEY,
    id VARCHAR(20),
    tsrange_field TSRANGE
);

CREATE EXTENSION pgcrypto;  -- needed for `gen_random_uuid` function

INSERT INTO public.random_table (uid, id, tsrange_field)
VALUES (gen_random_uuid(), 'random_id', tsrange('2000-01-01', '2020-01-01', '[)'));


-- actual query
WITH splitters AS (
    SELECT uid, datetime
    FROM random_table
        JOIN unnest(ARRAY['2015-04-15'::timestamp, '2016-04-15'::timestamp, '2017-01-01'::timestamp, '2017-04-15'::timestamp]) datetime
            ON tsrange_field @> datetime
    WHERE id = 'random_id'
        AND (lower(random_table.tsrange_field) IS NULL OR lower(random_table.tsrange_field) != datetime)
        AND (upper(random_table.tsrange_field) IS NULL OR upper(random_table.tsrange_field) != datetime)
), to_be_splitted AS (
    DELETE FROM random_table
    USING splitters
    WHERE splitters.uid = random_table.uid
    RETURNING random_table.uid, id, tsrange_field
)
INSERT INTO random_table (uid, id, tsrange_field)
SELECT DISTINCT ON (id, tsrange_field)
    gen_random_uuid() AS uid, id,
    unnest(ARRAY[
        tsrange(
            CASE
                WHEN LAG(splitters.datetime) OVER (PARTITION BY splitters.uid ORDER BY splitters.datetime) IS NOT NULL
                    THEN LAG(splitters.datetime) OVER (PARTITION BY splitters.uid ORDER BY splitters.datetime)
                ELSE lower(tsrange_field)
            END,
            splitters.datetime,
            '[)'
        ),
        tsrange(
            splitters.datetime,
            CASE
                WHEN LEAD(splitters.datetime) OVER (PARTITION BY splitters.uid ORDER BY splitters.datetime) IS NOT NULL
                    THEN LEAD(splitters.datetime) OVER (PARTITION BY splitters.uid ORDER BY splitters.datetime)
                ELSE upper(tsrange_field)
            END,
            '[)'
        )
    ]) AS tsrange_field
FROM to_be_splitted JOIN splitters ON to_be_splitted.uid = splitters.uid
ORDER BY tsrange_field
RETURNING *;

---

现在我想把它翻译成sqlalchemy，这就是我的问题所在。我生成了以下代码：

# pip install psycopg2, sqlalchemy
from datetime import datetime

from sqlalchemy import (and_, case, cast, column, Column, create_engine, delete, func,
                        insert, MetaData, or_, select, Table, VARCHAR)
from sqlalchemy.dialects.postgresql import array, TSRANGE, ARRAY

METADATA = MetaData()

RANDOM_TABLE = Table(
    'random_table', METADATA,
    Column('uid', VARCHAR(36), primary_key=True),
    Column('id', VARCHAR(20)),
    Column('tsrange_field', TSRANGE)
)

engine = create_engine('postgresql://test:test@localhost:5432/test')

def split_row(id, *datetimes):
    # this function contains the translation attempt
    splits = func.unnest([dt for dt in datetimes]).alias('datetime')
    datetime_col = column('datetime')

    splitters = (
        select([RANDOM_TABLE.c.uid, datetime_col])
        .select_from(RANDOM_TABLE.join(
            splits,
            onclause=RANDOM_TABLE.c.tsrange_field.op('@>')(datetime_col)
        ))
        .where(and_(
            RANDOM_TABLE.c.id == id,
            or_(func.lower(RANDOM_TABLE.c.tsrange_field) == None,
                func.lower(RANDOM_TABLE.c.tsrange_field).op('!=')(datetime_col)),
            or_(func.upper(RANDOM_TABLE.c.tsrange_field) == None,
                func.upper(RANDOM_TABLE.c.tsrange_field).op('!=')(datetime_col)),
        ))
    ).cte('splitters')

    to_be_split = (
        delete(RANDOM_TABLE)
        .where(splitters.c.uid == RANDOM_TABLE.c.uid)
        .returning(RANDOM_TABLE.c.uid, RANDOM_TABLE.c.tsrange_field)
    ).cte('to_be_split')

    window_params = {'partition_by': column('uid'),
                     'order_by': datetime_col}
    previous_splitter = func.lag(datetime_col).over(**window_params)
    next_splitter = func.lead(datetime_col).over(**window_params)

    lower_bound_case = case(
        [(previous_splitter != None, previous_splitter)],
        else_=func.lower(column('tsrange_field'))
    )
    upper_bound_case = case(
        [(next_splitter != None, next_splitter)],
        else_=func.upper(column('tsrange_field'))
    )

    split_tsranges = [
        func.tsrange(lower_bound_case, datetime_col, '[)'),
        func.tsrange(datetime_col, upper_bound_case, '[)')
    ]

    split_query = select([
        func.gen_random_uuid().label('uid'),
        func.unnest(split_tsranges).alias('tsrange_field')  # does not work
    ]).distinct(
        column('tsrange_field')
    ).select_from(
        to_be_split
        .join(splitters, onclause=to_be_split.c.uid == splitters.c.uid)
    ).order_by(
        column('tsrange_field')
    )

    whole_query = (
        insert(RANDOM_TABLE)
        .from_select([column('uid'), column('tsrange_field')], split_query)
        .returning(column('uid'), column('tsrange_field'))
    )
    return whole_query


with engine.connect() as conn:
    query = split_row('random_id', datetime.now())
    import pdb; pdb.set_trace()
    print(conn.execute(query).fetchall())

不幸的是，它失败了，并显示以下消息：

Traceback (most recent call last):
  File "/home/tryph/sql_split/.env/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1244, in _execute_context
    cursor, statement, parameters, context
  File "/home/tryph/sql_split/.env/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 552, in do_execute
    cursor.execute(statement, parameters)
psycopg2.ProgrammingError: can't adapt type 'Function'

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

Traceback (most recent call last):
  File "stuck.py", line 85, in <module>
    print(conn.execute(query).fetchall())
  File "/home/tryph/sql_split/.env/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 988, in execute
    return meth(self, multiparams, params)
  File "/home/tryph/sql_split/.env/lib/python3.6/site-packages/sqlalchemy/sql/elements.py", line 287, in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
  File "/home/tryph/sql_split/.env/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1107, in _execute_clauseelement
    distilled_params,
  File "/home/tryph/sql_split/.env/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1248, in _execute_context
    e, statement, parameters, cursor, context
  File "/home/tryph/sql_split/.env/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1466, in _handle_dbapi_exception
    util.raise_from_cause(sqlalchemy_exception, exc_info)
  File "/home/tryph/sql_split/.env/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 383, in raise_from_cause
    reraise(type(exception), exception, tb=exc_tb, cause=cause)
  File "/home/tryph/sql_split/.env/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 128, in reraise
    raise value.with_traceback(tb)
  File "/home/tryph/sql_split/.env/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1244, in _execute_context
    cursor, statement, parameters, context
  File "/home/tryph/sql_split/.env/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 552, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) can't adapt type 'Function'
[SQL: WITH splitters AS 
(SELECT random_table.uid AS uid, datetime 
FROM random_table JOIN unnest(%(unnest_3)s) AS datetime ON random_table.tsrange_field @> datetime 
WHERE random_table.id = %(id_1)s AND (lower(random_table.tsrange_field) IS NULL OR (lower(random_table.tsrange_field) != datetime)) AND (upper(random_table.tsrange_field) IS NULL OR (upper(random_table.tsrange_field) != datetime))), 
to_be_split AS 
(DELETE FROM random_table USING splitters WHERE splitters.uid = random_table.uid RETURNING random_table.uid, random_table.tsrange_field)
 INSERT INTO random_table (uid, tsrange_field) SELECT DISTINCT ON (tsrange_field) gen_random_uuid() AS uid, tsrange_field.unnest_1 
FROM unnest(%(unnest_2)s) AS tsrange_field, to_be_split JOIN splitters ON to_be_split.uid = splitters.uid ORDER BY tsrange_field RETURNING uid, tsrange_field]
[parameters: {'unnest_2': [<sqlalchemy.sql.functions.Function at 0x7fba18954a20; tsrange>, <sqlalchemy.sql.functions.Function at 0x7fba18954b00; tsrange>], 'unnest_3': [datetime.datetime(2019, 5, 3, 23, 37, 1, 773118)], 'id_1': 'random_id'}]
(Background on this error at: http://sqlalche.me/e/f405)

---

查看生成的SQL，我注意到原始SQL查询的unnest(ARRAY[...]) AS tsrange_field中的SELECT在FROM子句中呈现，我不知道为什么。此外，消息psycopg2.ProgrammingError: can't adapt type 'Function'并没有提供很大的帮助，似乎与unnest误译无关。你知道吗

任何关于发生了什么以及如何解决它的提示都将不胜感激。你知道吗

Answer 1

正如你所说的，有问题的部分是

split_query = select([
        func.gen_random_uuid().label('uid'),
        func.unnest(split_tsranges).alias('tsrange_field')  # does not work
    ])

问题是^{}用于生成适合FROM子句的命名别名，因此SQLAlchemy将其移到那里。使用label('tsrange_field')来生成AS tsrange_field。另一个问题是SQLAlchemy将列表按原样传递给DB-API驱动程序，而DB-API驱动程序不知道如何处理SQLAlchemy构造。在对^{}的调用中包装列表，以便SQLAlchemy使用嵌套表达式呈现ARRAY文本：

split_tsranges = array([
        func.tsrange(lower_bound_case, datetime_col, '[)'),
        func.tsrange(datetime_col, upper_bound_case, '[)')
    ])