SQLAlchemy: QueryableAttribute missing arguments

Question

I have a SQLAlchemy model mapped as a dataclass (single class example below)

class Base(DeclarativeBase):
    pass


class Players(Base, MappedAsDataclass):
    __tablename__ = "players"
    player_uid: Mapped[str] = mapped_column(String(128), primary_key=True)
    first_name: Mapped[str] = mapped_column(String(32))
    last_name: Mapped[str] = mapped_column(String(32))
    mobile_number: Mapped[str] = mapped_column(String(20))
    email_address: Mapped[str] = mapped_column(String(64))
    company: Mapped[str] = mapped_column(String(32))
    date_joined: Mapped[datetime] = mapped_column(DateTime, nullable=False)
    banned: Mapped[bool] = mapped_column(Boolean)
    date_banned: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
    created: Mapped[datetime] = mapped_column(
        DateTime, default=func.now(), nullable=False
    )
    modified: Mapped[datetime] = mapped_column(DateTime, nullable=True)

The delete function below is to delete one record, expected function call would look like this delete_record(Players.player_uid, primary_key='123456')

def delete_record(target_col: Base, primary_key: str = None) -> None:
        """
        Deletes record from db where primary key matches
        :param target_col of table
        :param primary_key record
        :returns None
        """
        try:
            with Session(engine) as session:
    
                if primary_key is None:
                    insert_kwarg_err_msg = "None was passed to delete_record()"
                    logging.error(insert_kwarg_err_msg)
                    raise ValueError(insert_kwarg_err_msg)
                elif primary_key:
                    statement = delete(target_col.__class__).where(
                        target_col == primary_key
                    )
                    session.execute(statement)
                else:
                    base_err_msg = (
                        f'Incorrect primary key for {target_col}: "{primary_key}"'
                    )
                    logging.error(base_err_msg)
                    raise ValueError(base_err_msg)
                session.commit()
    
        except (ValueError | sqlalchemy.exc.DatabaseError) as err:
            logging.error(err)
            raise err

However when called I am getting this error:

TypeError: QueryableAttribute.__clause_element__() missing 1 required positional argument: 'self'

I have a feeling the execption is caused by below, it works when I replace target_col.__class__ with Players however i need this to scale to other classes within the Model hence why its being passed as an arg:

statement = delete(target_col.__class__).where(
                            target_col == primary_key
                        )

Thanks for your help!

Answer 1

target_col.__class__ is fetching the class of the column, but you want to fetch the class of the parent model of the column. You can get to this through the mapper, which is accessible as the column's parent attribute.

statement = delete(
                target_col.parent.entity
            ).where(
                target_col == primary_key
            )
)

With this style of deletion you should consider setting a synchronisation strategy.