python psycopg cursor : str is not assignable to LiteralString

Question

I want to switch from synchronous db calls to asynchronous db calls. I don't want to change my query builder. My query builder outputs a sql query as a string (actually in reality it outputs a query with '%s' to escape parameters and the escaped parameters in a list).

This works great in psycopg2 but does not seem to work in psycopg since the cursor does not want to take an "str" as the query (quite annoying). How can I make it work ?

import asyncio
import psycopg


class MyQueryBuilderObj:
    def getFinalQuery(self) -> str:
        return "SELECT * FROM test_table"

async def fetch_data_from_db():
    qb_obj = MyQueryBuilderObj()
    async with await psycopg.AsyncConnection.connect("dbname=test user=postgres password=yourpassword host=localhost") as aconn:
        async with aconn.cursor() as acur:
            await acur.execute(qb_obj.getFinalQuery())
            result = await acur.fetchone()
            return result

async def main():
    try:
        result = await fetch_data_from_db()
        print("Fetched data:", result)
    except Exception as e:
        print("An error occurred:", e)

if __name__ == "__main__":
    asyncio.run(main())

Issue at "acur.execute" :

Argument of type "str" cannot be assigned to parameter "query" of type "Query" in function "execute"
  Type "str" is not assignable to type "Query"
    "str" is not assignable to "LiteralString"
    "str" is not assignable to "bytes"
    "str" is not assignable to "SQL"
    "str" is not assignable to "Composed"

Version :

• psycopg 3.2.5
• python 3.12.5

Answer 1

I encountered the same issue when building dynamic queries.

The problem came from type conversion. The type checker won't like it, but it will work.

def select_row(row_name: str):
    query_template = "SELECT {} FROM pg_catalog.pg_tables"
    query = query_template.format(row_name)
    with psycopg.connect(**connection) as conn:
        with conn.cursor() as cur:
            cur.execute(query)
            result = cur.fetchone()

select_row('table'+'name')

A simple change of type from str to LiteralString made the problem go away.

Another solution is type casting, where you explicitly give the function what it wants.

import typing
query = typing.cast(typing.LiteralString, query)

---

The other answer did not work as SQL didn't liked string or even casted to it's own psycopg Literal type. (...type "str"/"Literal" cannot be assigned to parameter of type "LiteralString"...)

from psycopg.sql import Literal,SQL
query = Literal(query)
SQL(query)