nopaque/app/blueprints/users/events.py

from flask import current_app, Flask
from flask_login import current_user
from flask_socketio import join_room, leave_room
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()


@socketio.on('users.delete')
@socketio_login_required
def delete_user(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'
    }


@socketio.on('users.get')
@socketio_login_required
def get_user(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.on('users.subscribe')
@socketio_login_required
def subscribe_user(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.on('users.unsubscribe')
@socketio_login_required
def unsubscribe_user(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'}