mirror of
https://gitlab.ub.uni-bielefeld.de/sfb1288inf/nopaque.git
synced 2024-12-26 11:24:18 +00:00
129 lines
3.6 KiB
Python
129 lines
3.6 KiB
Python
from flask import current_app, Flask
|
|
from flask_login import current_user
|
|
from flask_socketio import join_room, leave_room, Namespace
|
|
from app import db, hashids, socketio
|
|
from app.decorators import socketio_login_required
|
|
from app.models import User
|
|
|
|
|
|
def _delete_user(app: Flask, user_id: int):
|
|
with app.app_context():
|
|
user = User.query.get(user_id)
|
|
user.delete()
|
|
db.session.commit()
|
|
|
|
|
|
class UsersNamespace(Namespace):
|
|
@socketio_login_required
|
|
def on_get(self, user_hashid: str) -> dict:
|
|
if not isinstance(user_hashid, str):
|
|
return {'status': 400, 'statusText': 'Bad Request'}
|
|
|
|
user_id = hashids.decode(user_hashid)
|
|
|
|
if not isinstance(user_id, int):
|
|
return {'status': 400, 'statusText': 'Bad Request'}
|
|
|
|
user = User.query.get(user_id)
|
|
|
|
if user is None:
|
|
return {'status': 404, 'statusText': 'Not Found'}
|
|
|
|
if not (
|
|
user == current_user
|
|
or current_user.is_administrator
|
|
):
|
|
return {'status': 403, 'statusText': 'Forbidden'}
|
|
|
|
return {
|
|
'body': user.to_json_serializeable(
|
|
backrefs=True,
|
|
relationships=True
|
|
),
|
|
'status': 200,
|
|
'statusText': 'OK'
|
|
}
|
|
|
|
@socketio_login_required
|
|
def on_subscribe(self, user_hashid: str) -> dict:
|
|
if not isinstance(user_hashid, str):
|
|
return {'status': 400, 'statusText': 'Bad Request'}
|
|
|
|
user_id = hashids.decode(user_hashid)
|
|
|
|
if not isinstance(user_id, int):
|
|
return {'status': 400, 'statusText': 'Bad Request'}
|
|
|
|
user = User.query.get(user_id)
|
|
|
|
if user is None:
|
|
return {'status': 404, 'statusText': 'Not Found'}
|
|
|
|
if not (
|
|
user == current_user
|
|
or current_user.is_administrator
|
|
):
|
|
return {'status': 403, 'statusText': 'Forbidden'}
|
|
|
|
join_room(f'/users/{user.hashid}')
|
|
|
|
return {'status': 200, 'statusText': 'OK'}
|
|
|
|
@socketio_login_required
|
|
def on_unsubscribe(self, user_hashid: str) -> dict:
|
|
if not isinstance(user_hashid, str):
|
|
return {'status': 400, 'statusText': 'Bad Request'}
|
|
|
|
user_id = hashids.decode(user_hashid)
|
|
|
|
if not isinstance(user_id, int):
|
|
return {'status': 400, 'statusText': 'Bad Request'}
|
|
|
|
user = User.query.get(user_id)
|
|
|
|
if user is None:
|
|
return {'status': 404, 'statusText': 'Not Found'}
|
|
|
|
if not (
|
|
user == current_user
|
|
or current_user.is_administrator
|
|
):
|
|
return {'status': 403, 'statusText': 'Forbidden'}
|
|
|
|
leave_room(f'/users/{user.hashid}')
|
|
|
|
return {'status': 200, 'statusText': 'OK'}
|
|
|
|
@socketio_login_required
|
|
def on_delete(self, user_hashid: str) -> dict:
|
|
if not isinstance(user_hashid, str):
|
|
return {'status': 400, 'statusText': 'Bad Request'}
|
|
|
|
user_id = hashids.decode(user_hashid)
|
|
|
|
if not isinstance(user_id, int):
|
|
return {'status': 400, 'statusText': 'Bad Request'}
|
|
|
|
user = User.query.get(user_id)
|
|
|
|
if user is None:
|
|
return {'status': 404, 'statusText': 'Not Found'}
|
|
|
|
if not (
|
|
user == current_user
|
|
or current_user.is_administrator
|
|
):
|
|
return {'status': 403, 'statusText': 'Forbidden'}
|
|
|
|
socketio.start_background_task(
|
|
_delete_user,
|
|
current_app._get_current_object(),
|
|
user.id
|
|
)
|
|
|
|
return {
|
|
'body': f'User "{user.username}" marked for deletion',
|
|
'status': 202,
|
|
'statusText': 'Accepted'
|
|
}
|