Die INSERT FROM SELECT-Anweisung kann nicht mit bindparam ausgeführt werden ⇐ Python

[Anonymous]

Ich habe das folgende vereinfachte Skript, das versucht, Beiträge für vorhandene Benutzer automatisch zu generieren. Der Inhalt des Beitrags wird über bindparam eingefügt.

Code: Select all

from sqlalchemy import String, ForeignKey, Integer, bindparam, create_engine, insert, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
pass

class User(Base):
__tablename__ = "users"

id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)

posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
__tablename__ = "posts"

id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
title: Mapped[str] = mapped_column(String(200), nullable=False)
content: Mapped[str] = mapped_column(String, nullable=False)

author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
author: Mapped["User"] = relationship(back_populates="posts")

engine = create_engine("sqlite:///example.db", echo=False)
Base.metadata.create_all(engine)

users = [
User(username=f"user_{i}")
for i in range(1, 11)
]

with Session(engine) as session:
session.add_all(users)

cte_even_user = select(User).where(User.id % 2 == 0).cte("even_user")

stmt = (
insert(Post)
.from_select(
["title", "content", "author_id"],
select(
("Post for "  + User.username).label("title"),
bindparam('content').label("content"),
cte_even_user.c.id.label("author_id")
)
)
)

session.execute(stmt, {'content': 'Auto-generated post.'})

session.commit()

Wenn ich versuche, das Skript auszuführen, erhalte ich die folgende Fehlermeldung:

Code: Select all

Traceback (most recent call last):
File "app/backend/sqlalc.py", line 61, in 
session.execute(stmt, {'content': 'Auto-generated post.'})
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 2365, in execute
return self._execute_internal(
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 2251, in _execute_internal
result: Result[Any] = compile_state_cls.orm_execute_statement(
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/orm/bulk_persistence.py", line 1280, in orm_execute_statement
result = _bulk_insert(
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/orm/bulk_persistence.py", line 222, in _bulk_insert
result = persistence._emit_insert_statements(
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py", line 1048, in _emit_insert_statements
result = connection.execute(
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1416, in execute
return meth(
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/sql/elements.py", line 516, in _execute_on_connection
return connection._execute_clauseelement(
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1630, in _execute_clauseelement
compiled_sql, extracted_params, cache_hit = elem._compile_w_cache(
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/sql/elements.py", line 704, in _compile_w_cache
compiled_sql = self._compiler(
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/sql/elements.py", line 317, in _compiler
return dialect.statement_compiler(dialect, self, **kw)
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py", line 1429, in __init__
Compiled.__init__(self, dialect, statement, **kwargs)
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py", line 870, in __init__
self.string = self.process(self.statement, **compile_kwargs)
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py", line 915, in process
return obj._compiler_dispatch(self, **kwargs)
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/sql/annotation.py", line 344, in _compiler_dispatch
return self.__element.__class__._compiler_dispatch(
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/sql/visitors.py", line 141, in _compiler_dispatch
return meth(self, **kw)  # type: ignore  # noqa: E501
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/sql/compiler.py", line 5746, in visit_insert
compile_state = insert_stmt._compile_state_factory(
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/sql/base.py", line 683, in create_for_statement
return klass.create_for_statement(statement, compiler, **kw)
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/orm/bulk_persistence.py", line 1338, in create_for_statement
self._setup_for_bulk_insert(compiler)
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/orm/bulk_persistence.py", line 1385, in _setup_for_bulk_insert
self._dict_parameters = {
File "app/venv39/lib/python3.9/site-packages/sqlalchemy/orm/bulk_persistence.py", line 1388, in 
if col.table is emit_insert_table
AttributeError: 'str' object has no attribute 'table'

Aber wenn ich den bindparam so entfernen würde:

Code: Select all

   stmt = (
insert(Post)
.from_select(
["title", "content", "author_id"],
select(
("Post for " + cte_even_user.c.username).label("title"),
cte_even_user.c.username.label("content"),
cte_even_user.c.id.label("author_id")
)
.select_from(cte_even_user)
)
)

Dann funktioniert es.
Verwende ich bindparam falsch?