How to delete all users with specified role

Issue

These are my User and Role models:

class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    firstname = db.Column(db.String(100), nullable=False, server_default='')
    lastname = db.Column(db.String(100), nullable=False, server_default='')
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(80), unique=True, nullable=False)
    password = db.Column(db.String(200), nullable=False)
    answers = db.relationship('Answer', backref="examinee", lazy=True)
    date_created = db.Column(db.DateTime, default=datetime.utcnow)
    active = db.Column('is_active', db.Boolean(), nullable=False, server_default='0')
    roles = db.relationship('Role', secondary='user_roles', backref=db.backref('users', lazy='dynamic'))

class Role(db.Model, RoleMixin):
    id = db.Column(db.Integer(), primary_key=True, autoincrement=True)
    name = db.Column(db.String(50), unique=True)
    description = db.Column(db.String(255))

class UserRoles(db.Model):
    id = db.Column(db.Integer(), primary_key=True, autoincrement=True)
    user_id = db.Column(db.Integer(), db.ForeignKey('user.id', ondelete='CASCADE'))
    role_id = db.Column(db.Integer(), db.ForeignKey('role.id', ondelete='CASCADE'))
    UniqueConstraint('user_id', 'role_id', name='user_role_uniqueness')

How can I remove all users with specific role such as ‘student’?

I have tried the following snippet which did not work:

@app.route('/removeAllStudentUsers', methods=['GET', 'POST'])
def remove_all_student_users():
    default_role_name = "student"
    default_role = Role.query.filter(Role.name == default_role_name).first()
    User.query.filter().delete(and_(User.roles.contains(default_role), func.count(User.roles) == 1)) # this line causes the error
    db.session.commit()
    return redirect(url_for('users'))

The above code causes this error:

sqlalchemy.exc.ArgumentError: Valid strategies for session synchronization are ‘evaluate’, ‘fetch’, False

Solution

There are multiple issues with your attempt to achieve that, including the fact that you might need a group by in order to check the count of the roles.

I would implement this a following:

_ = (
    User.query
    .filter(User.roles.any(Role.name == default_role_name))   # 1
    .filter(~User.roles.any(Role.name != default_role_name))  # 2
    .delete(synchronize_session=False)  # 3
)

where:

1. makes sure to return only those User who has the given role
2. makes sure to exclude the User who has also other roles (this is what you try to solve with func.count)
3. this is the solution for the error you get.

Answered By – van

This Answer collected from stackoverflow, is licensed under cc by-sa 2.5 , cc by-sa 3.0 and cc by-sa 4.0