nopaque/app/namespaces/users.py

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:
        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:
        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:
        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:
        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'
        }