from flask_login import current_user from flask_socketio import join_room, leave_room, Namespace from app import hashids from app.decorators import socketio_login_required from app.models import User class UsersNamespace(Namespace): @socketio_login_required def on_get_user(self, user_hashid: str) -> dict: user_id = hashids.decode(user_hashid) if not isinstance(user_id, int): return {'code': 400, 'msg': '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_user(self, user_hashid: str) -> dict: user_id = hashids.decode(user_hashid) if not isinstance(user_id, int): return {'code': 400, 'msg': '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_user(self, user_hashid: str) -> dict: user_id = hashids.decode(user_hashid) if not isinstance(user_id, int): return {'code': 400, 'msg': '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'}