How to pass an array of customs type to a PostgreSQL function

Question

I have a custom array type in Postgres:

CREATE TYPE core.arr_message_input AS (
    idmessage uuid,
    idplugin integer,
    versionplugin numeric,
    ttl integer
);

There is a simple function to add records to the table:

CREATE OR REPLACE FUNCTION queue_push(
    arr_message_input[])
    RETURNS Bool
    LANGUAGE 'plpgsql'
AS $$
    BEGIN
        INSERT INTO queue(idmessage, idplugin, versionplugin, queuetime, ttl)
        SELECT idmessage, idplugin, versionplugin, now(), ttl
        FROM unnest ($1);
        RETURN True;
    EXCEPTION
        WHEN others THEN
            RETURN False;
    END $$;

Filling in values from Postgres is easy:

SELECT queue_push(
    array[
        ('e62c7924-2cd1-4dd6-9b55-d4e612816ce0', 2, 0, 0),
        ('a7e864af-4c4c-452d-9df2-f9d4f70ac02e', 2, 0, 0),
    ]::arr_message_input[]
);

But through SQLAlchemy I can't figure out how to do it. I pass it a list as an array, but there should be a list of lists or something similar. And I have no way of doing it from Python.

For example:

The function model is described as follows:

class QueuePush(GenericFunction):

    name = "queue_push"

    @staticmethod
    def mapped_objects(**kwargs):
        return select(
            [
                Column('queue_push', BOOLEAN),
            ]
        ).select_from(
            func.queue_push(
                kwargs['arr_message_input'],
            )
        ).alias(name="queue_push")

Request to the function:

import QueuePush


messages = [
    ['027d6e96-84b7-4f10-8640-13dfa1b05fd8', 3, 0, 2],
]

queue = db.query(QueuePush.mapped_objects(arr_message_input=messages)).all()

But still, the created type is a kind of data structure. I'm obviously doing something wrong.

Ilja Everilä · Accepted Answer · 2019-11-25 14:34:11Z

The tuple syntax can be produced using tuple_(), and using postgresql.array() you can produce array literals. Combining these you get:

from sqlalchemy import tuple_
from sqlalchemy.dialects.postgresql import array

class QueuePush(GenericFunction):

    name = "queue_push"

    @staticmethod
    def mapped_objects(**kwargs):
        input = array([tuple_(*t) for t in kwargs['arr_message_input']]
        return select(
            [
                Column('queue_push', BOOLEAN),
            ]
        ).select_from(
            func.queue_push( 
                # Eurgh, this cast handling is horrid, and uses the postgresql
                # specific cast syntax, but it'd require creating a UserDefinedType
                # or such just to use the proper cast method.
                input.op("::")(text("arr_message_input[]"))
            )
        ).alias(name="queue_push")

For advanced usage you might define a UserDefinedType and a bind_expression():

from sqlalchemy import tuple_
from sqlalchemy.types import UserDefinedType

class ArrMessageInput(UserDefinedType):

    def get_col_spec(self, **kw):
        return "arr_message_input"

    def bind_expression(self, val):
        return tuple_(*val)

and then

from sqlalchemy import tuple_
from sqlalchemy.dialects.postgresql import array, ARRAY

class QueuePush(GenericFunction):

    name = "queue_push"

    @staticmethod
    def mapped_objects(**kwargs):
        # For some reason the `bind_expression` is not applied to array elements,
        # even if using `array(..., type_=ArrMessageInput)`
        input = array([tuple_(*t) for t in kwargs['arr_message_input']])
        return select(
            [
                Column('queue_push', BOOLEAN),
            ]
        ).select_from(
            func.queue_push( 
                input.cast(ARRAY(ArrMessageInput))
            )
        ).alias(name="queue_push")

On the other hand if you are using psycopg2, you can rely on its adaptation of Python lists and tuples, and just pass the structure as a literal, cast to the correct type:

from sqlalchemy import literal
from sqlalchemy.dialects.postgresql import ARRAY

class QueuePush(GenericFunction):

    name = "queue_push"

    @staticmethod
    def mapped_objects(**kwargs):
        # `literal` is required, so that the cast doesn't apply SQLA type handling.
        # Note the use of Python's `tuple` instead of SQLAlchemy `tuple_`
        input = literal([tuple(t) for t in kwargs['arr_message_input']])
        return select(
            [
                Column('queue_push', BOOLEAN),
            ]
        ).select_from(
            func.queue_push( 
                input.cast(ARRAY(ArrMessageInput))
            )
        ).alias(name="queue_push")

Collectives™ on Stack Overflow

How to pass an array of customs type to a PostgreSQL function

1 Answer 1

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related