# Passing parameters into app environments `[[AppEnvironment]]`s support various parameter types that can be passed at deployment time. This includes primitive values, files, directories, and delayed values like `RunOutput` and `AppEndpoint`. ## Parameter types overview There are several parameter types: - **Primitive values**: Strings, numbers, booleans - **Files**: `flyte.io.File` objects (use `flyte.io.File.from_existing_remote(...)` for remote files) - **Directories**: `flyte.io.Dir` objects (use `flyte.io.Dir.from_existing_remote(...)` for remote directories) - **Delayed values**: `RunOutput` (from task runs) or `AppEndpoint` (inject endpoint urls of other apps) ## Basic parameter types ``` # /// script # requires-python = ">=3.12" # dependencies = [ # "flyte>=2.0.0b52", # "fastapi", # "scikit-learn", # "joblib", # ] # /// """Examples showing different ways to pass parameters into apps.""" import os import flyte import flyte.app import flyte.io from flyte.app import Parameter, get_parameter # {{docs-fragment basic-parameter-types}} # String parameters app_env = flyte.app.AppEnvironment( name="configurable-app", parameters=[ Parameter(name="environment", value="production"), Parameter(name="log_level", value="INFO"), ], # ... ) # File parameters app_env2 = flyte.app.AppEnvironment( name="app-with-model", parameters=[ Parameter( name="model_file", value=flyte.io.File.from_existing_remote("s3://bucket/models/model.pkl"), mount="/app/models/", ), ], # ... ) # Directory parameters app_env3 = flyte.app.AppEnvironment( name="app-with-data", parameters=[ Parameter( name="data_dir", value=flyte.io.Dir.from_existing_remote("s3://bucket/data/"), mount="/app/data", ), ], # ... ) # {{/docs-fragment basic-parameter-types}} # {{docs-fragment parameter-access-methods}} from fastapi import FastAPI from flyte.app.extras import FastAPIAppEnvironment DATA_MOUNT_PATH = "/tmp/data_file.txt" DATA_ENV_VAR = "DATA_FILE_PATH" demo_app = FastAPI() demo_env = FastAPIAppEnvironment( name="parameter-access-demo", app=demo_app, parameters=[ # With mount and env_var: the file is downloaded to the mount path, # and an environment variable is set to point to the file location. Parameter( name="data", type="file", mount=DATA_MOUNT_PATH, env_var=DATA_ENV_VAR, ), # With no mount or env_var: accessible only via get_parameter, # which returns the path to the downloaded file. Parameter(name="data_raw", type="file"), ], # ... ) @demo_app.get("/from-mount") def read_from_mount() -> dict: """Access the file directly at the mount path.""" with open(DATA_MOUNT_PATH, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-env-var") def read_from_env_var() -> dict: """Access the file through its environment variable.""" data_path = os.environ[DATA_ENV_VAR] with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-get-parameter") def read_from_get_parameter() -> dict: """Access the file using the get_parameter helper.""" data_path = get_parameter("data") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/raw") def read_raw() -> dict: """Access a parameter that has no mount or env_var.""" data_path = get_parameter("data_raw") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} # {{/docs-fragment parameter-access-methods}} # {{docs-fragment parameter-serve-override}} if __name__ == "__main__": import logging flyte.init_from_config(log_level=logging.DEBUG) app_handle = flyte.with_servecontext( parameter_values={ demo_env.name: { "data": flyte.io.File.from_existing_remote("s3://bucket/data.txt"), "data_raw": flyte.io.File.from_existing_remote("s3://bucket/raw.txt"), } } ).serve(demo_env) print(f"Deployed app: {app_handle.url}") # {{/docs-fragment parameter-serve-override}} # {{docs-fragment runoutput-example}} env = flyte.TaskEnvironment(name="training-env") @env.task async def train_model() -> flyte.io.File: # ... training logic ... return await flyte.io.File.from_local("/tmp/trained-model.pkl") app_env4 = flyte.app.AppEnvironment( name="serving-app", parameters=[ Parameter( name="model", value=flyte.app.RunOutput(type="file", task_name="training-env.train_model"), mount="/app/model", ), ], # ... ) # {{/docs-fragment runoutput-example}} # {{docs-fragment appendpoint-example}} app1_env = flyte.app.AppEnvironment(name="backend-api") app2_env = flyte.app.AppEnvironment( name="frontend-app", parameters=[ Parameter( name="backend_url", value=flyte.app.AppEndpoint(app_name="backend-api"), env_var="BACKEND_URL", ), ], # ... ) # {{/docs-fragment appendpoint-example}} # {{docs-fragment runoutput-serving-example}} import joblib from sklearn.ensemble import RandomForestClassifier training_env = flyte.TaskEnvironment(name="training-env") @training_env.task async def train_model_task() -> flyte.io.File: """Train a model and return it.""" model = RandomForestClassifier() # ... training logic ... path = "./trained-model.pkl" joblib.dump(model, path) return await flyte.io.File.from_local(path) serving_app = FastAPI() model_serving_env = FastAPIAppEnvironment( name="model-serving-app", app=serving_app, parameters=[ Parameter( name="model", value=flyte.app.RunOutput( type="file", task_name="training-env.train_model_task", ), mount="/app/model", env_var="MODEL_PATH", ), ], ) # {{/docs-fragment runoutput-serving-example}} ``` *Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/build-apps/passing-parameters-examples.py* > [!WARNING] > Do not use `flyte.io.File(...)` or `flyte.io.Dir(...)` directly as parameter values. > Use `flyte.io.File.from_existing_remote(...)` and `flyte.io.Dir.from_existing_remote(...)` to > reference existing remote files and directories. ## Accessing parameters in your app There are three ways to access parameter values at runtime: **mount paths**, **environment variables**, and the **`get_parameter` helper function**. ### `mount` When `mount` is specified on a `Parameter`, the file or directory is downloaded to the given path. If the mount path ends with a `/`, the file is placed inside that directory with its original name. Otherwise, the file is downloaded to the exact path specified. Your app reads directly from the mount path: ```python with open("/tmp/data_file.txt", "rb") as fh: contents = fh.read() ``` ### `env_var` When `env_var` is specified, an environment variable is set containing the path to the downloaded file or directory. This is useful when you want your app code to be decoupled from specific mount paths: ```python import os data_path = os.environ["DATA_FILE_PATH"] with open(data_path, "rb") as fh: contents = fh.read() ``` ### `get_parameter` The `get_parameter(name)` helper function from `flyte.app` returns the path to the downloaded file or the string value of the parameter. This works regardless of whether `mount` or `env_var` are specified and is the most flexible access method: ```python from flyte.app import get_parameter data_path = get_parameter("data") with open(data_path, "rb") as fh: contents = fh.read() ``` When a parameter has no `mount` or `env_var` configured, `get_parameter` is the only way to access its value at runtime. ### Full example The following example defines a FastAPI app with parameters using all three access methods: ``` # /// script # requires-python = ">=3.12" # dependencies = [ # "flyte>=2.0.0b52", # "fastapi", # "scikit-learn", # "joblib", # ] # /// """Examples showing different ways to pass parameters into apps.""" import os import flyte import flyte.app import flyte.io from flyte.app import Parameter, get_parameter # {{docs-fragment basic-parameter-types}} # String parameters app_env = flyte.app.AppEnvironment( name="configurable-app", parameters=[ Parameter(name="environment", value="production"), Parameter(name="log_level", value="INFO"), ], # ... ) # File parameters app_env2 = flyte.app.AppEnvironment( name="app-with-model", parameters=[ Parameter( name="model_file", value=flyte.io.File.from_existing_remote("s3://bucket/models/model.pkl"), mount="/app/models/", ), ], # ... ) # Directory parameters app_env3 = flyte.app.AppEnvironment( name="app-with-data", parameters=[ Parameter( name="data_dir", value=flyte.io.Dir.from_existing_remote("s3://bucket/data/"), mount="/app/data", ), ], # ... ) # {{/docs-fragment basic-parameter-types}} # {{docs-fragment parameter-access-methods}} from fastapi import FastAPI from flyte.app.extras import FastAPIAppEnvironment DATA_MOUNT_PATH = "/tmp/data_file.txt" DATA_ENV_VAR = "DATA_FILE_PATH" demo_app = FastAPI() demo_env = FastAPIAppEnvironment( name="parameter-access-demo", app=demo_app, parameters=[ # With mount and env_var: the file is downloaded to the mount path, # and an environment variable is set to point to the file location. Parameter( name="data", type="file", mount=DATA_MOUNT_PATH, env_var=DATA_ENV_VAR, ), # With no mount or env_var: accessible only via get_parameter, # which returns the path to the downloaded file. Parameter(name="data_raw", type="file"), ], # ... ) @demo_app.get("/from-mount") def read_from_mount() -> dict: """Access the file directly at the mount path.""" with open(DATA_MOUNT_PATH, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-env-var") def read_from_env_var() -> dict: """Access the file through its environment variable.""" data_path = os.environ[DATA_ENV_VAR] with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-get-parameter") def read_from_get_parameter() -> dict: """Access the file using the get_parameter helper.""" data_path = get_parameter("data") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/raw") def read_raw() -> dict: """Access a parameter that has no mount or env_var.""" data_path = get_parameter("data_raw") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} # {{/docs-fragment parameter-access-methods}} # {{docs-fragment parameter-serve-override}} if __name__ == "__main__": import logging flyte.init_from_config(log_level=logging.DEBUG) app_handle = flyte.with_servecontext( parameter_values={ demo_env.name: { "data": flyte.io.File.from_existing_remote("s3://bucket/data.txt"), "data_raw": flyte.io.File.from_existing_remote("s3://bucket/raw.txt"), } } ).serve(demo_env) print(f"Deployed app: {app_handle.url}") # {{/docs-fragment parameter-serve-override}} # {{docs-fragment runoutput-example}} env = flyte.TaskEnvironment(name="training-env") @env.task async def train_model() -> flyte.io.File: # ... training logic ... return await flyte.io.File.from_local("/tmp/trained-model.pkl") app_env4 = flyte.app.AppEnvironment( name="serving-app", parameters=[ Parameter( name="model", value=flyte.app.RunOutput(type="file", task_name="training-env.train_model"), mount="/app/model", ), ], # ... ) # {{/docs-fragment runoutput-example}} # {{docs-fragment appendpoint-example}} app1_env = flyte.app.AppEnvironment(name="backend-api") app2_env = flyte.app.AppEnvironment( name="frontend-app", parameters=[ Parameter( name="backend_url", value=flyte.app.AppEndpoint(app_name="backend-api"), env_var="BACKEND_URL", ), ], # ... ) # {{/docs-fragment appendpoint-example}} # {{docs-fragment runoutput-serving-example}} import joblib from sklearn.ensemble import RandomForestClassifier training_env = flyte.TaskEnvironment(name="training-env") @training_env.task async def train_model_task() -> flyte.io.File: """Train a model and return it.""" model = RandomForestClassifier() # ... training logic ... path = "./trained-model.pkl" joblib.dump(model, path) return await flyte.io.File.from_local(path) serving_app = FastAPI() model_serving_env = FastAPIAppEnvironment( name="model-serving-app", app=serving_app, parameters=[ Parameter( name="model", value=flyte.app.RunOutput( type="file", task_name="training-env.train_model_task", ), mount="/app/model", env_var="MODEL_PATH", ), ], ) # {{/docs-fragment runoutput-serving-example}} ``` *Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/build-apps/passing-parameters-examples.py* ## Delayed values Delayed values are parameters whose actual values are materialized at deployment time. ### RunOutput Use `RunOutput` to pass outputs from task runs as app parameters: ``` # /// script # requires-python = ">=3.12" # dependencies = [ # "flyte>=2.0.0b52", # "fastapi", # "scikit-learn", # "joblib", # ] # /// """Examples showing different ways to pass parameters into apps.""" import os import flyte import flyte.app import flyte.io from flyte.app import Parameter, get_parameter # {{docs-fragment basic-parameter-types}} # String parameters app_env = flyte.app.AppEnvironment( name="configurable-app", parameters=[ Parameter(name="environment", value="production"), Parameter(name="log_level", value="INFO"), ], # ... ) # File parameters app_env2 = flyte.app.AppEnvironment( name="app-with-model", parameters=[ Parameter( name="model_file", value=flyte.io.File.from_existing_remote("s3://bucket/models/model.pkl"), mount="/app/models/", ), ], # ... ) # Directory parameters app_env3 = flyte.app.AppEnvironment( name="app-with-data", parameters=[ Parameter( name="data_dir", value=flyte.io.Dir.from_existing_remote("s3://bucket/data/"), mount="/app/data", ), ], # ... ) # {{/docs-fragment basic-parameter-types}} # {{docs-fragment parameter-access-methods}} from fastapi import FastAPI from flyte.app.extras import FastAPIAppEnvironment DATA_MOUNT_PATH = "/tmp/data_file.txt" DATA_ENV_VAR = "DATA_FILE_PATH" demo_app = FastAPI() demo_env = FastAPIAppEnvironment( name="parameter-access-demo", app=demo_app, parameters=[ # With mount and env_var: the file is downloaded to the mount path, # and an environment variable is set to point to the file location. Parameter( name="data", type="file", mount=DATA_MOUNT_PATH, env_var=DATA_ENV_VAR, ), # With no mount or env_var: accessible only via get_parameter, # which returns the path to the downloaded file. Parameter(name="data_raw", type="file"), ], # ... ) @demo_app.get("/from-mount") def read_from_mount() -> dict: """Access the file directly at the mount path.""" with open(DATA_MOUNT_PATH, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-env-var") def read_from_env_var() -> dict: """Access the file through its environment variable.""" data_path = os.environ[DATA_ENV_VAR] with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-get-parameter") def read_from_get_parameter() -> dict: """Access the file using the get_parameter helper.""" data_path = get_parameter("data") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/raw") def read_raw() -> dict: """Access a parameter that has no mount or env_var.""" data_path = get_parameter("data_raw") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} # {{/docs-fragment parameter-access-methods}} # {{docs-fragment parameter-serve-override}} if __name__ == "__main__": import logging flyte.init_from_config(log_level=logging.DEBUG) app_handle = flyte.with_servecontext( parameter_values={ demo_env.name: { "data": flyte.io.File.from_existing_remote("s3://bucket/data.txt"), "data_raw": flyte.io.File.from_existing_remote("s3://bucket/raw.txt"), } } ).serve(demo_env) print(f"Deployed app: {app_handle.url}") # {{/docs-fragment parameter-serve-override}} # {{docs-fragment runoutput-example}} env = flyte.TaskEnvironment(name="training-env") @env.task async def train_model() -> flyte.io.File: # ... training logic ... return await flyte.io.File.from_local("/tmp/trained-model.pkl") app_env4 = flyte.app.AppEnvironment( name="serving-app", parameters=[ Parameter( name="model", value=flyte.app.RunOutput(type="file", task_name="training-env.train_model"), mount="/app/model", ), ], # ... ) # {{/docs-fragment runoutput-example}} # {{docs-fragment appendpoint-example}} app1_env = flyte.app.AppEnvironment(name="backend-api") app2_env = flyte.app.AppEnvironment( name="frontend-app", parameters=[ Parameter( name="backend_url", value=flyte.app.AppEndpoint(app_name="backend-api"), env_var="BACKEND_URL", ), ], # ... ) # {{/docs-fragment appendpoint-example}} # {{docs-fragment runoutput-serving-example}} import joblib from sklearn.ensemble import RandomForestClassifier training_env = flyte.TaskEnvironment(name="training-env") @training_env.task async def train_model_task() -> flyte.io.File: """Train a model and return it.""" model = RandomForestClassifier() # ... training logic ... path = "./trained-model.pkl" joblib.dump(model, path) return await flyte.io.File.from_local(path) serving_app = FastAPI() model_serving_env = FastAPIAppEnvironment( name="model-serving-app", app=serving_app, parameters=[ Parameter( name="model", value=flyte.app.RunOutput( type="file", task_name="training-env.train_model_task", ), mount="/app/model", env_var="MODEL_PATH", ), ], ) # {{/docs-fragment runoutput-serving-example}} ``` *Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/build-apps/passing-parameters-examples.py* The `type` argument is required and must be one of `string`, `file`, or `directory`. When the app is deployed, it will make the remote calls needed to figure out the actual value of the parameter. ### AppEndpoint Use `AppEndpoint` to pass endpoints from other apps: ``` # /// script # requires-python = ">=3.12" # dependencies = [ # "flyte>=2.0.0b52", # "fastapi", # "scikit-learn", # "joblib", # ] # /// """Examples showing different ways to pass parameters into apps.""" import os import flyte import flyte.app import flyte.io from flyte.app import Parameter, get_parameter # {{docs-fragment basic-parameter-types}} # String parameters app_env = flyte.app.AppEnvironment( name="configurable-app", parameters=[ Parameter(name="environment", value="production"), Parameter(name="log_level", value="INFO"), ], # ... ) # File parameters app_env2 = flyte.app.AppEnvironment( name="app-with-model", parameters=[ Parameter( name="model_file", value=flyte.io.File.from_existing_remote("s3://bucket/models/model.pkl"), mount="/app/models/", ), ], # ... ) # Directory parameters app_env3 = flyte.app.AppEnvironment( name="app-with-data", parameters=[ Parameter( name="data_dir", value=flyte.io.Dir.from_existing_remote("s3://bucket/data/"), mount="/app/data", ), ], # ... ) # {{/docs-fragment basic-parameter-types}} # {{docs-fragment parameter-access-methods}} from fastapi import FastAPI from flyte.app.extras import FastAPIAppEnvironment DATA_MOUNT_PATH = "/tmp/data_file.txt" DATA_ENV_VAR = "DATA_FILE_PATH" demo_app = FastAPI() demo_env = FastAPIAppEnvironment( name="parameter-access-demo", app=demo_app, parameters=[ # With mount and env_var: the file is downloaded to the mount path, # and an environment variable is set to point to the file location. Parameter( name="data", type="file", mount=DATA_MOUNT_PATH, env_var=DATA_ENV_VAR, ), # With no mount or env_var: accessible only via get_parameter, # which returns the path to the downloaded file. Parameter(name="data_raw", type="file"), ], # ... ) @demo_app.get("/from-mount") def read_from_mount() -> dict: """Access the file directly at the mount path.""" with open(DATA_MOUNT_PATH, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-env-var") def read_from_env_var() -> dict: """Access the file through its environment variable.""" data_path = os.environ[DATA_ENV_VAR] with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-get-parameter") def read_from_get_parameter() -> dict: """Access the file using the get_parameter helper.""" data_path = get_parameter("data") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/raw") def read_raw() -> dict: """Access a parameter that has no mount or env_var.""" data_path = get_parameter("data_raw") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} # {{/docs-fragment parameter-access-methods}} # {{docs-fragment parameter-serve-override}} if __name__ == "__main__": import logging flyte.init_from_config(log_level=logging.DEBUG) app_handle = flyte.with_servecontext( parameter_values={ demo_env.name: { "data": flyte.io.File.from_existing_remote("s3://bucket/data.txt"), "data_raw": flyte.io.File.from_existing_remote("s3://bucket/raw.txt"), } } ).serve(demo_env) print(f"Deployed app: {app_handle.url}") # {{/docs-fragment parameter-serve-override}} # {{docs-fragment runoutput-example}} env = flyte.TaskEnvironment(name="training-env") @env.task async def train_model() -> flyte.io.File: # ... training logic ... return await flyte.io.File.from_local("/tmp/trained-model.pkl") app_env4 = flyte.app.AppEnvironment( name="serving-app", parameters=[ Parameter( name="model", value=flyte.app.RunOutput(type="file", task_name="training-env.train_model"), mount="/app/model", ), ], # ... ) # {{/docs-fragment runoutput-example}} # {{docs-fragment appendpoint-example}} app1_env = flyte.app.AppEnvironment(name="backend-api") app2_env = flyte.app.AppEnvironment( name="frontend-app", parameters=[ Parameter( name="backend_url", value=flyte.app.AppEndpoint(app_name="backend-api"), env_var="BACKEND_URL", ), ], # ... ) # {{/docs-fragment appendpoint-example}} # {{docs-fragment runoutput-serving-example}} import joblib from sklearn.ensemble import RandomForestClassifier training_env = flyte.TaskEnvironment(name="training-env") @training_env.task async def train_model_task() -> flyte.io.File: """Train a model and return it.""" model = RandomForestClassifier() # ... training logic ... path = "./trained-model.pkl" joblib.dump(model, path) return await flyte.io.File.from_local(path) serving_app = FastAPI() model_serving_env = FastAPIAppEnvironment( name="model-serving-app", app=serving_app, parameters=[ Parameter( name="model", value=flyte.app.RunOutput( type="file", task_name="training-env.train_model_task", ), mount="/app/model", env_var="MODEL_PATH", ), ], ) # {{/docs-fragment runoutput-serving-example}} ``` *Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/build-apps/passing-parameters-examples.py* The endpoint URL will be injected as the parameter value when the app starts. This is particularly useful when you want to chain apps together (for example, a frontend app calling a backend app), without hardcoding URLs. ## Overriding parameters at serve time You can override parameter values when serving apps using `parameter_values` in `flyte.with_servecontext`. File and directory values must use `from_existing_remote`: ``` # /// script # requires-python = ">=3.12" # dependencies = [ # "flyte>=2.0.0b52", # "fastapi", # "scikit-learn", # "joblib", # ] # /// """Examples showing different ways to pass parameters into apps.""" import os import flyte import flyte.app import flyte.io from flyte.app import Parameter, get_parameter # {{docs-fragment basic-parameter-types}} # String parameters app_env = flyte.app.AppEnvironment( name="configurable-app", parameters=[ Parameter(name="environment", value="production"), Parameter(name="log_level", value="INFO"), ], # ... ) # File parameters app_env2 = flyte.app.AppEnvironment( name="app-with-model", parameters=[ Parameter( name="model_file", value=flyte.io.File.from_existing_remote("s3://bucket/models/model.pkl"), mount="/app/models/", ), ], # ... ) # Directory parameters app_env3 = flyte.app.AppEnvironment( name="app-with-data", parameters=[ Parameter( name="data_dir", value=flyte.io.Dir.from_existing_remote("s3://bucket/data/"), mount="/app/data", ), ], # ... ) # {{/docs-fragment basic-parameter-types}} # {{docs-fragment parameter-access-methods}} from fastapi import FastAPI from flyte.app.extras import FastAPIAppEnvironment DATA_MOUNT_PATH = "/tmp/data_file.txt" DATA_ENV_VAR = "DATA_FILE_PATH" demo_app = FastAPI() demo_env = FastAPIAppEnvironment( name="parameter-access-demo", app=demo_app, parameters=[ # With mount and env_var: the file is downloaded to the mount path, # and an environment variable is set to point to the file location. Parameter( name="data", type="file", mount=DATA_MOUNT_PATH, env_var=DATA_ENV_VAR, ), # With no mount or env_var: accessible only via get_parameter, # which returns the path to the downloaded file. Parameter(name="data_raw", type="file"), ], # ... ) @demo_app.get("/from-mount") def read_from_mount() -> dict: """Access the file directly at the mount path.""" with open(DATA_MOUNT_PATH, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-env-var") def read_from_env_var() -> dict: """Access the file through its environment variable.""" data_path = os.environ[DATA_ENV_VAR] with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-get-parameter") def read_from_get_parameter() -> dict: """Access the file using the get_parameter helper.""" data_path = get_parameter("data") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/raw") def read_raw() -> dict: """Access a parameter that has no mount or env_var.""" data_path = get_parameter("data_raw") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} # {{/docs-fragment parameter-access-methods}} # {{docs-fragment parameter-serve-override}} if __name__ == "__main__": import logging flyte.init_from_config(log_level=logging.DEBUG) app_handle = flyte.with_servecontext( parameter_values={ demo_env.name: { "data": flyte.io.File.from_existing_remote("s3://bucket/data.txt"), "data_raw": flyte.io.File.from_existing_remote("s3://bucket/raw.txt"), } } ).serve(demo_env) print(f"Deployed app: {app_handle.url}") # {{/docs-fragment parameter-serve-override}} # {{docs-fragment runoutput-example}} env = flyte.TaskEnvironment(name="training-env") @env.task async def train_model() -> flyte.io.File: # ... training logic ... return await flyte.io.File.from_local("/tmp/trained-model.pkl") app_env4 = flyte.app.AppEnvironment( name="serving-app", parameters=[ Parameter( name="model", value=flyte.app.RunOutput(type="file", task_name="training-env.train_model"), mount="/app/model", ), ], # ... ) # {{/docs-fragment runoutput-example}} # {{docs-fragment appendpoint-example}} app1_env = flyte.app.AppEnvironment(name="backend-api") app2_env = flyte.app.AppEnvironment( name="frontend-app", parameters=[ Parameter( name="backend_url", value=flyte.app.AppEndpoint(app_name="backend-api"), env_var="BACKEND_URL", ), ], # ... ) # {{/docs-fragment appendpoint-example}} # {{docs-fragment runoutput-serving-example}} import joblib from sklearn.ensemble import RandomForestClassifier training_env = flyte.TaskEnvironment(name="training-env") @training_env.task async def train_model_task() -> flyte.io.File: """Train a model and return it.""" model = RandomForestClassifier() # ... training logic ... path = "./trained-model.pkl" joblib.dump(model, path) return await flyte.io.File.from_local(path) serving_app = FastAPI() model_serving_env = FastAPIAppEnvironment( name="model-serving-app", app=serving_app, parameters=[ Parameter( name="model", value=flyte.app.RunOutput( type="file", task_name="training-env.train_model_task", ), mount="/app/model", env_var="MODEL_PATH", ), ], ) # {{/docs-fragment runoutput-serving-example}} ``` *Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/build-apps/passing-parameters-examples.py* > [!NOTE] > Parameter overrides are only available when using `flyte.with_servecontext().serve()`. > The `flyte.deploy()` function does not support parameter overrides — parameters must be specified in the `AppEnvironment` definition. This is useful for: - Testing different configurations during development - Using different models or data sources for testing - A/B testing different app configurations ## Example: FastAPI app with configurable model Here's a complete example showing how to use parameters in a FastAPI app: ``` # /// script # requires-python = "==3.13" # dependencies = [ # "fastapi", # "uvicorn", # "joblib", # "scikit-learn", # "flyte>=2.0.0b52", # ] # /// from contextlib import asynccontextmanager from pathlib import Path import flyte import flyte.app import flyte.io from flyte.app.extras import FastAPIAppEnvironment from fastapi import FastAPI # {{docs-fragment model-serving-api}} image = flyte.Image.from_uv_script(__file__, name="app-parameters-fastapi-example") task_env = flyte.TaskEnvironment( name="model_serving_task", image=image, resources=flyte.Resources(cpu=2, memory="1Gi"), cache="auto", ) @task_env.task async def train_model_task() -> flyte.io.File: """Train a model and return it.""" import joblib import sklearn.ensemble import sklearn.datasets X, y = sklearn.datasets.make_classification(n_samples=1000, n_features=5, n_classes=2, random_state=42) model = sklearn.ensemble.RandomForestClassifier() model.fit(X, y) model_dir = Path("/tmp/model") model_dir.mkdir(parents=True, exist_ok=True) model_path = model_dir / "model.joblib" joblib.dump(model, model_path) return await flyte.io.File.from_local(model_path) state = {} @asynccontextmanager async def lifespan(app: FastAPI): import joblib model = joblib.load("/root/models/model.joblib") state["model"] = model yield app = FastAPI(lifespan=lifespan) app_env = FastAPIAppEnvironment( name="model-serving-api", app=app, parameters=[ flyte.app.Parameter( name="model_file", # this is a placeholder value=flyte.io.File.from_existing_remote("s3://bucket/models/default.pkl"), mount="/root/models/", download=True, ), ], image=image, resources=flyte.Resources(cpu=2, memory="2Gi"), requires_auth=False, ) @app.post("/predict") async def predict(data: list[float]) -> dict[str, list[float]]: model = state["model"] return {"prediction": model.predict([data]).tolist()} if __name__ == "__main__": import logging flyte.init_from_config(log_level=logging.DEBUG) run = flyte.run(train_model_task) print(f"Run: {run.url}") run.wait() model_file = run.outputs()[0] print(f"Model file: {model_file.path}") app = flyte.with_servecontext( parameter_values={ "model-serving-api": { "model_file": flyte.io.File.from_existing_remote(model_file.path) } } ).serve(app_env) print(f"API URL: {app.url}") # {{/docs-fragment model-serving-api}} ``` *Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/configure-apps/app-parameters-fastapi-example.py* ## Example: Using RunOutput for model serving ``` # /// script # requires-python = ">=3.12" # dependencies = [ # "flyte>=2.0.0b52", # "fastapi", # "scikit-learn", # "joblib", # ] # /// """Examples showing different ways to pass parameters into apps.""" import os import flyte import flyte.app import flyte.io from flyte.app import Parameter, get_parameter # {{docs-fragment basic-parameter-types}} # String parameters app_env = flyte.app.AppEnvironment( name="configurable-app", parameters=[ Parameter(name="environment", value="production"), Parameter(name="log_level", value="INFO"), ], # ... ) # File parameters app_env2 = flyte.app.AppEnvironment( name="app-with-model", parameters=[ Parameter( name="model_file", value=flyte.io.File.from_existing_remote("s3://bucket/models/model.pkl"), mount="/app/models/", ), ], # ... ) # Directory parameters app_env3 = flyte.app.AppEnvironment( name="app-with-data", parameters=[ Parameter( name="data_dir", value=flyte.io.Dir.from_existing_remote("s3://bucket/data/"), mount="/app/data", ), ], # ... ) # {{/docs-fragment basic-parameter-types}} # {{docs-fragment parameter-access-methods}} from fastapi import FastAPI from flyte.app.extras import FastAPIAppEnvironment DATA_MOUNT_PATH = "/tmp/data_file.txt" DATA_ENV_VAR = "DATA_FILE_PATH" demo_app = FastAPI() demo_env = FastAPIAppEnvironment( name="parameter-access-demo", app=demo_app, parameters=[ # With mount and env_var: the file is downloaded to the mount path, # and an environment variable is set to point to the file location. Parameter( name="data", type="file", mount=DATA_MOUNT_PATH, env_var=DATA_ENV_VAR, ), # With no mount or env_var: accessible only via get_parameter, # which returns the path to the downloaded file. Parameter(name="data_raw", type="file"), ], # ... ) @demo_app.get("/from-mount") def read_from_mount() -> dict: """Access the file directly at the mount path.""" with open(DATA_MOUNT_PATH, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-env-var") def read_from_env_var() -> dict: """Access the file through its environment variable.""" data_path = os.environ[DATA_ENV_VAR] with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/from-get-parameter") def read_from_get_parameter() -> dict: """Access the file using the get_parameter helper.""" data_path = get_parameter("data") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} @demo_app.get("/raw") def read_raw() -> dict: """Access a parameter that has no mount or env_var.""" data_path = get_parameter("data_raw") with open(data_path, "rb") as fh: return {"contents": fh.read().decode()} # {{/docs-fragment parameter-access-methods}} # {{docs-fragment parameter-serve-override}} if __name__ == "__main__": import logging flyte.init_from_config(log_level=logging.DEBUG) app_handle = flyte.with_servecontext( parameter_values={ demo_env.name: { "data": flyte.io.File.from_existing_remote("s3://bucket/data.txt"), "data_raw": flyte.io.File.from_existing_remote("s3://bucket/raw.txt"), } } ).serve(demo_env) print(f"Deployed app: {app_handle.url}") # {{/docs-fragment parameter-serve-override}} # {{docs-fragment runoutput-example}} env = flyte.TaskEnvironment(name="training-env") @env.task async def train_model() -> flyte.io.File: # ... training logic ... return await flyte.io.File.from_local("/tmp/trained-model.pkl") app_env4 = flyte.app.AppEnvironment( name="serving-app", parameters=[ Parameter( name="model", value=flyte.app.RunOutput(type="file", task_name="training-env.train_model"), mount="/app/model", ), ], # ... ) # {{/docs-fragment runoutput-example}} # {{docs-fragment appendpoint-example}} app1_env = flyte.app.AppEnvironment(name="backend-api") app2_env = flyte.app.AppEnvironment( name="frontend-app", parameters=[ Parameter( name="backend_url", value=flyte.app.AppEndpoint(app_name="backend-api"), env_var="BACKEND_URL", ), ], # ... ) # {{/docs-fragment appendpoint-example}} # {{docs-fragment runoutput-serving-example}} import joblib from sklearn.ensemble import RandomForestClassifier training_env = flyte.TaskEnvironment(name="training-env") @training_env.task async def train_model_task() -> flyte.io.File: """Train a model and return it.""" model = RandomForestClassifier() # ... training logic ... path = "./trained-model.pkl" joblib.dump(model, path) return await flyte.io.File.from_local(path) serving_app = FastAPI() model_serving_env = FastAPIAppEnvironment( name="model-serving-app", app=serving_app, parameters=[ Parameter( name="model", value=flyte.app.RunOutput( type="file", task_name="training-env.train_model_task", ), mount="/app/model", env_var="MODEL_PATH", ), ], ) # {{/docs-fragment runoutput-serving-example}} ``` *Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/build-apps/passing-parameters-examples.py* ## Best practices 1. **Use `from_existing_remote`**: Always use `flyte.io.File.from_existing_remote(...)` or `flyte.io.Dir.from_existing_remote(...)` to reference remote files and directories as parameter values. 2. **Use delayed parameters**: Leverage `RunOutput` and `AppEndpoint` to create app dependencies between tasks and apps, or app-to-app chains. 3. **Override for testing**: Use the `parameter_values` argument in `flyte.with_servecontext()` to test different configurations without changing code. 4. **Mount paths clearly**: Use descriptive mount paths for file/directory parameters so your app code is easy to understand. 5. **Use environment variables**: For paths that your app needs to reference dynamically, use `env_var` to inject values as environment variables. 6. **Use `get_parameter` for flexibility**: When you want to keep your parameter access decoupled from specific mount paths or env var names, use `get_parameter(name)`. ## Limitations - Large files/directories can slow down app startup. - Parameter overrides are only available when using `flyte.with_servecontext(...).serve(...)`. --- **Source**: https://github.com/unionai/unionai-docs/blob/main/content/user-guide/configure-apps/passing-parameters.md **HTML**: https://www.union.ai/docs/v2/union/user-guide/configure-apps/passing-parameters/