SQLalchemy rowcount всегда -1 для операторов

Question

Я играл с SQLalchemy и Microsoft SQL Server, чтобы разобраться в функциях, когда обнаружил странное поведение. Меня учили, что атрибут rowcount в прокси-объекте результата покажет, сколько строк было выполнено при выполнении оператора. Однако, когда я выбираю или вставляю одну или несколько строк в свою тестовую базу данных, я всегда получаю -1. Как это могло быть и как это исправить, чтобы отразить реальность?

connection = engine.connect()
metadata = MetaData()

# Ex1: select statement for all values
student = Table('student', metadata, autoload=True, autoload_with=engine)
stmt = select([student])
result_proxy = connection.execute(stmt)
results = result_proxy.fetchall()
print(result_proxy.rowcount)

# Ex2: inserting single values
stmt = insert(student).values(firstname='Severus', lastname='Snape')
result_proxy = connection.execute(stmt)
print(result_proxy.rowcout)
 
# Ex3: inserting multiple values 
stmt = insert(student)
values_list = [{'firstname': 'Rubius', 'lastname': 'Hagrid'},
               {'firstname': 'Minerva', 'lastname': 'McGonogall'}]
result_proxy = connection.execute(stmt, values_list)
print(result_proxy.rowcount)

Функция печати для каждого блока, отдельно запускаемого примера кода, печатает -1. Ex1 успешно выбирает все строки, и оба оператора вставки успешно записывают данные в базу данных.

Согласно следующей проблеме, атрибуту rowcount не всегда можно доверять. Это правда и здесь? И когда, как я могу компенсировать это с помощью оператора Count в транзакции SQLalcehmy? PDO :: rowCount () возвращает -1

Answer 1

Однострочный элемент INSERT … VALUES ( … )тривиален: если оператор завершился успешно, то была затронута одна строка, а если он не прошел (вызывает ошибку), то затронуты нулевые строки.

Для многострочного типа INSERTпросто выполните это внутри транзакции и откатитесь в случае возникновения ошибки. Тогда количество затронутых строк будет либо равно нулю, либо len(values_list).

Чтобы получить количество строк, которое вернет SELECT , оберните запрос select в SELECT count(*)запрос и сначала запустите его, например:

select_stmt = sa.select([Parent])
count_stmt = sa.select([sa.func.count(sa.text("*"))]).select_from(
    select_stmt.alias("s")
)
with engine.connect() as conn:
    conn.execution_options(isolation_level="SERIALIZABLE")
    rows_found = conn.execute(count_stmt).scalar()
    print(f"{rows_found} row(s) found")
    results = conn.execute(select_stmt).fetchall()
for item in results:
    print(item.id)