diff --git a/.gitignore b/.gitignore
index a3786635..cd75f603 100644
--- a/.gitignore
+++ b/.gitignore
@@ -16,3 +16,4 @@ test_compute.sh
fastapi.log
compute-tests/
local.env
+test_filesystem.sh
\ No newline at end of file
diff --git a/Dockerfile b/Dockerfile
index 17d6f2dd..4e7ab23a 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -11,9 +11,11 @@ ENV IRI_API_ADAPTER_account="app.s3df.account_adapter.S3DFAccountAdapter"
ENV IRI_API_ADAPTER_status="app.s3df.status_adapter.S3DFStatusAdapter"
ENV IRI_API_ADAPTER_compute="app.s3df.compute_adapter.SLACComputeAdapter"
ENV IRI_API_ADAPTER_filesystem="app.s3df.filesystem_adapter.S3DFFilesystemAdapter"
+ENV IRI_API_ADAPTER_facility="app.s3df.facility_adapter.S3DFFacilityAdapter"
+ENV IRI_API_ADAPTER_task="app.s3df.task_adapter.S3DFTaskAdapter"
ENV IRI_SHOW_MISSING_ROUTES="true"
ENV DEX_JWKS_URL="https://dex.slac.stanford.edu/keys"
ENV DEX_ISSUER="https://dex.slac.stanford.edu"
-
+ENV API_URL_ROOT="https://sdf-iri-dev.slac.stanford.edu"
CMD ["fastapi", "run", "app/main.py", "--port", "8000"]
diff --git a/Makefile b/Makefile
index dae14644..388ab2c4 100644
--- a/Makefile
+++ b/Makefile
@@ -47,9 +47,9 @@ dev-s3df: deps
API_URL_ROOT='http://127.0.0.1:8000' fastapi dev
# --- Docker / GHCR targets ---
-GHCR_USERNAME ?= ""
+GHCR_USERNAME ?= slaclab
GHCR_IMAGE ?= ghcr.io/$(GHCR_USERNAME)/iri-s3df
-IMAGE_TAG ?= dev
+IMAGE_TAG ?= prod
docker-build:
docker build --platform linux/amd64 -t $(GHCR_IMAGE):$(IMAGE_TAG) .
diff --git a/app/config.py b/app/config.py
index 35c17c93..ebf176de 100644
--- a/app/config.py
+++ b/app/config.py
@@ -16,12 +16,14 @@
For more information, see: [https://iri.science/](https://iri.science/)
+
+
"""
# version is the openapi.json spec version
# /api/v1 mount point means it's the latest backward-compatible url
API_CONFIG = {
- "title": "IRI Facility API reference implementation",
+ "title": "SLAC IRI API implementation",
"description": description,
"version": API_VERSION,
"docs_url": "/",
@@ -36,7 +38,7 @@
logger.error(f"Error parsing IRI_API_PARAMS: {exc}")
-API_URL_ROOT = os.environ.get("API_URL_ROOT", "https://api.iri.nersc.gov")
+API_URL_ROOT = os.environ.get("API_URL_ROOT", "https://iri.slac.stanford.edu")
API_PREFIX = os.environ.get("API_PREFIX", "/")
API_URL = os.environ.get("API_URL", "api/v1")
diff --git a/app/main.py b/app/main.py
index 2019bece..92c662c3 100644
--- a/app/main.py
+++ b/app/main.py
@@ -21,7 +21,7 @@
from app.routers.task import task
from . import config
-from .request_context import set_api_url_base, _api_url_base
+from .request_context import set_api_url_base, set_auth_headers, _api_url_base
logging.basicConfig(
level=logging.INFO,
@@ -48,7 +48,7 @@
tracer = trace.get_tracer(__name__)
# ------------------------------------------------------------------
-APP = FastAPI(servers=[{"url": config.API_URL_ROOT}], **config.API_CONFIG)
+app = FastAPI(servers=[{"url": config.API_URL_ROOT}], **config.API_CONFIG)
class _ExternalRequestContextMiddleware(BaseHTTPMiddleware):
@@ -56,26 +56,27 @@ async def dispatch(self, request: Request, call_next):
token = _api_url_base.set(None)
try:
set_api_url_base(request)
+ set_auth_headers(request)
return await call_next(request)
finally:
_api_url_base.reset(token)
-APP.add_middleware(_ExternalRequestContextMiddleware)
+app.add_middleware(_ExternalRequestContextMiddleware)
if config.OPENTELEMETRY_ENABLED:
- FastAPIInstrumentor.instrument_app(APP)
+ FastAPIInstrumentor.instrument_app(app)
-install_error_handlers(APP)
+install_error_handlers(app)
api_prefix = f"{config.API_PREFIX}{config.API_URL}"
# Attach routers under the prefix
-APP.include_router(facility.router, prefix=api_prefix)
-APP.include_router(status.router, prefix=api_prefix)
-APP.include_router(account.router, prefix=api_prefix)
-APP.include_router(compute.router, prefix=api_prefix)
-APP.include_router(filesystem.router, prefix=api_prefix)
-APP.include_router(task.router, prefix=api_prefix)
+app.include_router(facility.router, prefix=api_prefix)
+app.include_router(status.router, prefix=api_prefix)
+app.include_router(account.router, prefix=api_prefix)
+app.include_router(compute.router, prefix=api_prefix)
+app.include_router(filesystem.router, prefix=api_prefix)
+app.include_router(task.router, prefix=api_prefix)
logging.getLogger().info(f"API path: {api_prefix}")
diff --git a/app/request_context.py b/app/request_context.py
index cb35184b..ce74d99d 100644
--- a/app/request_context.py
+++ b/app/request_context.py
@@ -7,6 +7,15 @@
_api_url_base: ContextVar[str | None] = ContextVar("_api_url_base", default=None)
+# Auth headers injected by the s3df-authnz Traefik middleware (ForwardAuth).
+_AUTH_HEADER_NAMES = (
+ "x-auth-request-primary-gid",
+ "x-auth-request-gids",
+ 'x-auth-request-uid'
+)
+
+_auth_headers: ContextVar[dict[str, str]] = ContextVar("_auth_headers", default={})
+
def set_api_url_base(request: Request) -> None:
"""Set the per-request API URL base from forwarding headers."""
@@ -21,6 +30,15 @@ def set_api_url_base(request: Request) -> None:
if host:
_api_url_base.set(f"{proto}://{host}{prefix}/{api_url}")
+def set_auth_headers(request: Request) -> None:
+ """Capture the authnz-injected headers from the current request into context."""
+ headers = {name: request.headers[name] for name in _AUTH_HEADER_NAMES if name in request.headers}
+ _auth_headers.set(headers)
+
+
+def get_auth_headers() -> dict[str, str]:
+ """Return the authnz-injected headers for the current request."""
+ return _auth_headers.get()
def get_url_prefix() -> str:
"""Return the per-request API URL base, or fall back to static config."""
diff --git a/app/routers/account/models.py b/app/routers/account/models.py
index e4c6d4f7..bf75d275 100644
--- a/app/routers/account/models.py
+++ b/app/routers/account/models.py
@@ -1,7 +1,7 @@
"""Models for account-related API endpoints, including users, projects, and allocations."""
import datetime
from pydantic import Field, computed_field, field_validator
-
+from typing import Optional
from ...request_context import get_url_prefix
from ...types.base import IRIBaseModel
from ...types.scalars import AllocationUnit
@@ -20,7 +20,7 @@ class Project(IRIBaseModel):
def _norm_dt_field(cls, v):
return cls.normalize_dt(v)
- last_modified: datetime.datetime = Field(..., description="Timestamp of the last modification of the project.", example="2026-02-21T14:30:00Z")
+ last_modified: Optional[datetime.datetime] = Field(None, description="Timestamp of the last modification of the project.", example="2026-02-21T14:30:00Z")
@computed_field(description="URI to this project resource")
@property
diff --git a/app/s3df/account_adapter.py b/app/s3df/account_adapter.py
index 34a96b9d..92509d11 100644
--- a/app/s3df/account_adapter.py
+++ b/app/s3df/account_adapter.py
@@ -61,7 +61,7 @@ async def get_user(self, user_id: str, api_key: str, client_ip: str | None, glob
# AccountFacilityAdapter methods
# -------------------------------------------------------------------------
- async def get_capabilities(self) -> list[Capability]:
+ async def get_capabilities(self, name: str | None = None, modified_since: str | None = None, offset: int = 0, limit: int = 1000) -> list[Capability]:
"""
coact.Cluster → IRI.Capability (compute)
Static storage types → IRI.Capability (storage)
@@ -114,11 +114,10 @@ async def get_projects(self, user: User) -> list[account_models.Project]:
for repo in repos:
all_users = set(repo.get("users", []) + repo.get("leaders", []) + [repo.get("principal", "")])
- # if user.id in all_users:
-
+
projects.append(account_models.Project(
id=repo["Id"],
- name=repo["name"],
+ name=f"{repo['facility']}:{repo['name']}",
description=repo.get("description", ""),
user_ids=list(all_users)
))
@@ -146,9 +145,11 @@ async def get_project_allocations(
allocations = []
# Map compute allocations
+ if len(repo_allocations) == 0:
+ raise HTTPException(status_code=404, detail=f"No compute allocations found for project {project.id}")
for comp_alloc in repo_allocations:
# comp_alloc_usage = [usage for usage in COACT_REPO_OVERALL_COMPUTE_USAGE if usage["allocation_id"] == comp_alloc["_id"]][0]
- overall_usage = comp_alloc['usage'][0]
+ overall_usage = comp_alloc['usage'][0] if comp_alloc.get('usage') else None
allocations.append(account_models.ProjectAllocation(
id=comp_alloc["_id"],
project_id=project.id,
diff --git a/app/s3df/clients/coact.py b/app/s3df/clients/coact.py
index 49d0c927..885a02eb 100644
--- a/app/s3df/clients/coact.py
+++ b/app/s3df/clients/coact.py
@@ -26,7 +26,6 @@ def __init__(
api_url: Optional[str] = None,
service_user: Optional[str] = None,
service_password: Optional[str] = None,
- use_basic_auth: bool = False
):
"""
Initialize the coact GraphQL client.
@@ -34,48 +33,49 @@ def __init__(
Args:
api_url: GraphQL endpoint URL (defaults to settings.coact_api_url)
service_user: Service account username (defaults to settings.coact_service_user)
- service_password: Service account password (for basic auth)
- use_basic_auth: If True, use Basic auth instead of coactimp headers
+ service_password: Service account password for Basic auth through nginx
"""
self.api_url = api_url or settings.coact_api_url
self.service_user = service_user or settings.coact_service_user
self.service_password = service_password
- self.use_basic_auth = use_basic_auth
- self._client: Optional[Client] = None
- LOG.info(f"Initialized CoactClient for endpoint: {self.api_url} (service_user={self.service_user}, use_basic_auth={self.use_basic_auth})")
+ LOG.info(f"Initialized CoactClient for endpoint: {self.api_url} (service_user={self.service_user})")
def _get_client(self, username: Optional[str] = None) -> Client:
"""
- Get or create a GQL client with appropriate headers.
+ Create a GQL client with appropriate headers for each request.
+
+ Always authenticates the service account via Basic auth (for nginx).
+ When ``username`` is provided, also sets the ``coactimp`` header so
+ coact-api executes the query in the context of that user.
Args:
- username: Username for impersonation (coactimp mode) or ignored (basic auth mode)
+ username: End-user username to impersonate via coactimp. Pass None
+ for service-account-level (admin) queries.
Returns:
Configured GQL Client instance
"""
- headers = {"Content-Type": "application/json"}
+ if not self.service_password:
+ raise ValueError("service_password is required")
- if self.use_basic_auth:
- # Basic Authentication for /graphql-service endpoint
- if not self.service_password:
- raise ValueError("service_password required when use_basic_auth=True")
-
- credentials = base64.b64encode(f"{self.service_user}:{self.service_password}".encode()).decode("ascii")
- print(f"Calling uri: {self.api_url} as {self.service_user} with basic auth")
- headers["Authorization"] = f"Basic {credentials}"
-
- LOG.debug(f"Using Basic auth as {self.service_user}")
- else:
- # UI-style headers for /graphql endpoint
- headers["coactimp"] = username or "null"
+ credentials = base64.b64encode(
+ f"{self.service_user}:{self.service_password}".encode()
+ ).decode("ascii")
+
+ headers = {
+ "Content-Type": "application/json",
+ "Authorization": f"Basic {credentials}",
+ }
+
+ if username:
+ headers["coactimp"] = username
headers["coactshowall"] = "true"
-
- if username:
- LOG.debug(f"Setting coactimp header for user: {username}")
+ LOG.debug(f"Impersonating user via coactimp: {username}")
+ else:
+ LOG.debug(f"Running as service account: {self.service_user}")
- transport = HTTPXAsyncTransport( # changed from HTTPXTransport
+ transport = HTTPXAsyncTransport(
url=self.api_url,
headers=headers,
timeout=30.0
@@ -222,9 +222,7 @@ async def get_user_from_lookup_service(self, username: str) -> Optional[Dict[str
result = await self.execute_query(
query,
variables={"filter": {"username": username}},
- username=self.service_user
)
- print(result)
users = result.get("usersLookupFromService", [])
return users[0] if users else None
except Exception as e:
@@ -258,7 +256,6 @@ async def get_access_groups_for_repo(self, repo_id: str) -> List[Dict[str, Any]]
result = await self.execute_query(
query,
variables={"filter": {"repoid": repo_id}},
- username=self.service_user
)
return result.get("access_groups", [])
except Exception as e:
@@ -510,7 +507,6 @@ async def get_repo_compute_allocations(self, repo_id: str) -> Optional[Dict[str,
result = await self.execute_query(
query,
variables={"repoId": repo_id},
- username=self.service_user # Use service user for this query
)
repo_allocations = result.get("repos", [])
if not repo_allocations:
@@ -645,7 +641,6 @@ async def get_user_allocation(
result = await self.execute_query(
query,
variables={"repoId": repo_id, "allocationId": allocation_id},
- username=self.service_user
)
repo = result.get("repos")
if not repo:
@@ -799,29 +794,26 @@ async def get_all_repos(self) -> List[Dict[str, Any]]:
"""
try:
- result = await self.execute_query(
- query,
- username=self.service_user
- )
+ result = await self.execute_query(query)
return result.get("repos", [])
- except Exception as e:
- LOG.error(f"Failed to get repos for user {self.service_user}: {e}")
+ except Exception as e:
+ LOG.error(f"Failed to get all repos: {e}")
return []
async def get_user_repos(self, username: str) -> List[Dict[str, Any]]:
"""
- Get repos for a user.
+ Get repos for a user by impersonating them via coactimp.
+
+ Delegates to get_my_repos so coact-api performs server-side filtering,
+ returning only the repos the user belongs to.
Args:
username: Username to query repos for
-
+
Returns:
List of repo objects
"""
- repos = await self.get_all_repos()
- user_repos = [repo for repo in repos if username in repo.get("users", []) or username in repo.get("leaders", []) or username == repo.get("principal")]
-
- return user_repos
+ return await self.get_my_repos(username)
async def get_all_repos_and_facility(self) -> List[Dict[str, Any]]:
"""
@@ -838,11 +830,11 @@ async def get_all_repos_and_facility(self) -> List[Dict[str, Any]]:
}
}
"""
- try:
- result = await self.execute_query(query, username=self.service_user)
+ try:
+ result = await self.execute_query(query)
return result.get("allreposandfacility", [])
except Exception as e:
- LOG.error(f"Failed to get all repos: {e}")
+ LOG.error(f"Failed to get all repos and facility: {e}")
return []
@@ -853,7 +845,7 @@ async def get_all_repos_and_facility(self) -> List[Dict[str, Any]]:
def get_coact_client() -> CoactClient:
"""
Get or create the default CoactClient instance.
- Reads basic auth settings from S3DFSettings.
+ Reads connection settings from S3DFSettings.
Returns:
Singleton CoactClient instance
@@ -865,7 +857,6 @@ def get_coact_client() -> CoactClient:
api_url=settings.coact_api_url,
service_user=settings.coact_service_user,
service_password=settings.coact_service_password,
- use_basic_auth=True,
)
return _default_client
diff --git a/app/s3df/clients/fs_facade.py b/app/s3df/clients/fs_facade.py
index ec5067b9..b25a8612 100644
--- a/app/s3df/clients/fs_facade.py
+++ b/app/s3df/clients/fs_facade.py
@@ -27,6 +27,9 @@
class FsFacadeError(Exception):
"""Raised when fs-facade returns an error or a task fails."""
+ def __init__(self, message: str, status_code: int = 502):
+ super().__init__(message)
+ self.status_code = status_code
class FsFacadeTimeout(Exception):
@@ -70,7 +73,8 @@ async def _request_task_id(self, method: str, path: str, **kwargs) -> str:
raise FsFacadeError(f"fs-facade transport error: {exc}") from exc
if resp.status_code >= 400:
raise FsFacadeError(
- f"fs-facade {method} {path} -> {resp.status_code}: {resp.text}"
+ f"fs-facade {method} {path} -> {resp.status_code}: {resp.text}",
+ status_code=resp.status_code,
)
data = resp.json()
# The controllers return either a bare string (response_model=str) or
@@ -123,6 +127,7 @@ async def call(
json_body: Any | None = None,
files: dict | None = None,
data: dict | None = None,
+ headers: dict | None = None,
timeout: float | None = None,
) -> Any:
"""Submit an operation, wait for terminal state, and return the parsed result.
@@ -140,6 +145,8 @@ async def call(
kwargs["files"] = files
if data is not None:
kwargs["data"] = data
+ if headers is not None:
+ kwargs["headers"] = headers
task_id = await self._request_task_id(method, path, **kwargs)
task = await self.wait(task_id, timeout=timeout)
@@ -158,6 +165,28 @@ async def call(
return result
return result
+ async def submit(
+ self,
+ method: str,
+ path: str,
+ *, #TODO: Get rid of the *
+ params: dict | None = None,
+ json_body: Any | None = None,
+ files: dict | None = None,
+ headers: dict | None = None,
+ ) -> str:
+ """Submit an operation to fs-facade and return the task_id immediately, without polling."""
+ kwargs: dict[str, Any] = {}
+ if params is not None:
+ kwargs["params"] = params
+ if json_body is not None:
+ kwargs["json"] = json_body
+ if files is not None:
+ kwargs["files"] = files
+ if headers is not None:
+ kwargs["headers"] = headers
+ return await self._request_task_id(method, path, **kwargs)
+
_default_client: Optional[FsFacadeClient] = None
diff --git a/app/s3df/compute_adapter.py b/app/s3df/compute_adapter.py
index d9400022..5529ec54 100644
--- a/app/s3df/compute_adapter.py
+++ b/app/s3df/compute_adapter.py
@@ -44,6 +44,7 @@
from ..routers.compute import models as compute_models
from ..types.user import User
from ..routers.status import models as status_models
+import shlex
logger = logging.getLogger(__name__)
@@ -208,6 +209,41 @@ def _job_from_slurm_info(job_info, include_spec: bool = False) -> dict:
}
return job_dict
+def _build_batch_script(job_spec: compute_models.JobSpec) -> str:
+ """
+ Build a Slurm batch-script body from an IRI JobSpec.
+
+ slurmrestd's `script` field requires an inline shell program, not a bare
+ path. IRI's `executable` is a path/command, so we wrap it here. This lets
+ callers pass `executable="/path/to/run.sh"` (plus `arguments`) without
+ embedding a shebang or the script's contents.
+ """
+ executable = job_spec.executable
+ if not executable:
+ raise HTTPException(
+ status_code=422,
+ detail="job_spec.executable is required for Slurm submission",
+ )
+
+ # Back-compat: caller already supplied a full script (starts with shebang).
+ if executable.startswith("#!"):
+ return executable
+
+ lines = ["#!/bin/bash"]
+ if job_spec.pre_launch:
+ lines.append(job_spec.pre_launch)
+
+ cmd = []
+ if job_spec.launcher:
+ cmd.append(job_spec.launcher)
+ cmd.append(shlex.quote(executable))
+ cmd.extend(shlex.quote(str(a)) for a in job_spec.arguments)
+ lines.append(" ".join(cmd))
+
+ if job_spec.post_launch:
+ lines.append(job_spec.post_launch)
+
+ return "\n".join(lines) + "\n"
# ---------------------------------------------------------------------------
# The adapter
@@ -251,7 +287,8 @@ def _get_slurm_context(self, user):
api = _build_slurm_api(token)
headers = _auth_headers(unix_user, token)
return api, headers
-
+
+
# -- submit_job ---------------------------------------------------------
async def submit_job(
@@ -331,7 +368,7 @@ async def submit_job(
memory_per_node=memory_per_node,
time_limit=SlurmV0041PostJobSubmitRequestJobsInnerTimeLimit(set=True, number=duration_mins),
name=name,
- script=executable,
+ script=_build_batch_script(job_spec),
argv=argv,
partition=partition,
account=account,
diff --git a/app/s3df/config.py b/app/s3df/config.py
index a01e58d6..4c880647 100644
--- a/app/s3df/config.py
+++ b/app/s3df/config.py
@@ -13,10 +13,10 @@ class S3DFSettings:
"""Configuration for S3DF coact-api integration."""
def __init__(self):
- self.coact_api_url = os.getenv("COACT_API_URL", "https://coact-dev.slac.stanford.edu/graphql-service-dev")
+ self.coact_api_url = os.getenv("COACT_API_URL", "https://coact.slac.stanford.edu/graphql-service")
self.coact_service_user = os.getenv("COACT_SERVICE_USER")
self.coact_service_password = os.getenv("COACT_SERVICE_PASSWORD")
- self.coact_use_basic_auth = True
+ self.coact_use_basic_auth = False
# Facility identification
self.facility_name = os.getenv("S3DF_FACILITY_NAME", "s3df")
diff --git a/app/s3df/filesystem_adapter.py b/app/s3df/filesystem_adapter.py
index 92ecc801..cbccb076 100644
--- a/app/s3df/filesystem_adapter.py
+++ b/app/s3df/filesystem_adapter.py
@@ -5,15 +5,12 @@
microservice, polls the task endpoint until it reaches a terminal state,
and converts the result into the matching IRI response model.
-Authentication: Dex JWT verification (mixin); POSIX identity is fetched
-from user-lookup and attached to the IRI ``User`` object. The user
-identity is not yet forwarded to fs-facade — see TODO below.
+Authentication: Dex JWT verification via S3DFAuthenticatedAdapter mixin.
See ``fs-facade-service/app/controllers/filesystem_controller.py`` for
the upstream wire format.
"""
-import base64
import logging
from fastapi import HTTPException
@@ -24,10 +21,10 @@
FsFacadeError,
FsFacadeTimeout,
get_fs_facade_client,
- get_user_lookup_client,
)
from app.routers.filesystem import facility_adapter, models
from app.routers.status import models as status_models
+from app.request_context import get_auth_headers
LOG = logging.getLogger(__name__)
@@ -35,8 +32,11 @@
async def _fs_call(method: str, path: str, **kwargs):
"""Submit an op via fs-facade and translate transport/timeout errors to HTTP."""
client = get_fs_facade_client()
+ auth_headers = get_auth_headers()
+ if auth_headers:
+ LOG.debug("Forwarding auth headers to fs-facade: %s", list(auth_headers.keys()))
try:
- return await client.call(method, path, **kwargs)
+ return await client.call(method, path, headers=auth_headers or None, **kwargs)
except FsFacadeTimeout as exc:
LOG.warning(f"fs-facade timeout on {method} {path}: {exc}")
raise HTTPException(status_code=504, detail=f"fs-facade timeout: {exc}") from exc
@@ -66,21 +66,14 @@ def _content_payload(text: str, *, content_type: models.ContentUnit, offset: int
class S3DFFilesystemAdapter(S3DFAuthenticatedAdapter, facility_adapter.FacilityAdapter):
"""Filesystem adapter that forwards operations to fs-facade-service."""
- async def get_user(self, user_id: str, api_key: str, client_ip: str | None, globus_introspect: dict | None = None) -> User:
- try:
- lookup_data = await get_user_lookup_client().get_user(user_id)
- except ValueError:
- raise HTTPException(status_code=403, detail=f"User '{user_id}' not found in directory")
- except Exception as e:
- LOG.error(f"user-lookup service error: {e}")
- raise HTTPException(status_code=502, detail="User lookup service unavailable")
-
- return User(
- id=user_id,
- name=lookup_data.get("username", user_id),
- api_key=api_key,
- client_ip=client_ip,
- )
+ async def get_user(self, user_id: str, api_key: str, client_ip: str | None, globus_introspect: dict | None = None):
+ class _User:
+ def __init__(self, uid: str, key: str):
+ self.id = uid
+ self.unix_username = uid
+ self.api_key = key
+
+ return _User(user_id, api_key)
# --- Permissions / ownership ------------------------------------------
@@ -230,23 +223,18 @@ async def symlink(
async def download(self, resource: status_models.Resource, user: User, path: str) -> models.GetFileDownloadResponse:
result = await _fs_call("GET", "/filesystem/download", params={"path": path})
- # fs-facade returns {"content": , "size": }.
- content = result.get("content") if isinstance(result, dict) else result
- return models.GetFileDownloadResponse(output=content)
+ # fs-facade returns the IRI-aligned shape: {"output": }.
+ return models.GetFileDownloadResponse(**result) if isinstance(result, dict) else models.GetFileDownloadResponse(output=result)
async def upload(self, resource: status_models.Resource, user: User, path: str, content: str) -> models.PutFileUploadResponse:
- try:
- raw = base64.b64decode(content)
- except (ValueError, TypeError) as exc:
- raise HTTPException(status_code=400, detail=f"Invalid base64 upload content: {exc}") from exc
result = await _fs_call(
"POST", "/filesystem/upload",
- params={"path": path},
- files={"file": (path.rsplit("/", 1)[-1] or "upload", raw, "application/octet-stream")},
- )
- return models.PutFileUploadResponse(
- output=result if isinstance(result, str) else "File uploaded successfully",
+ json_body={"path": path, "content": content},
)
+ # fs-facade returns the IRI-aligned shape: {"output": ""}.
+ if isinstance(result, dict):
+ return models.PutFileUploadResponse(**result)
+ return models.PutFileUploadResponse(output=result if isinstance(result, str) else "File uploaded successfully")
# --- Archive ----------------------------------------------------------
diff --git a/app/s3df/task_adapter.py b/app/s3df/task_adapter.py
new file mode 100644
index 00000000..176b0c05
--- /dev/null
+++ b/app/s3df/task_adapter.py
@@ -0,0 +1,202 @@
+"""
+S3DF Task Adapter
+
+Forwards IRI task operations to fs-facade-service:
+ - put_task: submits the filesystem operation to fs-facade without polling,
+ stores an IRI-task-id → fs-facade-task-id mapping, and returns immediately.
+ - get_task: proxies GET /task/{fs_task_id} on fs-facade and translates the
+ result into the IRI Task model.
+ - delete_task: removes from the mapping and deletes from fs-facade.
+
+The class-level _id_map ensures all IriRouter adapter instances (filesystem
+router's task_adapter and the task router's own adapter) share the same state.
+"""
+
+import json
+import logging
+import uuid
+
+from fastapi import HTTPException
+
+from app.s3df.auth.authenticated_adapter import S3DFAuthenticatedAdapter
+from app.s3df.clients import FsFacadeError, get_fs_facade_client
+from app.request_context import get_auth_headers
+from app.routers.task import facility_adapter as task_adapter, models as task_models
+from app.routers.status import models as status_models
+from app.types.user import User
+
+LOG = logging.getLogger(__name__)
+
+
+def _model_dict(val) -> dict:
+ """Return a non-None-valued dict from a Pydantic model or plain dict."""
+ if hasattr(val, "model_dump"):
+ return {k: v for k, v in val.model_dump().items() if v is not None}
+ return {k: v for k, v in val.items() if v is not None}
+
+
+def _strip_none(d: dict) -> dict:
+ return {k: v for k, v in d.items() if v is not None}
+
+
+async def _submit_to_fs_facade(task: task_models.TaskCommand) -> str:
+ """Map an IRI TaskCommand to the matching fs-facade HTTP call and return the fs task_id."""
+ client = get_fs_facade_client()
+ auth = get_auth_headers() or None
+ cmd, args = task.command, task.args
+
+ # GET endpoints — params only
+ if cmd == "file":
+ return await client.submit("GET", "/filesystem/file",
+ params={"path": args["path"]}, headers=auth)
+ if cmd == "stat":
+ return await client.submit("GET", "/filesystem/stat",
+ params=_strip_none({"path": args["path"], "dereference": args.get("dereference")}),
+ headers=auth)
+ if cmd == "ls":
+ return await client.submit("GET", "/filesystem/ls",
+ params=_strip_none({"path": args["path"], "show_hidden": args.get("show_hidden")}),
+ headers=auth)
+ if cmd == "head":
+ return await client.submit("GET", "/filesystem/head",
+ params=_strip_none({"path": args["path"], "lines": args.get("lines"), "bytes": args.get("file_bytes")}),
+ headers=auth)
+ if cmd == "tail":
+ return await client.submit("GET", "/filesystem/tail",
+ params=_strip_none({"path": args["path"], "lines": args.get("lines"), "bytes": args.get("file_bytes")}),
+ headers=auth)
+ if cmd == "view":
+ return await client.submit("GET", "/filesystem/view",
+ params=_strip_none({"path": args["path"], "size": args.get("size"), "offset": args.get("offset")}),
+ headers=auth)
+ if cmd == "checksum":
+ return await client.submit("GET", "/filesystem/checksum",
+ params={"path": args["path"]}, headers=auth)
+ if cmd == "download":
+ return await client.submit("GET", "/filesystem/download",
+ params={"path": args["path"]}, headers=auth)
+
+ # DELETE endpoint
+ if cmd == "rm":
+ return await client.submit("DELETE", "/filesystem/rm",
+ params={"path": args["path"]}, headers=auth)
+
+ # POST endpoints — JSON body
+ if cmd == "mkdir":
+ return await client.submit("POST", "/filesystem/mkdir",
+ json_body=_model_dict(args["request_model"]), headers=auth)
+ if cmd == "symlink":
+ return await client.submit("POST", "/filesystem/symlink",
+ json_body=_model_dict(args["request_model"]), headers=auth)
+ if cmd == "compress":
+ return await client.submit("POST", "/filesystem/compress",
+ json_body=_model_dict(args["request_model"]), headers=auth)
+ if cmd == "extract":
+ return await client.submit("POST", "/filesystem/extract",
+ json_body=_model_dict(args["request_model"]), headers=auth)
+ if cmd == "mv":
+ return await client.submit("POST", "/filesystem/mv",
+ json_body=_model_dict(args["request_model"]), headers=auth)
+ if cmd == "cp":
+ return await client.submit("POST", "/filesystem/cp",
+ json_body=_model_dict(args["request_model"]), headers=auth)
+ if cmd == "upload":
+ return await client.submit("POST", "/filesystem/upload",
+ json_body={"path": args["path"], "content": args["content"]}, headers=auth)
+
+ # PUT endpoints — JSON body
+ if cmd == "chmod":
+ return await client.submit("PUT", "/filesystem/chmod",
+ json_body=_model_dict(args["request_model"]), headers=auth)
+ if cmd == "chown":
+ return await client.submit("PUT", "/filesystem/chown",
+ json_body=_model_dict(args["request_model"]), headers=auth)
+
+ raise ValueError(f"Unknown filesystem command: {cmd}")
+
+
+class S3DFTaskAdapter(S3DFAuthenticatedAdapter, task_adapter.FacilityAdapter):
+ """Task adapter that proxies operations to fs-facade-service."""
+
+ # Class-level maps shared across all IriRouter-created instances.
+ _id_map: dict[str, str] = {} # iri_task_id → fs_facade_task_id
+ _cmd_map: dict[str, task_models.TaskCommand] = {} # iri_task_id → original command
+
+ def __init__(self):
+ pass
+
+ async def get_user(self, user_id: str, api_key: str, client_ip: str | None, globus_introspect: dict | None = None):
+ class _User:
+ def __init__(self, uid: str, key: str):
+ self.id = uid
+ self.unix_username = uid
+ self.api_key = key
+
+ return _User(user_id, api_key)
+
+ async def put_task(
+ self,
+ user: User,
+ resource: status_models.Resource,
+ task: task_models.TaskCommand,
+ ) -> task_models.TaskSubmitResponse:
+ try:
+ fs_task_id = await _submit_to_fs_facade(task)
+ except FsFacadeError as exc:
+ LOG.error("fs-facade submit failed for %s:%s: %s", task.router, task.command, exc)
+ raise HTTPException(status_code=exc.status_code, detail=str(exc)) from exc
+
+ iri_task_id = str(uuid.uuid4())
+ S3DFTaskAdapter._id_map[iri_task_id] = fs_task_id
+ S3DFTaskAdapter._cmd_map[iri_task_id] = task
+ LOG.info("submitted task %s (fs: %s) %s:%s", iri_task_id, fs_task_id, task.router, task.command)
+ return task_models.TaskSubmitResponse(task_id=iri_task_id)
+
+ async def get_task(self, user: User, task_id: str) -> task_models.Task | None:
+ fs_task_id = S3DFTaskAdapter._id_map.get(task_id)
+ if fs_task_id is None:
+ return None
+ try:
+ fs_task = await get_fs_facade_client().get_task(fs_task_id)
+ except FsFacadeError as exc:
+ LOG.warning("fs-facade get_task failed for %s: %s", fs_task_id, exc)
+ return None
+
+ raw_result = fs_task.get("result")
+ # fs-facade workers emit JSON strings shaped like the IRI response models
+ # (e.g. {"output": ...}). Parse to a dict so Task.result deserialises
+ # cleanly on the IRI side; fall back to {"output": } for plain strings.
+ if isinstance(raw_result, str):
+ try:
+ raw_result = json.loads(raw_result)
+ except (ValueError, TypeError):
+ raw_result = {"output": raw_result}
+
+ # Wrap lists — Task.result expects dict | None
+ if isinstance(raw_result, list):
+ raw_result = {"output": raw_result}
+
+ return task_models.Task(
+ id=task_id,
+ status=task_models.TaskStatus(fs_task["status"]),
+ result=raw_result,
+ command=S3DFTaskAdapter._cmd_map.get(task_id),
+ )
+
+ async def get_tasks(self, user: User) -> list[task_models.Task]:
+ tasks = []
+ for iri_id in list(S3DFTaskAdapter._id_map):
+ t = await self.get_task(user, iri_id)
+ if t is not None:
+ tasks.append(t)
+ return tasks
+
+ async def delete_task(self, user: User, task_id: str) -> None:
+ fs_task_id = S3DFTaskAdapter._id_map.pop(task_id, None)
+ S3DFTaskAdapter._cmd_map.pop(task_id, None)
+ if fs_task_id:
+ try:
+ client = get_fs_facade_client()
+ await client._get_client().delete(f"/task/{fs_task_id}")
+ except Exception as exc:
+ LOG.debug("fs-facade delete_task %s ignored: %s", fs_task_id, exc)
diff --git a/logo/SLAC_primary_red.png b/logo/SLAC_primary_red.png
new file mode 100644
index 00000000..2d1a0283
Binary files /dev/null and b/logo/SLAC_primary_red.png differ