115 lines
3.1 KiB
Python
115 lines
3.1 KiB
Python
import json
|
|
|
|
from flask import Blueprint, request, make_response, jsonify
|
|
from flask_jwt_extended import get_jwt_identity, verify_jwt_in_request
|
|
from werkzeug.security import generate_password_hash
|
|
|
|
import database
|
|
import sipro.api.general
|
|
|
|
admin_blueprint = Blueprint('admin', __name__)
|
|
|
|
|
|
@admin_blueprint.before_request
|
|
def admin_check():
|
|
if request.method == 'OPTIONS':
|
|
return
|
|
if not verify_jwt_in_request(optional=True):
|
|
return {'error': 'Unauthorized'}, 401
|
|
|
|
user_id = get_jwt_identity()
|
|
is_admin = database.db.session.get(database.User, user_id).is_admin
|
|
if not is_admin:
|
|
return {'error': 'Unauthorized'}, 401
|
|
|
|
|
|
def format_user(user: database.User):
|
|
return {
|
|
'id': user.id,
|
|
'login': user.login,
|
|
'city_id': user.city_id,
|
|
'is_admin': user.is_admin
|
|
}
|
|
|
|
|
|
@admin_blueprint.get('/ping')
|
|
def ping():
|
|
return {"response": "pong"}
|
|
|
|
|
|
@admin_blueprint.get('/user')
|
|
def get_users():
|
|
response = make_response(jsonify(
|
|
[format_user(user) for user in database.User.query.all()]
|
|
))
|
|
|
|
response.headers['Content-Range'] = 'user 0-1/1'
|
|
return response
|
|
|
|
|
|
@admin_blueprint.get('/user/<int:user_id>')
|
|
def get_user(user_id):
|
|
user = database.db.session.get(database.User, user_id)
|
|
return format_user(user)
|
|
|
|
|
|
@admin_blueprint.put('/user/<int:user_id>')
|
|
def put_user(user_id):
|
|
params: dict = request.json
|
|
password = params.get('password')
|
|
if password:
|
|
password = password.strip()
|
|
if password:
|
|
params['password_hash'] = generate_password_hash(password)
|
|
|
|
if 'password' in params:
|
|
del params['password']
|
|
database.db.session.bulk_update_mappings(database.User, [params])
|
|
database.db.session.commit()
|
|
user = database.db.session.get(database.User, user_id)
|
|
return format_user(user)
|
|
|
|
|
|
@admin_blueprint.post('/user')
|
|
def create_user():
|
|
args: dict = request.json
|
|
args['password_hash'] = generate_password_hash(args['password'])
|
|
del args['password']
|
|
new_user = database.User(**args)
|
|
database.db.session.add(new_user)
|
|
database.db.session.flush()
|
|
database.db.session.commit()
|
|
return format_user(new_user)
|
|
|
|
|
|
@admin_blueprint.delete('/user/<int:user_id>')
|
|
def delete_user(user_id):
|
|
user = database.db.session.get(database.User, user_id)
|
|
if user:
|
|
database.db.session.delete(user)
|
|
database.db.session.commit()
|
|
return '', 200
|
|
|
|
|
|
@admin_blueprint.get('/city')
|
|
def get_cities():
|
|
filters = request.args.get('filter')
|
|
if filters:
|
|
filters = json.loads(filters)
|
|
else:
|
|
filters = {}
|
|
cities = sipro.api.general.get_cities()
|
|
for key, value in filters.items():
|
|
match key:
|
|
case 'id':
|
|
cities = list(filter(lambda city: city['id'] in value, cities))
|
|
total_cities = len(cities)
|
|
range_start, range_end = 0, total_cities - 1
|
|
range_raw = request.args.get('range')
|
|
if range_raw:
|
|
range_start, range_end = json.loads(range_raw)
|
|
cities = cities[range_start:range_end + 1]
|
|
response = make_response(jsonify(cities))
|
|
response.headers['Content-Range'] = f'city {range_start}-{range_end}/{total_cities}'
|
|
return response
|