133 lines
4.5 KiB
Python
133 lines
4.5 KiB
Python
"""
|
|
app/utils/scope.py
|
|
------------------
|
|
Facility-scoping utilities for the Janitorial QC portal.
|
|
|
|
get_customer_scope(user) -> list[int] | None
|
|
Facility IDs a customer may access via CustomerAssignment rows.
|
|
|
|
get_inspector_scope(user) -> list[int] | None
|
|
Facility IDs an inspector may access via InspectorAssignment rows.
|
|
Applies to BOTH 'inspector' (internal) and 'external_inspector'
|
|
(customer / third-party) — see User.INSPECTOR_ROLES.
|
|
Returns [] (empty list) when the inspector has no contract assignments,
|
|
meaning they see nothing (strict mode).
|
|
|
|
For non-customer / non-inspector roles both functions return None, signalling
|
|
that no facility-level scoping is required (full access applies).
|
|
"""
|
|
|
|
import logging
|
|
from app import db
|
|
from app.models.project import CustomerAssignment
|
|
from app.models.facility import Facility
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_customer_scope(user) -> list[int] | None:
|
|
"""Return the list of facility IDs accessible to a customer user.
|
|
|
|
Parameters
|
|
----------
|
|
user : User
|
|
The currently authenticated user.
|
|
|
|
Returns
|
|
-------
|
|
list[int]
|
|
Facility IDs the customer may access. May be empty if no assignments
|
|
exist yet — callers should treat an empty list as "no access".
|
|
None
|
|
Returned for non-customer roles, indicating unrestricted access.
|
|
"""
|
|
if user.role != 'customer':
|
|
return None # no scoping needed for internal staff
|
|
|
|
# Select only the two columns needed. The previous .all() built full
|
|
# CustomerAssignment ORM objects (and their identity-map entries) purely to
|
|
# read two integers off each one; this function runs on nearly every
|
|
# request for a customer, sometimes more than once.
|
|
assignments = db.session.query(
|
|
CustomerAssignment.project_id,
|
|
CustomerAssignment.facility_id,
|
|
).filter(CustomerAssignment.user_id == user.id).all()
|
|
|
|
if not assignments:
|
|
return []
|
|
|
|
# Separate direct facility assignments from project-level assignments
|
|
direct_facility_ids = {fac_id for _, fac_id in assignments if fac_id}
|
|
project_ids = {proj_id for proj_id, fac_id in assignments if not fac_id}
|
|
|
|
facility_ids = set(direct_facility_ids)
|
|
|
|
# Single bulk query for all project-scoped facilities — replaces the
|
|
# previous per-assignment Facility.query loop (N+1 pattern). Only the id
|
|
# column is read; nothing here needs a hydrated Facility.
|
|
if project_ids:
|
|
facility_ids.update(
|
|
fid for (fid,) in db.session.query(Facility.id).filter(
|
|
Facility.project_id.in_(project_ids),
|
|
Facility.active == True,
|
|
).all()
|
|
)
|
|
|
|
logger.debug(
|
|
'SCOPE | customer_scope | user_id=%s username=%s facility_ids=%s',
|
|
user.id, user.username, sorted(facility_ids),
|
|
)
|
|
|
|
return sorted(facility_ids)
|
|
|
|
|
|
def get_inspector_scope(user) -> list[int] | None:
|
|
"""Return the list of facility IDs accessible to a contract-scoped inspector.
|
|
|
|
Parameters
|
|
----------
|
|
user : User
|
|
The currently authenticated user.
|
|
|
|
Returns
|
|
-------
|
|
list[int]
|
|
Facility IDs the inspector may access. An empty list means the
|
|
inspector has no contract assignments and should see nothing.
|
|
None
|
|
Returned for non-inspector roles, indicating unrestricted access.
|
|
"""
|
|
# phase49: covers BOTH 'inspector' and 'external_inspector'. An external
|
|
# (customer / third-party) inspector is scoped by exactly the same
|
|
# InspectorAssignment rows — the contracts an admin grants them.
|
|
from app.models.user import User
|
|
|
|
if user.role not in User.INSPECTOR_ROLES:
|
|
return None
|
|
|
|
from app.models.inspector_assignment import InspectorAssignment
|
|
|
|
# Column-only selects — see the note in get_customer_scope(). This runs on
|
|
# every scoped request for both inspector roles.
|
|
project_ids = [
|
|
pid for (pid,) in
|
|
db.session.query(InspectorAssignment.project_id)
|
|
.filter(InspectorAssignment.user_id == user.id).all()
|
|
]
|
|
|
|
if not project_ids:
|
|
return [] # strict: no assignments = no access
|
|
|
|
facility_ids = [
|
|
fid for (fid,) in db.session.query(Facility.id).filter(
|
|
Facility.project_id.in_(project_ids),
|
|
Facility.active == True,
|
|
).all()
|
|
]
|
|
|
|
logger.debug(
|
|
'SCOPE | inspector_scope | user_id=%s username=%s facility_ids=%s',
|
|
user.id, user.username, sorted(facility_ids),
|
|
)
|
|
|
|
return sorted(facility_ids) |