(
self,
query: QueryT,
*,
usernames: Sequence[str] = (),
emails: Sequence[str] = (),
roles: Sequence[Role] = (),
has_ra_username: bool | None = None,
)
| 12 | |
| 13 | class DBUsersHandler(DBBaseHandler): |
| 14 | def filter[QueryT: Select[tuple[User]] | Update | Delete]( |
| 15 | self, |
| 16 | query: QueryT, |
| 17 | *, |
| 18 | usernames: Sequence[str] = (), |
| 19 | emails: Sequence[str] = (), |
| 20 | roles: Sequence[Role] = (), |
| 21 | has_ra_username: bool | None = None, |
| 22 | ) -> QueryT: |
| 23 | if usernames: |
| 24 | query = query.filter( |
| 25 | func.lower(User.username).in_([u.lower() for u in usernames]) |
| 26 | ) |
| 27 | if emails: |
| 28 | query = query.filter( |
| 29 | func.lower(User.email).in_([e.lower() for e in emails]) |
| 30 | ) |
| 31 | if roles: |
| 32 | query = query.filter(User.role.in_(roles)) |
| 33 | if has_ra_username is not None: |
| 34 | predicate = and_(User.ra_username != "", User.ra_username.isnot(None)) |
| 35 | if not has_ra_username: |
| 36 | predicate = not_(predicate) |
| 37 | query = query.filter(predicate) |
| 38 | return query |
| 39 | |
| 40 | @begin_session |
| 41 | def add_user( |
no outgoing calls
no test coverage detected