Skip to main content

Introduction to AppEnvironment

Introduction to AppEnvironment in flyte-sdk​

Flyte is designed to orchestrate both batch-processing tasks and long-running services. While TaskEnvironment is used to define the execution context for individual, often ephemeral, tasks, AppEnvironment provides the necessary configuration for deploying and managing persistent, serving applications such as APIs, dashboards, or machine learning model servers.

The AppEnvironment class extends the base Environment class, which establishes common infrastructure settings like container images, compute resources, and environment variables. You typically do not instantiate Environment directly; instead, you use AppEnvironment for services or TaskEnvironment for batch tasks.

Defining a Long-Running Application​

When you need to deploy an application that continuously serves requests, such as a Streamlit dashboard or a FastAPI service, you define it using flyte.app.AppEnvironment. This class encapsulates all the necessary information for Flyte to build, deploy, and manage your service.

For instance, to define a simple Streamlit application, you specify its name, the Docker image it should run in, and the command-line arguments needed to start it:

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: 'flyte._resources.Resources' does not accept keyword(s): cpu, memory

import flyte
from flyte.app._app_environment import AppEnvironment
from flyte._image import Image
from flyte._resources import Resources

image = Image.from_debian_base(python_version=(3, 12)).with_pip_packages("streamlit==1.41.1")

app_env = AppEnvironment(
    name="streamlit-hello-v2",
    image=image,
    args="streamlit hello --server.port 8080",
    resources=Resources(cpu=1, memory="1Gi"),
)

# In a real application, you would then use flyte.deploy(app_env) or serve it locally.

In this example, AppEnvironment is initialized with:

  • name: A unique identifier for the application, such as "streamlit-hello-v2". As inherited from Environment, this must be in snake_case or kebab-case.
  • image: The container image where your application code will run. Here, Image.from_debian_base is used to build a Python 3.12 image with streamlit installed.
  • args: Command-line arguments passed to the container's entrypoint to start the application. For Streamlit, this is typically streamlit hello --server.port 8080.
  • resources: Specifies the compute resources (CPU, memory) allocated to the application's instances using Resources.

Internally, AppEnvironment extends flyte.Environment and adds app-specific fields. The __post_init__ method of AppEnvironment performs validation specific to app names and ensures that args and command are correctly typed:

# From flyte.app._app_environment.AppEnvironment
def _validate_name(self):
    # APP_NAME_RE is assumed to be defined in the context of AppEnvironment
    if not APP_NAME_RE.fullmatch(self.name):
        raise ValueError(
            f"App name \'{self.name}\' must consist of lower case alphanumeric characters or '-, "
            "and must start and end with an alphanumeric character."
        )

def __post_init__(self):
    super().__post_init__()
    if self.args is not None and not isinstance(self.args, (list, str)):
        raise TypeError(f"Expected args to be of type List[str] or str, got {type(self.args)}")
    # ... other validations ...

Configuring Application Ports​

Applications typically listen on a specific network port. You can configure this port using the port parameter of AppEnvironment.

If your application, for example, an http.server, listens on port 8082, you would specify it as follows:

import flyte
from flyte.app._app_environment import AppEnvironment
from flyte._image import Image
from flyte._resources import Resources

app_env = AppEnvironment(
    name="local-static-server",
    image=Image.from_debian_base(python_version=(3, 12)),
    command="python -m http.server 8082",
    port=8082,
    requires_auth=False,
)

By default, AppEnvironment uses port 8080. You can also provide a flyte.app.Port object for more advanced configurations, though typically an integer is sufficient. The AppEnvironment ensures that the specified port is not one of the reserved ports:

if isinstance(self.port, int):
    self.port = Port(port=self.port)  # Name should be blank can be h2c / http1
    if self.port.port in INVALID_APP_PORTS:
        raise ValueError(f"Port {self.port.port} is reserved and cannot be used for AppEnvironment")

The flyte.app.Port dataclass itself also validates against INVALID_APP_PORTS:

# From flyte.app._types.Port
from dataclasses import dataclass
from typing import Optional

@dataclass(frozen=True)
class Port:
    port: int
    name: Optional[str] = None

    def __post_init__(self):
        # INVALID_APP_PORTS is assumed to be defined in the context of Port
        if self.port in INVALID_APP_PORTS:
            invalid_ports = ", ".join(str(p) for p in INVALID_APP_PORTS)
            msg = f"port {self.port} is not allowed. Please do not use ports: {invalid_ports}"
            raise ValueError(msg)

Container Execution: command vs. args​

