Extract employee directory into a plugin

Third core feature pulled into a plugin (blueprint-only, like slides). The
employee directory is a read-only lookup over a separate HR database.

- plugins/employees/: manifest (api_prefix /api/employees, no deps), api/ (moved
  blueprint, contract-pure: success/error/ErrorCodes + employee_connection all
  from shopdb.api), plugin.py (get_blueprint, get_models -> []).
- employee_connection STAYS core infrastructure in shopdb.api (config-driven
  external DB connector, shared by search + the notifications shopfloor feed). So
  no get_services needed and no contract change - the plugin owns the directory
  FEATURE, core owns the shared connector.
- Fixed a latent bug in the move: error paths used ErrorCodes.DATABASE_ERROR
  which does not exist -> ErrorCodes.INTERNAL_ERROR (so a directory outage now
  returns a clean 500 envelope instead of an AttributeError crash).
- De-cored: deleted shopdb/core/api/employees.py, removed from
  CORE_BLUEPRINT_NAMES + core/api/__init__ import/__all__. Registered in
  instance/plugins.json.

Pinned first: validation (400) + graceful-degrade (500) characterization tests;
the degrade test caught the DATABASE_ERROR bug and goes green with the fix.
184 tests pass, naming green, app boots 9 bundled plugins, endpoint verified live.

Plugin extractions complete: knowledgebase, slides, employees.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-06-26 21:13:15 -04:00
parent b6ec4cb577
commit d20682fd06
10 changed files with 128 additions and 13 deletions

View File

@@ -14,7 +14,6 @@ from .applications import applications_bp
from .search import search_bp
from .reports import reports_bp
from .collector import collector_bp
from .employees import employees_bp
from .settings import settings_bp
from .auditlogs import auditlogs_bp
from .users import users_bp
@@ -34,7 +33,6 @@ __all__ = [
'search_bp',
'reports_bp',
'collector_bp',
'employees_bp',
'settings_bp',
'auditlogs_bp',
'users_bp',

View File

@@ -1,144 +0,0 @@
"""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
)