Get Tapis App Schema#
by Silvia Mazzoni, DesignSafe, 2025
In this module, we’ll use the Tapis API to explore application definitions and input schemas. The input schema defines the input requirements and default values for each app. Specifically, you’ll learn how to:
Retrieve a list of available Tapis Apps using
getAppsView detailed metadata for a specific app using
getAppFetch the most recent version of an app using
getAppLatestVersion
We’ll also use some custom utility functions to streamline queries and improve how results are displayed.
By the end, you’ll understand how to extract an app’s input schema — a powerful tool for building and validating input files for any Tapis-enabled application.
import time
# from tapipy.tapis import TapisResult
Connect to Tapis#
Yes, you need to first connect to Tapis, this authenticates you
t=OpsUtils.connect_tapis()
-- Checking Tapis token --
Token loaded from file. Token is still valid!
Token expires at: 2025-09-10T07:14:42+00:00
Token expires in: 1:54:06.979914
-- LOG IN SUCCESSFUL! --
get list of apps from Tapis using query_tapis_apps()#
Let’s use a python utility function that will search for the tapis app that meets your criteria.
query_tapis_apps.py
# /home/jupyter/CommunityData/OpenSees/TrainingMaterial/training-OpenSees-on-DesignSafe/OpsUtils/OpsUtils/Tapis/query_tapis_apps.py
def query_tapis_apps(t,idquery=[],version='',select=''):
# by Silvia Mazzoni, 2025
# examples:
# results = query_tapis_apps(t,['opensees','mp'],version='latest',select = 'id,created,description,version')
# results = query_tapis_apps(t,['opensees','mp'],select = 'id,created,description,version')
# results = query_tapis_apps(t,['opensees','mp'],version='latest')
# results = query_tapis_apps(t,['opensees','sp'])
listType = 'ALL'
inputs = [t]
searchQuery = ''
if len(idquery)>0:
searchQuery = "id.like.*"
for thisQ in idquery:
searchQuery += f"{thisQ}*"
if len(version)>0:
endstr = ''
if len(searchQuery)>0:
searchQuery = f'({searchQuery})~('
endstr = ')'
searchQuery += f'version.eq.{version}'
searchQuery += endstr
results = t.apps.getApps(search=searchQuery,
listType=listType,select=select)
return results
get the schema for a specific app: OpenSeesMP#
we want the latest version of the OpenSeesMP tapis app
results = OpsUtils.query_tapis_apps(t,['opensees','mp'],version='latest',select = 'id,created,description,version')
select the index of the app that meets our criteria.#
app_index = 0; # the first (and only)
appMeta = results[app_index]
print('appMeta',appMeta)
appMeta
created: 2025-02-20T18:01:49.005183Z
description: Runs all the processors in parallel. Requires understanding of parallel processing and the capabilities to write parallel scripts.
id: opensees-mp-s3
version: latest
appId = appMeta.id
appVersion = appMeta.version
use a utility function:#
get_tapis_app_schema.py
# /home/jupyter/CommunityData/OpenSees/TrainingMaterial/training-OpenSees-on-DesignSafe/OpsUtils/OpsUtils/Tapis/get_tapis_app_schema.py
def get_tapis_app_schema(t, appId: str, version: str = "latest", quiet: bool = False):
"""
Fetch a Tapis App schema by ID and version (or the latest version).
Behavior
--------
- If `version` is empty or equals "latest" (case-insensitive), retrieves the app's
latest available version via `t.apps.getAppLatestVersion`.
- Otherwise, retrieves the specified version via `t.apps.getApp(appVersion=...)`.
- Returns the schema object (typically a TapisResult) on success, `None` on failure.
Parameters
----------
t : tapipy.tapis.Tapis
An authenticated Tapis client.
appId : str
The Tapis app ID (e.g., "opensees-mp-s3").
version : str, default "latest"
App version string (e.g., "1.0.3") or "latest" (case-insensitive).
quiet : bool, default False
If True, suppresses error prints and simply returns `None` on failure.
Returns
-------
tapipy.tapis.TapisResult | dict | None
The app schema object on success (commonly a TapisResult). Returns `None` when
not found or if an error occurs.
Example
-------
# Get latest
schema = get_tapis_app_schema(t, "opensees-mp-s3")
# Get a specific version
schema_v = get_tapis_app_schema(t, "opensees-mp-s3", version="2.1.0")
Author
------
Silvia Mazzoni, DesignSafe (silviamazzoni@yahoo.com)
Date
----
2025-08-14
Version
-------
1.0
"""
from tapipy.errors import BaseTapyException
# Normalize version
ver = (version or "").strip().lower()
try:
if ver == "" or ver == "latest":
return t.apps.getAppLatestVersion(appId=appId)
else:
return t.apps.getApp(appId=appId, appVersion=version)
except BaseTapyException as e:
if not quiet:
print(f"I was unable to find Tapis app: '{appId}', version='{version}'. Error: {e}")
return None
except Exception as e:
if not quiet:
print(f"Unexpected error retrieving app '{appId}' (version='{version}'): {e}")
return None
thisAppData_MP = OpsUtils.get_tapis_app_schema(t,appId,version='latest')
use a utility function to display the schema#
display_tapis_app_schema.py
# /home/jupyter/CommunityData/OpenSees/TrainingMaterial/training-OpenSees-on-DesignSafe/OpsUtils/OpsUtils/Tapis/display_tapis_app_schema.py
def display_tapis_app_schema(thisAppSchema):
"""
Pretty-print a Tapis App schema (or any nested TapisResult/dict/list) in a
JSON-like format with readable grouping and indentation.
Behavior
--------
- Accepts a TapisResult, dict, or list.
- Groups and prints keys in the order: scalars → lists → nested dicts.
- Handles TapisResult values by flattening their internal __dict__.
- Prints arrays either inline (simple scalars) or expanded (objects).
- Shows a header with app id and version (if present).
Parameters
----------
thisAppSchema : tapipy.tapis.TapisResult | dict | list
The app schema or object to render.
Returns
-------
None
This function prints to stdout. It does not return a value.
Example
-------
# Given a Tapis app schema returned by tapipy:
display_tapis_app_schema(app_schema)
Author
------
Silvia Mazzoni, DesignSafe (silviamazzoni@yahoo.com)
Date
----
2025-08-14
Version
-------
1.0
"""
try:
from tapipy.tapis import TapisResult
except Exception:
class TapisResult: # sentinel so isinstance checks won’t crash if tapipy not available
pass
def _is_simple_scalar(x):
return isinstance(x, (type(None), bool, int, float, str))
def _quote_if_str(x):
return f'"{x}"' if isinstance(x, str) else x
def print_nested(key_prefix, obj, indent=1):
sp = ' ' * indent
# Unwrap TapisResult to dict
if isinstance(obj, TapisResult):
obj = obj.__dict__
# Dict handling
if isinstance(obj, dict):
dict_keys, list_keys, scalar_keys = [], [], []
for k, v in obj.items():
v_unwrap = v.__dict__ if isinstance(v, TapisResult) else v
if isinstance(v_unwrap, dict):
dict_keys.append(k)
elif isinstance(v_unwrap, list):
list_keys.append(k)
else:
scalar_keys.append(k)
# print in order: scalars, lists, dicts
ordered = [*scalar_keys, *list_keys, *dict_keys]
for i, k in enumerate(ordered):
v = obj[k]
v_unwrap = v.__dict__ if isinstance(v, TapisResult) else v
# Nested dict or object
if isinstance(v_unwrap, dict):
# opening brace for top-level objects
if key_prefix:
print(f'{sp}{key_prefix}{k}: ' + '{')
next_prefix = ""
else:
print(f'{sp}{k}: ' + '{')
next_prefix = ""
print_nested(next_prefix, v_unwrap, indent + 1)
print(f'{sp}' + '}')
if i != len(ordered) - 1:
pass # stylistically omit commas for readability
# Lists
elif isinstance(v_unwrap, list):
# Decide inline vs expanded
contains_objects = any(
(isinstance(it, (dict, TapisResult))) for it in v_unwrap
)
label = f'{sp}{key_prefix}{k}: ' if key_prefix else f'{sp}{k}: '
if not v_unwrap:
print(label + '[]')
elif contains_objects:
print(label + '[')
for j, it in enumerate(v_unwrap):
it_unwrap = it.__dict__ if isinstance(it, TapisResult) else it
if isinstance(it_unwrap, dict):
print(' ' * (indent + 1) + '{')
print_nested("", it_unwrap, indent + 2)
print(' ' * (indent + 1) + '}')
else:
val = _quote_if_str(it_unwrap)
print(' ' * (indent + 1) + f'{val}')
if j != len(v_unwrap) - 1:
pass # omit commas for readability
print(sp + ']')
else:
# all simple values → inline
vals = [_quote_if_str(x) for x in v_unwrap]
print(label + f'[{", ".join(map(str, vals))}]')
# Scalars
else:
val = _quote_if_str(v_unwrap)
label = f'{sp}{key_prefix}{k}: ' if key_prefix else f'{sp}{k}: '
print(label + f'{val}')
# List handling (rare for top-level)
elif isinstance(obj, list):
sp = ' ' * indent
if not obj:
print(sp + '[]')
return
contains_objects = any(isinstance(it, (dict, TapisResult)) for it in obj)
if contains_objects:
print(sp + '[')
for it in obj:
it_unwrap = it.__dict__ if isinstance(it, TapisResult) else it
if isinstance(it_unwrap, dict):
print(' ' * (indent + 1) + '{')
print_nested("", it_unwrap, indent + 2)
print(' ' * (indent + 1) + '}')
else:
print(' ' * (indent + 1) + f'{_quote_if_str(it_unwrap)}')
print(sp + ']')
else:
vals = [_quote_if_str(x) for x in obj]
print(sp + f'[{", ".join(map(str, vals))}]')
# Fallback scalar
else:
val = _quote_if_str(obj)
print(sp + f'{val}')
# Header
print('########################################')
print('########### TAPIS-APP SCHEMA ###########')
print('########################################')
# Best-effort id/version extraction
app_id = getattr(thisAppSchema, 'id', None) or (thisAppSchema.get('id') if isinstance(thisAppSchema, dict) else None)
version = getattr(thisAppSchema, 'version', None) or (thisAppSchema.get('version') if isinstance(thisAppSchema, dict) else None)
if app_id is not None:
print(f'######## appID: {app_id}')
if version is not None:
print(f'######## version: {version}')
print('########################################')
# Body (start with a label to indicate the root)
print('{')
print_nested("", thisAppSchema, indent=1)
print('}')
print('########################################')
OpsUtils.display_tapis_app_schema(thisAppData_MP)
########################################
########### TAPIS-APP SCHEMA ###########
########################################
######## appID: opensees-mp-s3
######## version: latest
########################################
{
sharedAppCtx: "wma_prtl"
isPublic: True
tenant: "designsafe"
id: "opensees-mp-s3"
version: "latest"
description: "Runs all the processors in parallel. Requires understanding of parallel processing and the capabilities to write parallel scripts."
owner: "wma_prtl"
enabled: True
versionEnabled: True
locked: False
runtime: "ZIP"
runtimeVersion: None
runtimeOptions: None
containerImage: "tapis://cloud.data/corral/tacc/aci/CEP/applications/v3/opensees/latest/OpenSees/opensees.zip"
jobType: "BATCH"
maxJobs: 2147483647
maxJobsPerUser: 2147483647
strictFileInputs: True
uuid: "1410a584-0c5e-4e47-b3b0-3a7bea0e1187"
deleted: False
created: "2025-02-20T18:01:49.005183Z"
updated: "2025-08-28T20:17:39.426067Z"
sharedWithUsers: []
tags: ["portalName: DesignSafe", "portalName: CEP"]
jobAttributes: {
description: None
dynamicExecSystem: False
execSystemConstraints: None
execSystemId: "stampede3"
execSystemExecDir: "${JobWorkingDir}"
execSystemInputDir: "${JobWorkingDir}"
execSystemOutputDir: "${JobWorkingDir}"
dtnSystemInputDir: "!tapis_not_set"
dtnSystemOutputDir: "!tapis_not_set"
execSystemLogicalQueue: "skx"
archiveSystemId: "stampede3"
archiveSystemDir: "HOST_EVAL($WORK)/tapis-jobs-archive/${JobCreateDate}/${JobName}-${JobUUID}"
archiveOnAppError: True
isMpi: False
mpiCmd: None
cmdPrefix: None
nodeCount: 2
coresPerNode: 48
memoryMB: 192000
maxMinutes: 120
fileInputs: [
{
name: "Input Directory"
description: "Input directory that includes the tcl script as well as any other required files. Example input is in tapis://designsafe.storage.community/app_examples/opensees/OpenSeesMP"
inputMode: "REQUIRED"
autoMountLocal: True
envKey: "inputDirectory"
sourceUrl: None
targetPath: "inputDirectory"
notes: {
selectionMode: "directory"
}
}
]
fileInputArrays: []
subscriptions: []
tags: []
parameterSet: {
appArgs: [
{
arg: "OpenSeesMP"
name: "mainProgram"
description: None
inputMode: "FIXED"
notes: {
isHidden: True
}
}
{
arg: None
name: "Main Script"
description: "The filename only of the OpenSees TCL script to execute. This file should reside in the Input Directory specified. To use with test input, use 'freeFieldEffective.tcl'"
inputMode: "REQUIRED"
notes: {
inputType: "fileInput"
}
}
]
containerArgs: []
schedulerOptions: [
{
arg: "--tapis-profile OpenSees_default"
name: "OpenSees TACC Scheduler Profile"
description: "Scheduler profile for the default version of OpenSees"
inputMode: "FIXED"
notes: {
isHidden: True
}
}
{
arg: None
name: "TACC Reservation"
description: "Reservation input string"
inputMode: "INCLUDE_ON_DEMAND"
notes: {
isHidden: True
}
}
]
envVariables: []
archiveFilter: {
includeLaunchFiles: True
includes: []
excludes: []
}
logConfig: {
stdoutFilename: ""
stderrFilename: ""
}
}
}
notes: {
icon: "OpenSees"
label: "OpenSeesMP"
helpUrl: "https://www.designsafe-ci.org/user-guide/tools/simulation/#opensees-user-guide"
category: "Simulation"
isInteractive: False
showReservation: True
hideNodeCountAndCoresPerNode: False
}
}
########################################
OpenSees-Express#
tapis_apps = OpsUtils.query_tapis_apps(t,['opensees','express'],version='latest',select = 'id,created,description,version')
print(tapis_apps)
[
created: 2025-02-20T18:41:03.661272Z
description: OpenSees-EXPRESS provides users with a sequential OpenSees interpreter. It is ideal to run small sequential scripts on DesignSafe resources freeing up your own machine.
id: opensees-express
version: latest,
created: 2025-02-20T21:27:38.534908Z
description: OpenSees-EXPRESS provides users with a sequential OpenSees interpreter. It is ideal to run small sequential scripts on DesignSafe resources freeing up your own machine.
id: opensees-express.tms
version: latest]
app_index = 0; # the first (and only)
appMeta = tapis_apps[app_index]
print('appMeta',appMeta)
appId = appMeta.id
appVersion = appMeta.version
thisAppSchema_OpenSeesExpress = OpsUtils.get_tapis_app_schema(t,appId,version='latest')
OpsUtils.display_tapis_app_schema(thisAppSchema_OpenSeesExpress)
appMeta
created: 2025-02-20T18:41:03.661272Z
description: OpenSees-EXPRESS provides users with a sequential OpenSees interpreter. It is ideal to run small sequential scripts on DesignSafe resources freeing up your own machine.
id: opensees-express
version: latest
########################################
########### TAPIS-APP SCHEMA ###########
########################################
######## appID: opensees-express
######## version: latest
########################################
{
sharedAppCtx: "wma_prtl"
isPublic: True
tenant: "designsafe"
id: "opensees-express"
version: "latest"
description: "OpenSees-EXPRESS provides users with a sequential OpenSees interpreter. It is ideal to run small sequential scripts on DesignSafe resources freeing up your own machine."
owner: "wma_prtl"
enabled: True
versionEnabled: True
locked: False
runtime: "ZIP"
runtimeVersion: None
runtimeOptions: None
containerImage: "tapis://cloud.data/corral/tacc/aci/CEP/applications/v3/opensees/latest/OpenSees-EXPRESS/opensees_express.zip"
jobType: "FORK"
maxJobs: 2147483647
maxJobsPerUser: 2147483647
strictFileInputs: True
uuid: "30cb1fa1-e7c7-44a8-a0e8-d2f64043fc65"
deleted: False
created: "2025-02-20T18:41:03.661272Z"
updated: "2025-07-30T20:19:29.367292Z"
sharedWithUsers: []
tags: ["portalName: DesignSafe", "portalName: CEP"]
jobAttributes: {
description: None
dynamicExecSystem: False
execSystemConstraints: None
execSystemId: "wma-exec-01"
execSystemExecDir: "${JobWorkingDir}"
execSystemInputDir: "${JobWorkingDir}"
execSystemOutputDir: "${JobWorkingDir}"
dtnSystemInputDir: "!tapis_not_set"
dtnSystemOutputDir: "!tapis_not_set"
execSystemLogicalQueue: None
archiveSystemId: "cloud.data"
archiveSystemDir: "/tmp/${JobOwner}/tapis-jobs-archive/${JobCreateDate}/${JobName}-${JobUUID}"
archiveOnAppError: True
isMpi: False
mpiCmd: None
cmdPrefix: None
nodeCount: 1
coresPerNode: 1
memoryMB: 100
maxMinutes: 1440
fileInputs: [
{
name: "Input Directory"
description: "Input directory that includes the tcl script as well as any other required files. Example input is in tapis://designsafe.storage.community/app_examples/opensees/OpenSeesEXPRESS"
inputMode: "REQUIRED"
autoMountLocal: True
envKey: "inputDirectory"
sourceUrl: None
targetPath: "*"
notes: {
selectionMode: "directory"
}
}
]
fileInputArrays: []
subscriptions: []
tags: []
parameterSet: {
appArgs: []
containerArgs: []
schedulerOptions: []
envVariables: [
{
key: "mainProgram"
value: "OpenSees"
description: "Choose the OpenSees binary to use."
inputMode: "REQUIRED"
notes: {
label: "Main Program"
enum_values: [
{
OpenSees: "OpenSees"
}
{
OpenSeesSP: "OpenSeesSP"
}
]
}
}
{
key: "tclScript"
value: ""
description: "The filename of the OpenSees TCL script to execute, e.g. "freeFieldEffective.tcl"."
inputMode: "REQUIRED"
notes: {
label: "Main Script"
inputType: "fileInput"
}
}
]
archiveFilter: {
includeLaunchFiles: True
includes: []
excludes: ["opensees-express.zip", "tapisjob.env"]
}
logConfig: {
stdoutFilename: ""
stderrFilename: ""
}
}
}
notes: {
icon: "OpenSees"
label: "OpenSees-EXPRESS (VM)"
helpUrl: "https://www.designsafe-ci.org/user-guide/tools/simulation/#opensees-user-guide"
category: "Simulation"
isInteractive: False
hideNodeCountAndCoresPerNode: True
}
}
########################################