Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions api/ai_readiness.py
Original file line number Diff line number Diff line change
Expand Up @@ -308,12 +308,16 @@ def generate_data_card(
- Class balance (outlier/calibration ratios)
- Temporal coverage (runs per month)
- Known limitations / bias warnings
Every statistic covers only the runs the caller can see (row-level
security on etcher_runs: their projects plus open projects; admins see
all). Without a user id an admin call is a system job and reads all runs.
"""
from data_loader_pg import get_pg_worker_connection
from data_loader_pg import _scoped_connection
from psycopg2.extras import RealDictCursor

try:
conn = get_pg_worker_connection()
conn = _scoped_connection(nanohub_user_id, is_admin)

with conn.cursor(cursor_factory=RealDictCursor) as cur:
# ── Overview statistics ──────────────────────────────────────
Expand Down
17 changes: 11 additions & 6 deletions api/data_loader_pg.py
Original file line number Diff line number Diff line change
Expand Up @@ -5504,21 +5504,26 @@ def json_number(json_column: str, key: str) -> str:
return None


def get_etcher_run_stats_pg() -> Dict[str, Any]:
def get_etcher_run_stats_pg(viewer: Any = None) -> Dict[str, Any]:
"""
Total run count and most recent run timestamp for the etcher.
Always uses the worker connection so the equipment detail card shows
the *true* fleet-wide count, independent of the caller's RLS scope.
The etcher_runs table has no equipment_id column today — every row in it
belongs to the etcher — so a plain COUNT(*) is the correct answer.
For a non-admin ``viewer`` (a PlatformUser) the count covers only the
runs they can see (row-level security: their projects plus open
projects). Admins and internal callers get the fleet-wide count from the
worker connection. The etcher_runs table has no equipment_id column
today — every row in it belongs to the etcher — so a plain COUNT(*) is
the correct answer.
Returns ``{"total_runs": int, "last_data_at": iso_string | None}``.
Returns ``{"total_runs": None, "last_data_at": None}`` on error so the UI
can render "—" without crashing.
"""
try:
conn = get_pg_worker_connection()
if viewer is not None and getattr(viewer, "role", "") != "admin":
conn = get_pg_connection(viewer)
else:
conn = get_pg_worker_connection()
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"SELECT count(*) AS total_runs, max(run_date) AS last_data_at FROM etcher_runs"
Expand Down
209 changes: 165 additions & 44 deletions api/metadata_pg.py
Original file line number Diff line number Diff line change
Expand Up @@ -1194,6 +1194,48 @@ def equipment_visible_to_user(
return True


def equipment_manageable_by(record: dict[str, Any], user: PlatformUser) -> bool:
"""The owner, a trusted maintainer or an admin (same rule as editing)."""
if user.role == "admin" or (record.get("owner_id") or "") == user.id:
return True
maintainer_ids = set(record.get("_maintainer_ids") or [])
maintainer_ids.update(
maintainer.get("user_id")
for maintainer in record.get("maintainers") or []
if isinstance(maintainer, dict)
)
return user.id in maintainer_ids


_CONNECTION_FIELDS = ("connection_host", "connection_port", "connection_database")


def redact_equipment_for_viewer(
item: dict[str, Any],
user: PlatformUser,
*,
manageable: bool | None = None,
) -> dict[str, Any]:
"""Hide the data-source connection (host, port, database) from users who
can see the equipment but not manage it. ``manageable`` overrides the check
when the caller already knows (serialized list items carry no maintainers).
"""
if manageable is None:
manageable = equipment_manageable_by(item, user)
if manageable:
return item
redacted = dict(item)
for field in _CONNECTION_FIELDS:
if field in redacted:
redacted[field] = ""
config = redacted.get("config_json")
if isinstance(config, dict) and "source_connection" in config:
redacted["config_json"] = {
key: value for key, value in config.items() if key != "source_connection"
}
return redacted


def approved_equipment_available_to_user(
record: dict[str, Any],
user: PlatformUser,
Expand Down Expand Up @@ -1256,20 +1298,25 @@ def _serialize_equipment(
}


def _fetch_equipment_run_metrics(_conn=None) -> dict[str, dict[str, Any]]:
"""Aggregate fleet-wide run counts and latest-data timestamps per equipment.
def _fetch_equipment_run_metrics(
viewer: PlatformUser | None = None,
) -> dict[str, dict[str, Any]]:
"""Aggregate run counts and latest-data timestamps per equipment.
Combines the canonical etcher_runs (attributed via the project's equipment)
with the generic equipment_runs table (attributed via the run's own
equipment_id) so manually uploaded data for any registered equipment is
reflected in the inventory metrics.
Runs on the superuser connection so these non-sensitive aggregate counts are
complete (fleet-wide) regardless of the caller's project membership. The
RLS-scoped connection passed by callers would otherwise hide runs in
private/shared projects, even from members and admins, and undercount.
For a non-admin ``viewer`` the counts cover only the runs that user can
see (row-level security: their projects plus open projects), so private
projects' activity isn't revealed. Admins and internal callers (no
viewer) get the fleet-wide counts from the worker connection.
"""
conn = get_pg_worker_connection()
if viewer is not None and viewer.role != "admin":
conn = get_pg_connection(viewer)
else:
conn = get_pg_worker_connection()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
Expand Down Expand Up @@ -1359,23 +1406,32 @@ def list_equipment_pg(user: PlatformUser) -> list[dict[str, Any]]:
for arow in access_rows:
maintainer_map.setdefault(arow["equipment_id"], []).append(arow["user_id"])

metrics = _fetch_equipment_run_metrics(conn)
conn.commit()
metrics = _fetch_equipment_run_metrics(user)

visible: list[dict[str, Any]] = []
for row in rows:
record = dict(row)
record["_maintainer_ids"] = maintainer_map.get(record["domain_id"], [])
if equipment_visible_to_user(record, user):
visible.append(
_serialize_equipment(record, metrics.get(record["domain_id"]))
redact_equipment_for_viewer(
_serialize_equipment(record, metrics.get(record["domain_id"])),
user,
manageable=equipment_manageable_by(record, user),
)
)
return visible
finally:
conn.close()


def get_equipment_pg(domain_id: str) -> dict[str, Any] | None:
def get_equipment_pg(
domain_id: str,
viewer: PlatformUser | None = None,
) -> dict[str, Any] | None:
"""One equipment record. Pass ``viewer`` when the record is returned to a
user, so its run counts cover only runs that user can see."""
conn = get_pg_connection()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
Expand Down Expand Up @@ -1457,8 +1513,8 @@ def get_equipment_pg(domain_id: str) -> dict[str, Any] | None:
for r in maintainer_detail_rows
]

