ARTICLE AD BOX
I have the following simplified script which tries to auto generate posts for existing user. The content of the post is inserted through bindparam.
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()When I try to run the script, I get the following error:
Traceback (most recent call last): File "app/backend/sqlalc.py", line 61, in <module> 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 <dictcomp> if col.table is emit_insert_table AttributeError: 'str' object has no attribute 'table'But if I were to remove the bindparam like this:
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) ) )Then it works.
Am I using bindparam the wrong way?
