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)
]
# reusable statement
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")
)
)
)
with Session(engine) as session:
session.add_all(users)
session.flush()
session.execute(stmt, {'content': 'Auto-generated post.'})
session.execute(stmt, {'content': 'Another one.'})
session.execute(stmt, {'content': 'Yet another one.'})
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?
insert(Post) executed with the data mapping {"content": "Auto-generated post."} triggers bulk insert by passing Post as a mapped ORM entity, instead of passing a table which would have treated insert as a regular table insert. In bulk insert rows are inserted using dictionaries rather than ORM objects. In which the keys are column names and the values are the data to insert. Executing the ORM DML statement through a session with parameter mapping causes SQLAlchemy to interpret the dictionary as row data for the Post.content ORM column and not as a SQL bind-parameter dictionary. bindparam("content") causes "content" to become the name of a SQL bind param and when executed, the bulk insert causes it to be treated as {"content": <value for Post.content column>}.
basically, SQLAlchemy has two different things named "content". The ORM column Post.content and the SQL bind parameter :content. Inside the bulk insert compiler this causes confusion, and it eventually attempts to interpret the string "content" as a column object ("content".table), producing the following AttributeError.
AttributeError: 'str' object has no attribute 'table'
you can fix this by using insert(Post.__table__) and passing the table instead of the ORM object, making your code look like this:
stmt = (
insert(Post.__table__)
.from_select(
[Post.title, Post.content, Post.author_id],
select(
("Post for " + cte_even_user.c.username).label("title"),
bindparam("content").label("content"),
cte_even_user.c.id.label("author_id"),
).select_from(cte_even_user)
)
)
session.execute(stmt, {"content": "Auto-generated post."})