metrics = _fetch_equipment_run_metrics(conn)
conn.commit()
metrics = _fetch_equipment_run_metrics(viewer)
return _serialize_equipment(record, metrics.get(domain_id))
finally:
conn.close()
Expand Down Expand Up @@ -2456,9 +2512,12 @@ def _visible_experiment_type_filter(
"""
``AND ...`` clause limiting ``experiment_types`` (referenced as ``alias``)
to templates ``user`` may see or use: their own, shared or published ones,
their organization's, and templates attached to a project they can read.
Row-level security enforces the same rule in the database; this keeps it
when enforcement is off.
templates attached to a project they can read, and templates used by an
experiment they can read (theirs, in a project they can read, or an
execution request on equipment they own or maintain). A private template
is not visible to the rest of the owner's organization. Row-level security
(app_experiment_type_readable) enforces the same rule in the database;
this keeps it when enforcement is off.
"""
if user.role == "admin":
return "", []
Expand All @@ -2467,7 +2526,6 @@ def _visible_experiment_type_filter(
AND (
{alias}.owner_id = %s
OR {alias}.visibility IN ('shared', 'published')
OR ({alias}.owner_org <> '' AND {alias}.owner_org = %s)
OR EXISTS (
SELECT 1
FROM project_experiments vis_pe
Expand All @@ -2483,9 +2541,39 @@ def _visible_experiment_type_filter(
)
)
)
OR EXISTS (
SELECT 1
FROM experiment_definitions vis_e
LEFT JOIN projects vis_ep ON vis_ep.id = vis_e.project_id
LEFT JOIN equipment_metadata vis_em ON vis_em.domain_id = vis_e.equipment_id
WHERE vis_e.type_id = {alias}.id
AND (
vis_e.owner_id = %s
OR vis_ep.access_mode = 'open'
OR EXISTS (
SELECT 1
FROM project_members vis_epm
WHERE vis_epm.project_id = vis_e.project_id
AND vis_epm.nanohub_user_id = %s
)
OR (
vis_e.project_id IS NOT NULL
AND COALESCE(vis_e.execution_request_id, '') <> ''
AND (
vis_em.owner_id = %s
OR EXISTS (
SELECT 1
FROM equipment_access vis_ea
WHERE vis_ea.equipment_id = vis_e.equipment_id
AND vis_ea.user_id = %s
)
)
)
)
)
)
""",
[user.id, user.organization, user.id],
[user.id, user.id, user.id, user.id, user.id, user.id],
)


Expand Down Expand Up @@ -2581,7 +2669,7 @@ def list_experiment_types_pg(
params.append(equipment_id)
where_sql = f"WHERE {' AND '.join(conditions)}" if conditions else "WHERE true"
# Same rule as everywhere else templates are used (and as RLS): own,
# shared/published, same organization, or attached to a readable project.
# shared/published, or attached to a readable project.
visibility_sql, visibility_params = _visible_experiment_type_filter(user)
where_sql += visibility_sql
params.extend(visibility_params)
Expand Down Expand Up @@ -7004,17 +7092,48 @@ def list_experiment_definitions_pg(
conn.close()


def list_execution_queue_pg(user: PlatformUser) -> list[dict[str, Any]]:
"""List project-run execution requests visible to a facility manager."""
if user.role not in {"admin", "equipment_owner"}:
raise PermissionError("Facility execution queue access required")
_MANAGED_EQUIPMENT_SQL = """(
em.owner_id = %s
OR EXISTS (
SELECT 1 FROM equipment_access ea
WHERE ea.equipment_id = em.domain_id AND ea.user_id = %s
)
)"""


def user_manages_any_equipment_pg(user: PlatformUser) -> bool:
"""Whether ``user`` owns or is a trusted maintainer of any equipment, and
so has an execution queue (admins always do)."""
if user.role == "admin":
return True
conn = get_pg_connection(user)
try:
with conn.cursor() as cur:
cur.execute(
"""
SELECT
EXISTS (SELECT 1 FROM equipment_metadata WHERE owner_id = %s)
OR EXISTS (SELECT 1 FROM equipment_access WHERE user_id = %s)
""",
(user.id, user.id),
)
row = cur.fetchone()
conn.commit()
finally:
conn.close()
return bool(row and row[0])


def list_execution_queue_pg(user: PlatformUser) -> list[dict[str, Any]]:
"""List project-run execution requests for the equipment ``user`` owns
or is a trusted maintainer of (admins: all equipment). The same
organization is not enough. Mirrors app_manages_equipment."""
conn = get_pg_connection(user)
params: list[Any] = []
conditions = ["e.project_id IS NOT NULL", "e.execution_request_id <> ''"]
if user.role != "admin":
conditions.append("(em.owner_id = %s OR em.owner_org = %s)")
params.extend([user.id, user.organization])
conditions.append(_MANAGED_EQUIPMENT_SQL)
params.extend([user.id, user.id])

where_sql = " AND ".join(conditions)
try:
Expand Down Expand Up @@ -7144,8 +7263,6 @@ def update_execution_queue_status_pg(
scheduled_for: str,
user: PlatformUser,
) -> dict[str, Any]:
if user.role not in {"admin", "equipment_owner"}:
raise PermissionError("Facility execution queue access required")
if status not in EXECUTION_STATUSES:
raise ValueError("Invalid execution status")

Expand All @@ -7156,25 +7273,20 @@ def update_execution_queue_status_pg(
"""
SELECT
e.id::text,
em.owner_id AS equipment_owner_id,
em.owner_org AS equipment_owner_org
COALESCE({managed}, false) AS manages_equipment
FROM experiment_definitions e
LEFT JOIN equipment_metadata em ON em.domain_id = e.equipment_id
WHERE e.id = %s
AND e.project_id IS NOT NULL
""",
(experiment_id,),
""".format(managed=_MANAGED_EQUIPMENT_SQL),
(user.id, user.id, experiment_id),
)
existing = cur.fetchone()
if not existing:
return {"status": "not_found", "id": experiment_id}
if user.role != "admin":
allowed = (
existing.get("equipment_owner_id") == user.id
or existing.get("equipment_owner_org") == user.organization
)
if not allowed:
raise PermissionError("Not authorized to update this execution request")
if user.role != "admin" and not existing.get("manages_equipment"):
# Only the equipment's owner or a trusted maintainer.
raise PermissionError("Not authorized to update this execution request")

