Mateusz Gruszczyński
72 lines
3.9 KiB
Python
72 lines
3.9 KiB
Python
from collections import defaultdict
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
|
|
from flask import current_app, jsonify, request, url_for
|
|
from flask_login import current_user, login_required, login_user
|
|
from sqlalchemy import extract
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from . import api
|
|
from .common import accessible_vehicles, boolv, fail, payload, response, role_required, station_dict, user_dict, vehicle_dict
|
|
from .. import THEMES
|
|
from ..api_tokens import create_access_token
|
|
from ..extensions import db
|
|
from ..models import AppSetting, CompanySettings, FuelCard, FuelCardPolicy, FuelCardStationRule, FuelEntry, FuelStationCompany, FuelStationPoint, OrlenPrice, User, Vehicle
|
|
from ..services import POLISH_REGIONS, aggregate_ure_companies, calculate_costs, fetch_orlen_price, fetch_orlen_range, fetch_ure_stations, invoice_period
|
|
from ..station_catalog import sync_station_catalog
|
|
|
|
FUEL_TYPES = ("PB95", "PB98", "DIESEL", "LPG")
|
|
|
|
@api.get('/users')
|
|
@role_required('admin')
|
|
def users():
|
|
page=request.args.get('page',1,type=int);per=min(request.args.get('per_page',25,type=int),100);q=request.args.get('q','').strip();query=User.query
|
|
if q: query=query.filter(db.or_(User.name.ilike(f'%{q}%'),User.email.ilike(f'%{q}%')))
|
|
p=query.order_by(User.name).paginate(page=page,per_page=per,error_out=False);return response({'items':[user_dict(x) for x in p.items],'page':p.page,'pages':p.pages,'total':p.total})
|
|
|
|
@api.post('/users')
|
|
@role_required('boss','admin')
|
|
def create_user():
|
|
d=payload()
|
|
try:
|
|
if len(d.get('password',''))<8:raise ValueError('Hasło musi mieć co najmniej 8 znaków')
|
|
company_id=int(d.get('company_id') or current_user.company_id or CompanySettings.query.first().id)
|
|
if current_user.role=='boss' and company_id!=current_user.company_id:return fail('Brak uprawnień',403)
|
|
u=User(name=d['user_name'].strip(),email=d['email'].strip().lower(),role=d.get('role','user'),company_id=company_id);u.set_password(d['password']);db.session.add(u);db.session.commit();return response(user_dict(u),'Dodano użytkownika',201)
|
|
except Exception as exc:db.session.rollback();return fail(str(getattr(exc,'orig',exc)))
|
|
|
|
@api.put('/users/<int:user_id>')
|
|
@role_required('admin')
|
|
def update_user(user_id):
|
|
u=User.query.get_or_404(user_id);d=payload()
|
|
try:
|
|
u.name=d['name'].strip();u.email=d['email'].strip().lower();u.role=d['role'];u.active=boolv(d,'active');u.company_id=int(d.get('company_id') or u.company_id or CompanySettings.query.first().id);u.fuel_card_id=int(d['fuel_card_id']) if d.get('fuel_card_id') else None;pw=d.get('password','')
|
|
if pw:
|
|
if len(pw)<8:raise ValueError('Nowe hasło musi mieć co najmniej 8 znaków')
|
|
u.set_password(pw)
|
|
db.session.commit();return response(user_dict(u),'Zaktualizowano użytkownika')
|
|
except Exception as exc:db.session.rollback();return fail(str(getattr(exc,'orig',exc)))
|
|
|
|
@api.delete('/users/<int:user_id>')
|
|
@role_required('admin')
|
|
def delete_user(user_id):
|
|
user = User.query.get_or_404(user_id)
|
|
if user.id == current_user.id:
|
|
return fail('Nie możesz usunąć własnego konta administratora', 400)
|
|
if user.role == 'admin':
|
|
remaining_active_admins = User.query.filter(
|
|
User.role == 'admin', User.active.is_(True), User.id != user.id
|
|
).count()
|
|
if remaining_active_admins < 1:
|
|
return fail('Nie można usunąć ostatniego aktywnego administratora', 409)
|
|
if user.owned_vehicles:
|
|
return fail('Najpierw przypisz pojazdy tego użytkownika innemu właścicielowi', 409)
|
|
if FuelEntry.query.filter_by(user_id=user.id).first():
|
|
return fail('Użytkownik ma historię tankowań i nie może zostać usunięty. Możesz go dezaktywować.', 409)
|
|
user.shared_vehicles.clear()
|
|
user.favorite_stations.clear()
|
|
db.session.delete(user)
|
|
db.session.commit()
|
|
return response(message='Usunięto użytkownika')
|