Flyte provides two primary ways to specify how your application starts within its container: command and args.

  • args: Use args when your Docker image has an ENTRYPOINT defined, and you only need to pass additional arguments to it. This is common for framework-specific images (e.g., a Streamlit image where streamlit is the entrypoint).
Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: unexpected indent

```python
# From examples/apps/basic_app.py
from flyte.app._app_environment import AppEnvironment
from flyte._image import Image
from flyte._resources import Resources

app_env = AppEnvironment(
    name="streamlit-hello-v2",
    image=Image.from_debian_base(python_version=(3, 12)).with_pip_packages("streamlit==1.41.1"),
    args="streamlit hello --server.port 8080", # These args are passed to the image's ENTRYPOINT
    resources=Resources(cpu=1, memory="1Gi"),
)
```
  • command: Use command when you need to completely override the container's default ENTRYPOINT and CMD. This is useful for running custom scripts or executables directly.
Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: unexpected indent

```python
# From examples/apps/local_command_app.py
from flyte.app._app_environment import AppEnvironment
from flyte._image import Image
from flyte._resources import Resources

app_env = AppEnvironment(
    name="local-static-server",
    image=Image.from_debian_base(python_version=(3, 12)),
    command="python -m http.server 8082", # This command overrides the container's entrypoint
    resources=Resources(cpu=1, memory="512Mi"),
)
```

Internally, the flyte._serve._Serve class, responsible for local and remote serving, determines how to start the application based on these parameters. The _serve_local method checks for _server (for decorator-based apps), then command, then args:

# From flyte._serve._Serve._serve_local
if app_env._server is not None:
    # Use the @app_env.server decorator function - run in a background thread
    return self._serve_local_with_server_func(app_env, _LOCAL_HOST, port, materialized_parameters)
elif app_env.command is not None:
    # Use the command / args specification - run as a subprocess
    return self._serve_local_with_command(app_env, _LOCAL_HOST, port)
elif app_env.args is not None:
    return self._serve_local_with_command(app_env, _LOCAL_HOST, port)
else:
    raise ValueError(
        f"AppEnvironment \'{app_env.name}\' has no server function, command, or args defined. "
        "Cannot serve locally."
    )

Application Lifecycle Hooks​

AppEnvironment allows you to define functions that run at specific points in your application's lifecycle: on_startup, server, and on_shutdown. These are defined using decorators on methods of your AppEnvironment instance.

  • @app_env.on_startup: This function runs once before your main server process starts. It's ideal for initializing resources like database connections or loading models.
  • @app_env.server: This is the core function that starts your application's server. It should typically block indefinitely (e.g., by running a uvicorn or fastapi server).
  • @app_env.on_shutdown: This function runs when your application is gracefully shutting down, allowing you to clean up resources.

These functions can be synchronous or asynchronous and can accept parameters defined in your AppEnvironment.

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: 'flyte.app._parameter.Parameter' does not accept keyword(s): default, name, type

import flyte
from flyte.app._app_environment import AppEnvironment
from flyte.app._parameter import Parameter
from flyte._image import Image
import uvicorn
from fastapi import FastAPI

# Define an AppEnvironment for a FastAPI application
my_api_env = AppEnvironment(
    name="my-fastapi-app",
    image=Image.from_debian_base(python_version=(3, 12)).with_pip_packages("fastapi", "uvicorn"),
    port=8000,
    parameters=[
        Parameter(name="message", default="Hello from Flyte!", type=str),
    ],
)

app = FastAPI()

@app.get("/")
async def read_root(message: str = "Default message"):
    return {"message": message}

@my_api_env.on_startup
def startup_logic(message: str):
    print(f"Starting up with message: {message}")
    # Initialize database connections, load models, etc.

@my_api_env.server
def run_server():
    print("Running FastAPI server...")
    uvicorn.run(app, host="0.0.0.0", port=my_api_env.get_port().port)

@my_api_env.on_shutdown
def shutdown_logic():
    print("Shutting down FastAPI server...")
    # Close database connections, release resources, etc.

# To run this locally:
# if __name__ == "__main__":
#     flyte.init()
#     local_app = flyte.with_servecontext(mode="local").serve(my_api_env)
#     local_app.activate(wait=True)
#     # ... test app ...
#     local_app.deactivate(wait=True)

When you use these decorators, AppEnvironment stores references to your functions in private attributes (_on_startup, _server, _on_shutdown). During local serving, _Serve._serve_local_with_server_func is responsible for invoking these functions:

