shopdb-flask/shopdb/core/api/locations.py

"""Locations API endpoints - Full CRUD."""

from flask import Blueprint, request
from flask_jwt_extended import jwt_required

from shopdb.extensions import db
from shopdb.core.models import Location, LocationType
from shopdb.utils.responses import (
    success_response,
    error_response,
    paginated_response,
    ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query

from shopdb.utils.authz import require_role
from shopdb.utils.import_mode import apply_import_timestamps

locations_bp = Blueprint('locations', __name__)


def _loc_type_dict(t):
    return {
        'locationtypeid': t.locationtypeid,
        'locationtype': t.locationtype,
        'description': t.description,
        'color': t.color,
        'isactive': t.isactive,
    }


@locations_bp.route('/types', methods=['GET'])
@jwt_required(optional=True)
def list_location_types():
    """List location types. ?active=false includes inactive ones."""
    query = LocationType.query
    if request.args.get('active', 'true').lower() != 'false':
        query = query.filter_by(isactive=True)
    types = query.order_by(LocationType.locationtype).all()
    return success_response([_loc_type_dict(t) for t in types])


@locations_bp.route('/types', methods=['POST'])
@jwt_required()
@require_role('admin')
def create_location_type():
    data = request.get_json() or {}
    if not data.get('locationtype'):
        return error_response(ErrorCodes.VALIDATION_ERROR, 'locationtype is required')
    existing = LocationType.query.filter_by(locationtype=data['locationtype']).first()
    if existing:
        if not existing.isactive:
            existing.isactive = True
            for key in ('description', 'color'):
                if data.get(key) is not None:
                    setattr(existing, key, data[key])
            db.session.commit()
            return success_response(_loc_type_dict(existing), message='Reactivated existing type')
        return error_response(ErrorCodes.CONFLICT,
                              f"Location type '{data['locationtype']}' already exists", http_code=409)
    t = LocationType(locationtype=data['locationtype'],
                     description=data.get('description'), color=data.get('color'))
    db.session.add(t)
    db.session.commit()
    return success_response(_loc_type_dict(t), message='Location type created', http_code=201)


@locations_bp.route('/types/<int:type_id>', methods=['PUT'])
@jwt_required()
@require_role('admin')
def update_location_type(type_id):
    t = db.session.get(LocationType, type_id)
    if not t:
        return error_response(ErrorCodes.NOT_FOUND, 'Location type not found', http_code=404)
    data = request.get_json() or {}
    if 'locationtype' in data and data['locationtype'] != t.locationtype:
        if LocationType.query.filter_by(locationtype=data['locationtype']).first():
            return error_response(ErrorCodes.CONFLICT,
                                  f"Location type '{data['locationtype']}' already exists", http_code=409)
    for key in ('locationtype', 'description', 'color', 'isactive'):
        if key in data:
            setattr(t, key, data[key])
    db.session.commit()
    return success_response(_loc_type_dict(t), message='Location type updated')


@locations_bp.route('/types/<int:type_id>', methods=['DELETE'])
@jwt_required()
@require_role('admin')
def delete_location_type(type_id):
    t = db.session.get(LocationType, type_id)
    if not t:
        return error_response(ErrorCodes.NOT_FOUND, 'Location type not found', http_code=404)
    inuse = Location.query.filter_by(locationtypeid=type_id).count()
    if inuse:
        return error_response(ErrorCodes.CONFLICT,
                              f"Cannot delete: {inuse} location(s) still use this type", http_code=409)
    db.session.delete(t)
    db.session.commit()
    return success_response(message='Location type deleted')


@locations_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_locations():
    """List all locations."""
    page, per_page = get_pagination_params(request)

    query = Location.query

    if request.args.get('active', 'true').lower() != 'false':
        query = query.filter(Location.isactive == True)

    # Exact-match natural-key lookup for idempotent import (location name).
    if exactname := request.args.get('locationname'):
        query = query.filter(Location.locationname == exactname)

    if search := request.args.get('search'):
        query = query.filter(
            db.or_(
                Location.locationname.ilike(f'%{search}%'),
                Location.building.ilike(f'%{search}%')
            )
        )

    query = query.order_by(Location.locationname)

    items, total = paginate_query(query, page, per_page)
    data = [loc.to_dict() for loc in items]

    return paginated_response(data, page, per_page, total)


@locations_bp.route('/<int:location_id>', methods=['GET'])
@jwt_required(optional=True)
def get_location(location_id: int):
    """Get a single location."""
    loc = db.session.get(Location, location_id)

    if not loc:
        return error_response(
            ErrorCodes.NOT_FOUND,
            f'Location with ID {location_id} not found',
            http_code=404
        )

    return success_response(loc.to_dict())


@locations_bp.route('', methods=['POST'])
@jwt_required()
@require_role('admin')
def create_location():
    """Create a new location."""
    data = request.get_json()

    if not data or not data.get('locationname'):
        return error_response(ErrorCodes.VALIDATION_ERROR, 'locationname is required')

    if Location.query.filter_by(locationname=data['locationname']).first():
        return error_response(
            ErrorCodes.CONFLICT,
            f"Location '{data['locationname']}' already exists",
            http_code=409
        )

    loc = Location(
        locationname=data['locationname'],
        building=data.get('building'),
        floor=data.get('floor'),
        room=data.get('room'),
        description=data.get('description'),
        locationtypeid=data.get('locationtypeid'),
        parentlocationid=data.get('parentlocationid'),
        mapimage=data.get('mapimage'),
        mapwidth=data.get('mapwidth'),
        mapheight=data.get('mapheight')
    )

    db.session.add(loc)
    apply_import_timestamps(loc, data)
    db.session.commit()

    return success_response(loc.to_dict(), message='Location created', http_code=201)


@locations_bp.route('/<int:location_id>', methods=['PUT'])
@jwt_required()
@require_role('admin')
def update_location(location_id: int):
    """Update a location."""
    loc = db.session.get(Location, location_id)

    if not loc:
        return error_response(
            ErrorCodes.NOT_FOUND,
            f'Location with ID {location_id} not found',
            http_code=404
        )

    data = request.get_json()
    if not data:
        return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')

    if 'locationname' in data and data['locationname'] != loc.locationname:
        if Location.query.filter_by(locationname=data['locationname']).first():
            return error_response(
                ErrorCodes.CONFLICT,
                f"Location '{data['locationname']}' already exists",
                http_code=409
            )

    for key in ['locationname', 'building', 'floor', 'room', 'description',
                'locationtypeid', 'parentlocationid', 'mapimage', 'mapwidth',
                'mapheight', 'isactive']:
        if key in data:
            setattr(loc, key, data[key])

    apply_import_timestamps(loc, data)
    db.session.commit()
    return success_response(loc.to_dict(), message='Location updated')


@locations_bp.route('/<int:location_id>', methods=['DELETE'])
@jwt_required()
@require_role('admin')
def delete_location(location_id: int):
    """Delete (deactivate) a location."""
    loc = db.session.get(Location, location_id)

    if not loc:
        return error_response(
            ErrorCodes.NOT_FOUND,
            f'Location with ID {location_id} not found',
            http_code=404
        )

    loc.isactive = False
    db.session.commit()

    return success_response(message='Location deleted')