cur.execute(
"""
Expand Down Expand Up @@ -7579,7 +7691,18 @@ def create_upload_record_pg(
conn.close()


def list_uploads_pg(user: PlatformUser) -> list[dict[str, Any]]:
def list_uploads_pg(
user: PlatformUser,
*,
include_shared_db3: bool = False,
) -> list[dict[str, Any]]:
"""Uploads the caller may see: their own (admins see all).
Uploads hold raw files that are not yet in any project, so other users,
including those in the same organization, never see them. The temporary
DB3 pilot is the one exception: its participants share one queue, so
``include_shared_db3`` adds every .db3 upload for them.
"""
conn = get_pg_connection()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
Expand Down Expand Up @@ -7615,13 +7738,11 @@ def list_uploads_pg(user: PlatformUser) -> list[dict[str, Any]]:
uploads: list[dict[str, Any]] = []
for row in rows:
record = dict(row)
if user.role != "admin":
same_owner = record.get("owner_id") == user.id
same_org = (
record.get("owner_org")
and record.get("owner_org") == user.organization
)
if not same_owner and not same_org:
if user.role != "admin" and record.get("owner_id") != user.id:
shared_db3 = include_shared_db3 and str(
record.get("filename") or ""
).lower().endswith(".db3")
if not shared_db3:
continue

uploads.append(
Expand Down
Loading