# From flyte._serve._Serve._serve_local_with_server_func
if app_env._on_startup is not None:
    bound_params = _bind_parameters(app_env._on_startup, materialized_parameters)
    if asyncio.iscoroutinefunction(app_env._on_startup):
        loop.run_until_complete(app_env._on_startup(**bound_params))
    else:
        app_env._on_startup(**bound_params)

# Run the server function
bound_params = _bind_parameters(app_env._server, materialized_parameters)
if asyncio.iscoroutinefunction(app_env._server):
    loop.run_until_complete(app_env._server(**bound_params))
else:
    app_env._server(**bound_params)

# ... (shutdown logic is handled in finally block)

Scaling and Resource Configuration​

Controlling how your application scales and the resources it consumes is crucial for efficiency and reliability. AppEnvironment provides the scaling and resources parameters for this purpose.

  • scaling: Use flyte.app.Scaling to define the number of replicas and autoscaling behavior. You can configure fixed replicas, scale-to-zero, or burstable scaling.
Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: unexpected indent

```python
from flyte.app._app_environment import AppEnvironment
from flyte.app._types import Scaling

# Scale-to-zero: 0 replicas when idle, up to 1 on demand (default)
scale_to_zero_app = AppEnvironment(name="my-app-zero", scaling=Scaling(replicas=(0, 1)))

# Always-on: exactly 1 replica at all times
always_on_app = AppEnvironment(name="my-app-always-on", scaling=Scaling(replicas=(1, 1)))

# Burstable: 1 replica minimum, scales up to 5 based on demand
burstable_app = AppEnvironment(name="my-app-burstable", scaling=Scaling(replicas=(1, 5)))

# Fixed size: exactly 3 replicas
fixed_size_app = AppEnvironment(name="my-app-fixed", scaling=Scaling(replicas=3))

# Autoscaling based on concurrency
concurrency_scaled_app = AppEnvironment(
    name="my-app-concurrency",
    scaling=Scaling(replicas=(1, 10), metric=Scaling.Concurrency(val=10)), # Scale up if 10 concurrent requests per replica
)
```

The `flyte.app.Scaling` dataclass handles various replica configurations and optional autoscaling metrics:

```python

From flyte.app._types.Scaling

from dataclasses import dataclass from typing import Optional, Union, Tuple from datetime import timedelta

@dataclass class Scaling: # ... replicas: Union[int, Tuple[int, int]] = (0, 1) metric: Optional[Union["Concurrency", "RequestRate"]] = None # Removed "Scaling." prefix scaledown_after: int | timedelta | None = None # ... def post_init(self): if isinstance(self.replicas, int): if self.replicas < 0: raise ValueError("replicas must be greater than or equal to 0") self.replicas = (self.replicas, self.replicas) # ... other validations ... ```

  • resources: As seen in earlier examples, flyte._resources.Resources allows you to specify CPU, memory, GPU, and disk requirements for your application's containers. This is inherited from the base Environment class.
Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: unexpected indent

```python
# From examples/apps/basic_app.py
from flyte.app._app_environment import AppEnvironment
from flyte._image import Image
from flyte._resources import Resources

app_env = AppEnvironment(
    name="streamlit-hello-v2",
    image=Image.from_debian_base(python_version=(3, 12)).with_pip_packages("streamlit==1.41.1"),
    resources=Resources(cpu=1, memory="1Gi"),
)
```

Other AppEnvironment Configurations​

AppEnvironment offers a rich set of additional parameters to fine-tune your application's deployment and behavior:

  • requires_auth: A boolean (default True) indicating whether the app endpoint requires authentication. Set to False for public endpoints.
  • env_vars: A dictionary of environment variables to inject into the application's container, inherited from Environment.
  • secrets: Specifies secrets to be mounted into the container, using flyte.SecretRequest.
  • depends_on: A list of other Environment objects that should be deployed alongside this application, ensuring co-location or specific deployment order.
  • include: Extra files or directories to bundle with the application's code. This is useful for non-Python assets like HTML templates or configuration files. Note that include is not supported in interactive/pkl serves.
  • cluster_pool: Specifies the cluster pool for scheduling the application, defaulting to "default".
  • timeouts: Configures various timeouts for the application, such as request timeouts, using flyte.app.Timeouts.
  • domain: Allows custom domain configuration via flyte.app.Domain.
  • links: A list of flyte.app.Link objects to define custom links associated with the application.
  • parameters: A list of flyte.app.Parameter objects to define inputs for your application, which can be connected to task outputs or other app endpoints.

These configurations provide comprehensive control over how your long-running applications are deployed and managed within the flyte-sdk ecosystem.