import psycopg2
from psycopg2 import sql
import datetime


def connect(conn):
    """ Connect to the PostgreSQL database server """
    # create a cursor
    cur = conn.cursor()

    # https://www.psycopg.org/docs/sql.html

    col_names = sql.SQL(', ').join([
        sql.Identifier('lookup'),
        sql.Identifier('LOL1'),
        sql.Identifier('LOL2'),
        sql.Identifier('identifier1'),
        sql.Identifier('identifier2'),
        sql.Identifier('timestamp_first_entry'),
        sql.Identifier('timestamp_last_entry'),
        sql.Identifier('some_number')
    ]),

    values = sql.SQL(', ').join([
        "hash",
        [  # LOL1
            [[1.0, 2.0],  # list of lists of lists 1.1
             [3.0, 0.5]],
            [[-1.0, -2.0], # list of lists of lists 1.2
             [-3.0, -4.0]]
        ], [  # LOL2
            [[1.0, 2.0],  # list of lists of lists 2.1
             [3.0, 0.5]],
            [[-1.0, -2.0], # list of lists of lists 2.2
             [-3.0, -4.0]]
        ],
        "identifier1",
        "identifier2",
        datetime.datetime(2021, 5, 10),
        datetime.datetime(2021, 5, 12),
        20
    ])

    query_base = sql.SQL("insert into {table_name}({col_names} values ({values}))".format(
        table_name=sql.Identifier("raw_orderbook"),
        col_names=col_names,
        values=values
    ))

    cur.execute(query_base)

    cur.close()


# https://www.psycopg.org/docs/sql.html

if __name__ == '__main__':
    conn = psycopg2.connect(
        host="localhost",
        database="test",
        user="ian",
        password="")

    connect(conn)

results in an error of

Traceback (most recent call last):
  File "/home/ian/PycharmProjects/panthera/src/database/tester.py", line 73, in <module>
    connect(conn)
  File "/home/ian/PycharmProjects/panthera/src/database/tester.py", line 33, in connect
    values = sql.SQL(', ').join([
  File "/home/ian/anaconda3/envs/panthera/lib/python3.9/site-packages/psycopg2/sql.py", line 288, in join
    return Composed(rv)
  File "/home/ian/anaconda3/envs/panthera/lib/python3.9/site-packages/psycopg2/sql.py", line 109, in __init__
    raise TypeError(
TypeError: Composed elements must be Composable, got 'lookup' instead

Given that composable is the base class, I’m not really sure what the error is because of. As far as I know it seems correct based on the examples from the docs. My sql_schema

DROP TABLE IF EXISTS "raw_orderbooks";
CREATE TABLE "raw_orderbooks"
(

    lookup                text, -- essentially our lookup key comprising: hash(exchange_pair_timestapOfFirstEntry)
    LOL1                  DOUBLE PRECISION[][][],
    LOL2                  DOUBLE PRECISION[][][],
    identifier1                  text,
    identifier2            text,
    timestamp_first_entry TIMESTAMP,
    timestamp_last_entry  TIMESTAMP,
    some_number       int
);

one other idea I had was to wrap the non-datetime values in sql.Literal but then I run into the datetime issue.

Looking around, I’ve seen other people using different notation but it feels less “clean” which is definitely subjective but IDK.

Edit

Based on the comments from the answer, my updated script is

import psycopg2
from psycopg2 import sql
import datetime


def connect(conn):
    """ Connect to the PostgreSQL database server """
    # create a cursor
    cur = conn.cursor()

    # https://www.psycopg.org/docs/sql.html

    col_names = sql.SQL(', ').join([
        sql.Identifier('lookup'),
        sql.Identifier('LOL1'),
        sql.Identifier('LOL2'),
        sql.Identifier('identifier1'),
        sql.Identifier('identifier2'),
        sql.Identifier('timestamp_first_entry'),
        sql.Identifier('timestamp_last_entry'),
        sql.Identifier('some_number')
    ]),

    values = sql.SQL(', ').join([
        "hash",
        [  # LOL1
            [[1.0, 2.0],  # list of lists of lists 1.1
             [3.0, 0.5]],
            [[-1.0, -2.0], # list of lists of lists 1.2
             [-3.0, -4.0]]
        ], [  # LOL2
            [[1.0, 2.0],  # list of lists of lists 2.1
             [3.0, 0.5]],
            [[-1.0, -2.0], # list of lists of lists 2.2
             [-3.0, -4.0]]
        ],
        "identifier1",
        "identifier2",
        datetime.datetime(2021, 5, 10),
        datetime.datetime(2021, 5, 12),
        20
    ])
    table_col_names = ['lookup','LOL1','LOL2','identifier1','identifier2','timestamp_first_entry','timestamp_last_entry','some_number']

    col_names = sql.SQL(', ').join(sql.Identifier(n) for n in table_col_names )

    place_holders = sql.SQL(', ').join(sql.Placeholder() * len(table_col_names ))

    query_base = sql.SQL("insert into {table_name} ({col_names}) values ({values})").format(
        table_name=sql.Identifier("raw_orderbook"),
        col_names=col_names,
        values=place_holders
    )

    print(query_base.as_string(conn))
    cur.execute(query_base,values)

    cur.close(

but I’m still running into the issue TypeError: Composed elements must be Composable, got 'lookup' instead

Answer

sql.SQL.join join a sequence of Composable but you are providing normal variable like string , list etc in values variable.

secondly i will suggest use placeholder for values in query then pass values to execute .

table_col_names = ['lookup','LOL1','LOL2','identifier1','identifier2','timestamp_first_entry','timestamp_last_entry','some_number']

col_names = sql.SQL(', ').join(sql.Identifier(n) for n in table_col_names )

place_holders = sql.SQL(', ').join(sql.Placeholder() * len(table_col_names ))

query_base = sql.SQL("insert into {table_name} ({col_names}) values ({values})").format(
        table_name=sql.Identifier("raw_orderbook"),
        col_names=col_names,
        values=place_holders
    )

print(query_base.as_string(conn))
cur.execute(query_base,values)

psycopg2 sql.SQL: Composed elements must be Composable

Edit

Advertisement

Answer