SNOW-1511473: "type[type]" is incompatible with "type[SnowflakeCursor]"

[tekumara]

Python version

3.9

Operating system and processor architecture

macos

Installed packages

snowflake-connector-python==3.11.0

What did you do?

*_, dcur = conn.execute_string("select 1",cursor_class=snowflake.connector.cursor.DictCursor)
dcur.fetchall()

What did you expect to see?

No type error, but get this in pyright/pylance/vscode:

Can you set logging to DEBUG and collect the logs?

No response

[sfc-gh-sghosh]

Hello @tekumara ,

Thanks for raising the issue, we are looking into it, will update.

Regards,
Sujan

[sfc-gh-sghosh]

Hello @tekumara ,

Could you try the code in this way, I just tried and I am not getting any compilation error in VScode

`cursor1 = con.cursor(snowflake.connector.cursor.DictCursor)

try:
    r1 = cursor1.execute("select 1").fetchall()
finally:
    cursor1.close()`

[tekumara]

That's works but I'm not using execute I'm using execute_string in my codebase to execute multiple sql statements (I simplified it for the MRE above)

[sfc-gh-sghosh]

Hello @tekumara ,

You need to split the statements if using multiple statements

Example
`main_cursor = con.cursor(snowflake.connector.cursor.DictCursor)
print("cursor created")

sql_script = """
CREATE OR REPLACE TABLE my_table (id INT, name STRING);
INSERT INTO my_table VALUES (1, 'Alice'), (2, 'Bob');
SELECT * FROM my_table;
"""

# Split the script into individual statements
sql_statements = sql_script.strip().split(';')


for statement in sql_statements:
    if statement.strip():  # Only execute non-empty statements
        main_cursor.execute(statement)
        try:
            results_cursor = main_cursor.fetchall()
            for row in results_cursor:
                print(row)
        except snowflake.connector.errors.ProgrammingError:
            
            pass

main_cursor.close()
con.close()`

Output:
cursor created
{'status': 'Table MY_TABLE successfully created.'}
{'number of rows inserted': 2}
{'ID': 1, 'NAME': 'Alice'}
{'ID': 2, 'NAME': 'Bob'}

Regards,
Sujan

[tekumara]

Isn't that what execute_string does for me?

[sfc-gh-sghosh]

Hello @tekumara ,

Sorry for the dealy.
As of now, for execute_string, you have to split the statements, as recommended above.

Example:
from snowflake.connector.util_text import split_statements

def execute_file(filename, conn):
with open(filename,'r') as f:
text = f.read()
try:
for stmt in split_statements(StringIO(text)):
conn.execute_string(stmt)
except:
print("Execution aborted")

[tekumara]

Ok but I want a DictCursor and I don't really want to roll my own loop, when I could just use execute_string

ie: I want to use

conn.execute_string("select 1",cursor_class=snowflake.connector.cursor.DictCursor)

But I get this type error from pylance/vscode:

Argument of type "type[DictCursor]" cannot be assigned to parameter "cursor_class" of type "SnowflakeCursor" in function "execute_string"
  Type "type[DictCursor]" is not assignable to type "SnowflakeCursor"

Can the type annotations be fixed to prevent this error?

[sfc-gh-sghosh]

Hello @tekumara ,

Thanks for the update.
At present with execute_string, you can only use cursor_class: SnowflakeCursor = SnowflakeCursor.

So, as a workaround, you can try below code to avoid above error, but you have to split the statements.
later you can convert the rows to dictionary format as below
`
sql_script = """
CREATE OR REPLACE TABLE my_table (id INT, name STRING);
INSERT INTO my_table VALUES (1, 'Alice'), (2, 'Bob');
SELECT * FROM my_table;
"""

with snowflake.connector.connect(**connection_params) as conn:

for result in conn.execute_string(sql_script):
    try:
       
        rows = result.fetchall()  # Directly fetch rows from the result (cursor)
        if rows:
            # Convert rows to dictionary format
            for row in rows:
                print({desc[0]: value for desc, value in zip(result.description, row)})
    except snowflake.connector.errors.ProgrammingError:
                 pass`

Regards,
Sujan

[tekumara]

At present with execute_string, you can only use cursor_class: SnowflakeCursor = SnowflakeCursor.

Using cursor_class=snowflake.connector.cursor.DictCursor with execute_string works for me, ie: fetchall() on the result returns a list of dict.

But the type hints are incorrect - they produce a type error but shouldn't.