b516b9b771
- GET /settings now masks password/token values (were returned in plaintext to anonymous callers); sending the mask back on update is a no-op so the real secret is never clobbered. - Move the employee-directory DB credentials out of source into env-backed config (shopdb.utils.employee_db); employees + notification recognition use the shared helper. Employee lookups stop leaking exception strings. - Fix low-supplies report using loc.location instead of loc.locationname. Employee/notification read endpoints stay unauthenticated by design (public shopfloor kiosk displays consume them). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
145 lines
4.2 KiB
Python
145 lines
4.2 KiB
Python
"""Employee lookup API endpoints.
|
|
|
|
These read from the separate employee directory DB (see shopdb.utils.employee_db).
|
|
They are intentionally reachable by the unauthenticated shopfloor kiosk displays
|
|
(recognition wall), so they are not JWT-gated; keep them read-only and never
|
|
return more than the directory fields below.
|
|
"""
|
|
|
|
import logging
|
|
|
|
from flask import Blueprint, request
|
|
from shopdb.utils.responses import success_response, error_response, ErrorCodes
|
|
from shopdb.utils.employee_db import employee_connection
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
employees_bp = Blueprint('employees', __name__)
|
|
|
|
# Columns safe to expose to the directory/recognition UI
|
|
_FIELDS = 'SSO, First_Name, Last_Name, Team, Role, Picture'
|
|
|
|
|
|
@employees_bp.route('/search', methods=['GET'])
|
|
def search_employees():
|
|
"""
|
|
Search employees by name.
|
|
|
|
Query parameters:
|
|
- q: Search query (searches first and last name)
|
|
- limit: Max results (default 10)
|
|
"""
|
|
query = request.args.get('q', '').strip()
|
|
limit = min(int(request.args.get('limit', 10)), 50)
|
|
|
|
if len(query) < 2:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Search query must be at least 2 characters'
|
|
)
|
|
|
|
try:
|
|
conn = employee_connection()
|
|
with conn.cursor() as cur:
|
|
cur.execute(f'''
|
|
SELECT {_FIELDS}
|
|
FROM employees
|
|
WHERE First_Name LIKE %s
|
|
OR Last_Name LIKE %s
|
|
OR CAST(SSO AS CHAR) LIKE %s
|
|
ORDER BY Last_Name, First_Name
|
|
LIMIT %s
|
|
''', (f'%{query}%', f'%{query}%', f'%{query}%', limit))
|
|
employees = cur.fetchall()
|
|
conn.close()
|
|
return success_response(employees)
|
|
except Exception:
|
|
logger.exception('Employee search failed')
|
|
return error_response(
|
|
ErrorCodes.DATABASE_ERROR,
|
|
'Employee lookup failed',
|
|
http_code=500
|
|
)
|
|
|
|
|
|
@employees_bp.route('/lookup/<sso>', methods=['GET'])
|
|
def lookup_employee(sso):
|
|
"""Look up a single employee by SSO."""
|
|
if not sso.isdigit():
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'SSO must be numeric'
|
|
)
|
|
|
|
try:
|
|
conn = employee_connection()
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
f'SELECT {_FIELDS} FROM employees WHERE SSO = %s',
|
|
(int(sso),)
|
|
)
|
|
employee = cur.fetchone()
|
|
conn.close()
|
|
|
|
if not employee:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Employee with SSO {sso} not found',
|
|
http_code=404
|
|
)
|
|
|
|
return success_response(employee)
|
|
except Exception:
|
|
logger.exception('Employee lookup failed for SSO %s', sso)
|
|
return error_response(
|
|
ErrorCodes.DATABASE_ERROR,
|
|
'Employee lookup failed',
|
|
http_code=500
|
|
)
|
|
|
|
|
|
@employees_bp.route('/lookup', methods=['GET'])
|
|
def lookup_employees():
|
|
"""
|
|
Look up multiple employees by SSO list.
|
|
|
|
Query parameters:
|
|
- sso: Comma-separated list of SSOs
|
|
"""
|
|
sso_list = request.args.get('sso', '')
|
|
ssos = [s.strip() for s in sso_list.split(',') if s.strip().isdigit()]
|
|
|
|
if not ssos:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'At least one valid SSO is required'
|
|
)
|
|
|
|
try:
|
|
conn = employee_connection()
|
|
with conn.cursor() as cur:
|
|
placeholders = ','.join(['%s'] * len(ssos))
|
|
cur.execute(
|
|
f'SELECT {_FIELDS} FROM employees WHERE SSO IN ({placeholders})',
|
|
[int(s) for s in ssos]
|
|
)
|
|
employees = cur.fetchall()
|
|
conn.close()
|
|
|
|
names = ', '.join(
|
|
f"{e['First_Name'].strip()} {e['Last_Name'].strip()}"
|
|
for e in employees
|
|
)
|
|
|
|
return success_response({
|
|
'employees': employees,
|
|
'names': names
|
|
})
|
|
except Exception:
|
|
logger.exception('Employee multi-lookup failed')
|
|
return error_response(
|
|
ErrorCodes.DATABASE_ERROR,
|
|
'Employee lookup failed',
|
|
http_code=500
|
|
)
|