Self-hosted employee directory (in-app management + CSV import)
Most sites have no external HR database, so add a self-hosted directory mode. - New employee_directory_mode setting: 'external' (default; read a separate HR DB, unchanged) or 'selfhosted' (app-owned table). - DirectoryEmployee model + directoryemployees table (migration 7d16). to_dict emits the same keys the external contract uses (SSO/First_Name/...), so both modes share one response shape and the frontend is unchanged. - Employee search / single / batch lookup branch on the mode. - Self-hosted-only management endpoints: list, create, update, delete, and CSV import (upsert by SSO). Guarded so they only work in self-hosted mode. - EmployeeDirectory.vue management page (Settings > Locations & Organization): table + search + pagination, add/edit/delete, CSV import (file or paste). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
cproudlock
@@ -7,16 +7,24 @@ displays (recognition wall), so they are not JWT-gated; keep them read-only and
|
||||
never return more than the directory fields below.
|
||||
"""
|
||||
|
||||
import csv
|
||||
import io
|
||||
import logging
|
||||
|
||||
from flask import Blueprint, request
|
||||
from flask_jwt_extended import jwt_required
|
||||
|
||||
from shopdb.api import (
|
||||
db,
|
||||
success_response,
|
||||
error_response,
|
||||
ErrorCodes,
|
||||
employee_connection,
|
||||
require_role,
|
||||
)
|
||||
from shopdb.core.models import Setting
|
||||
|
||||
from ..models import DirectoryEmployee
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -26,6 +34,22 @@ employees_bp = Blueprint('employees', __name__)
|
||||
_FIELDS = 'SSO, First_Name, Last_Name, Team, Role, Picture'
|
||||
|
||||
|
||||
def _selfhosted():
|
||||
"""True when the directory is the app-owned table, not an external HR DB."""
|
||||
row = Setting.query.filter_by(key='employee_directory_mode').first()
|
||||
return (row.value if row and row.value else 'external').lower() == 'selfhosted'
|
||||
|
||||
|
||||
def _require_selfhosted():
|
||||
"""Guard for management endpoints - only valid in self-hosted mode."""
|
||||
if not _selfhosted():
|
||||
return error_response(
|
||||
ErrorCodes.VALIDATION_ERROR,
|
||||
'Directory is in external mode; manage people in the source HR database.',
|
||||
http_code=400)
|
||||
return None
|
||||
|
||||
|
||||
@employees_bp.route('/search', methods=['GET'])
|
||||
def search_employees():
|
||||
"""
|
||||
@@ -44,6 +68,16 @@ def search_employees():
|
||||
'Search query must be at least 2 characters'
|
||||
)
|
||||
|
||||
if _selfhosted():
|
||||
term = f'%{query}%'
|
||||
rows = (DirectoryEmployee.query
|
||||
.filter(db.or_(DirectoryEmployee.firstname.ilike(term),
|
||||
DirectoryEmployee.lastname.ilike(term),
|
||||
db.cast(DirectoryEmployee.sso, db.String).ilike(term)))
|
||||
.order_by(DirectoryEmployee.lastname, DirectoryEmployee.firstname)
|
||||
.limit(limit).all())
|
||||
return success_response([e.to_dict() for e in rows])
|
||||
|
||||
try:
|
||||
conn = employee_connection()
|
||||
with conn.cursor() as cur:
|
||||
@@ -77,6 +111,13 @@ def lookup_employee(sso):
|
||||
'SSO must be numeric'
|
||||
)
|
||||
|
||||
if _selfhosted():
|
||||
emp = DirectoryEmployee.query.get(int(sso))
|
||||
if not emp:
|
||||
return error_response(ErrorCodes.NOT_FOUND,
|
||||
f'Employee with SSO {sso} not found', http_code=404)
|
||||
return success_response(emp.to_dict())
|
||||
|
||||
try:
|
||||
conn = employee_connection()
|
||||
with conn.cursor() as cur:
|
||||
@@ -121,6 +162,14 @@ def lookup_employees():
|
||||
'At least one valid SSO is required'
|
||||
)
|
||||
|
||||
if _selfhosted():
|
||||
rows = DirectoryEmployee.query.filter(
|
||||
DirectoryEmployee.sso.in_([int(s) for s in ssos])).all()
|
||||
employees = [e.to_dict() for e in rows]
|
||||
names = ', '.join(f"{e['First_Name'].strip()} {e['Last_Name'].strip()}"
|
||||
for e in employees)
|
||||
return success_response({'employees': employees, 'names': names})
|
||||
|
||||
try:
|
||||
conn = employee_connection()
|
||||
with conn.cursor() as cur:
|
||||
@@ -148,3 +197,143 @@ def lookup_employees():
|
||||
'Employee lookup failed',
|
||||
http_code=500
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Self-hosted directory management (only when directory_mode=selfhosted)
|
||||
# =============================================================================
|
||||
|
||||
@employees_bp.route('/directory', methods=['GET'])
|
||||
@jwt_required(optional=True)
|
||||
def list_directory():
|
||||
"""Full self-hosted directory (for the management page)."""
|
||||
guard = _require_selfhosted()
|
||||
if guard:
|
||||
return guard
|
||||
rows = (DirectoryEmployee.query
|
||||
.order_by(DirectoryEmployee.lastname, DirectoryEmployee.firstname).all())
|
||||
return success_response([e.to_dict() for e in rows])
|
||||
|
||||
|
||||
def _employee_from_payload(data):
|
||||
"""Build kwargs from a payload accepting either external-style (SSO,
|
||||
First_Name...) or plain (sso, firstname...) keys."""
|
||||
def pick(*keys):
|
||||
for key in keys:
|
||||
if data.get(key) not in (None, ''):
|
||||
return data.get(key)
|
||||
return None
|
||||
return {
|
||||
'sso': pick('sso', 'SSO'),
|
||||
'firstname': pick('firstname', 'First_Name'),
|
||||
'lastname': pick('lastname', 'Last_Name'),
|
||||
'team': pick('team', 'Team'),
|
||||
'role': pick('role', 'Role'),
|
||||
'picture': pick('picture', 'Picture'),
|
||||
}
|
||||
|
||||
|
||||
@employees_bp.route('/directory', methods=['POST'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def create_directory_employee():
|
||||
guard = _require_selfhosted()
|
||||
if guard:
|
||||
return guard
|
||||
fields = _employee_from_payload(request.get_json() or {})
|
||||
if not (fields['sso'] and fields['firstname'] and fields['lastname']):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'sso, firstname and lastname are required')
|
||||
try:
|
||||
sso = int(fields['sso'])
|
||||
except (ValueError, TypeError):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'sso must be numeric')
|
||||
if DirectoryEmployee.query.get(sso):
|
||||
return error_response(ErrorCodes.CONFLICT, f'SSO {sso} already exists', http_code=409)
|
||||
emp = DirectoryEmployee(sso=sso, firstname=fields['firstname'], lastname=fields['lastname'],
|
||||
team=fields['team'], role=fields['role'], picture=fields['picture'])
|
||||
db.session.add(emp)
|
||||
db.session.commit()
|
||||
return success_response(emp.to_dict(), message='Employee added', http_code=201)
|
||||
|
||||
|
||||
@employees_bp.route('/directory/<int:sso>', methods=['PUT'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def update_directory_employee(sso):
|
||||
guard = _require_selfhosted()
|
||||
if guard:
|
||||
return guard
|
||||
emp = DirectoryEmployee.query.get(sso)
|
||||
if not emp:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'Employee not found', http_code=404)
|
||||
fields = _employee_from_payload(request.get_json() or {})
|
||||
if fields['firstname']:
|
||||
emp.firstname = fields['firstname']
|
||||
if fields['lastname']:
|
||||
emp.lastname = fields['lastname']
|
||||
for key in ('team', 'role', 'picture'):
|
||||
if key in (request.get_json() or {}) or fields[key] is not None:
|
||||
setattr(emp, key, fields[key])
|
||||
db.session.commit()
|
||||
return success_response(emp.to_dict(), message='Employee updated')
|
||||
|
||||
|
||||
@employees_bp.route('/directory/<int:sso>', methods=['DELETE'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def delete_directory_employee(sso):
|
||||
guard = _require_selfhosted()
|
||||
if guard:
|
||||
return guard
|
||||
emp = DirectoryEmployee.query.get(sso)
|
||||
if not emp:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'Employee not found', http_code=404)
|
||||
db.session.delete(emp)
|
||||
db.session.commit()
|
||||
return success_response(message='Employee removed')
|
||||
|
||||
|
||||
@employees_bp.route('/directory/import', methods=['POST'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def import_directory():
|
||||
"""Bulk upsert from CSV. Accepts headers SSO,First_Name,Last_Name,Team,Role,
|
||||
Picture (case-insensitive; sso/firstname/... also accepted)."""
|
||||
guard = _require_selfhosted()
|
||||
if guard:
|
||||
return guard
|
||||
text = ''
|
||||
if 'file' in request.files:
|
||||
text = request.files['file'].read().decode('utf-8-sig', errors='replace')
|
||||
else:
|
||||
data = request.get_json(silent=True) or {}
|
||||
text = data.get('csv', '')
|
||||
if not text.strip():
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'No CSV provided')
|
||||
|
||||
reader = csv.DictReader(io.StringIO(text))
|
||||
# Normalize headers to lower for tolerant matching.
|
||||
added = updated = skipped = 0
|
||||
for raw in reader:
|
||||
row = {(k or '').strip().lower(): (v or '').strip() for k, v in raw.items()}
|
||||
sso_raw = row.get('sso') or row.get('sso ')
|
||||
first = row.get('first_name') or row.get('firstname')
|
||||
last = row.get('last_name') or row.get('lastname')
|
||||
if not (sso_raw and sso_raw.isdigit() and first and last):
|
||||
skipped += 1
|
||||
continue
|
||||
sso = int(sso_raw)
|
||||
team = row.get('team') or None
|
||||
role = row.get('role') or None
|
||||
picture = row.get('picture') or None
|
||||
emp = DirectoryEmployee.query.get(sso)
|
||||
if emp:
|
||||
emp.firstname, emp.lastname, emp.team, emp.role, emp.picture = first, last, team, role, picture
|
||||
updated += 1
|
||||
else:
|
||||
db.session.add(DirectoryEmployee(sso=sso, firstname=first, lastname=last,
|
||||
team=team, role=role, picture=picture))
|
||||
added += 1
|
||||
db.session.commit()
|
||||
return success_response({'added': added, 'updated': updated, 'skipped': skipped},
|
||||
message=f'Import done: {added} added, {updated} updated, {skipped} skipped.')
|
||||
|
||||
Reference in New Issue
Block a user