This class provides a high-level interface for managing and interacting with application deployments. It supports lifecycle operations such as creating, updating, activating, and deleting apps, while providing real-time status monitoring through watch mechanisms and context managers. Users can retrieve application metadata, deployment states, and public endpoints via a set of properties and class-level query methods.
Attributes
| Attribute | Type | Description |
|---|
| pb2 | app_definition_pb2.App | The underlying protobuf message containing the complete application definition and state. |
| name | str | Get the name of the app. |
| revision | int | Get the revision number of the app. |
| endpoint | str | Get the public endpoint URL of the app. |
| deployment_status | app_definition_pb2.Status.DeploymentStatus | Get the deployment status of the app |
| desired_state | app_definition_pb2.Spec.DesiredState | Get the desired state of the app. |
| url | str | Get the console URL for viewing the app. |
Constructor
Signature
def App(
pb2: app_definition_pb2.App
) - > null
Parameters
| Name | Type | Description |
|---|
| pb2 | app_definition_pb2.App | The protobuf message containing the app definition and status. |
Methods
name()
@classmethod
def name() - > str
Get the name of the app.
Returns
| Type | Description |
|---|
str | The unique identifier name of the application |
revision()
@classmethod
def revision() - > int
Get the revision number of the app.
Returns
| Type | Description |
|---|
int | The version sequence number for the current application deployment |
endpoint()
@classmethod
def endpoint() - > str
Get the public endpoint URL of the app.
Returns
| Type | Description |
|---|
str | The public-facing URL where the application is accessible |
deployment_status()
@classmethod
def deployment_status() - > app_definition_pb2.Status.DeploymentStatus
Get the deployment status of the app
Returns
| Type | Description |
|---|
app_definition_pb2.Status.DeploymentStatus | The current deployment lifecycle state of the application |
desired_state()
@classmethod
def desired_state() - > app_definition_pb2.Spec.DesiredState
Get the desired state of the app.
Returns
| Type | Description |
|---|
app_definition_pb2.Spec.DesiredState | The target state (e.g., started or stopped) requested for the application |
is_active()
@classmethod
def is_active() - > bool
Check if the app is currently active or started.
Returns
| Type | Description |
|---|
bool | True if the application is in an active deployment state |
is_deactivated()
@classmethod
def is_deactivated() - > bool
Check if the app is currently deactivated or stopped.
Returns
| Type | Description |
|---|
bool | True if the application is in a stopped or deactivated state |
url()
@classmethod
def url() - > str
Get the console URL for viewing the app.
Returns
| Type | Description |
|---|
str | The management console URL for the application based on its project and domain |
watch()
@classmethod
def watch(
wait_for: WaitFor = "activated"
) - > [App](app.md?sid=flyte_remote__app_app)
Watch for the app to reach activated or deactivated state.
Parameters
| Name | Type | Description |
|---|
| wait_for | WaitFor = "activated" | The target state to wait for, either "activated" or "deactivated" |
Returns
| Type | Description |
|---|
[App](app.md?sid=flyte_remote__app_app) | The app in the desired state. |
activate()
@classmethod
def activate(
wait: bool = false
) - > [App](app.md?sid=flyte_remote__app_app)
Start the app
Parameters
| Name | Type | Description |
|---|
| wait | bool = false | Whether to block execution until the app reaches the activated state |
Returns
| Type | Description |
|---|
[App](app.md?sid=flyte_remote__app_app) | The application instance after the activation request has been sent |
deactivate()
@classmethod
def deactivate(
wait: bool = false
) - > [App](app.md?sid=flyte_remote__app_app)
Stop the app
Parameters
| Name | Type | Description |
|---|
| wait | bool = false | Whether to block execution until the app reaches the deactivated state |
Returns
| Type | Description |
|---|
[App](app.md?sid=flyte_remote__app_app) | The application instance after the deactivation request has been sent |
ephemeral_ctx()
@classmethod
def ephemeral_ctx() - > AsyncGenerator[None, None]
Async context manager that activates the app and deactivates it when the context is exited.
Returns
| Type | Description |
|---|
AsyncGenerator[None, None] | An async generator for use in an 'async with' block |
ephemeral_ctx_sync()
@classmethod
def ephemeral_ctx_sync() - > Generator[None, None, None]
Context manager that activates the app and deactivates it when the context is exited.
Returns
| Type | Description |
|---|
Generator[None, None, None] | A generator for use in a standard 'with' block |
update()
@classmethod
def update(
updated_app_proto: app_definition_pb2.App,
reason: str
) - > [App](app.md?sid=flyte_remote__app_app)
Updates the application configuration on the server using a provided protobuf definition.
Parameters
| Name | Type | Description |
|---|
| updated_app_proto | app_definition_pb2.App | The full application protobuf message containing the new configuration |
| reason | str | Audit log message explaining the purpose of the update |
Returns
| Type | Description |
|---|
[App](app.md?sid=flyte_remote__app_app) | The updated application instance returned by the server |
delete()
@classmethod
def delete(
name: str,
project: str | None = null,
domain: str | None = null
)
Delete an app by name.
Parameters
| Name | Type | Description |
|---|
| name | str | The name of the app to delete. |
| project | `str | None` = null |
| domain | `str | None` = null |
replace()
@classmethod
def replace(
name: str,
updated_app_spec: app_definition_pb2.Spec,
reason: str,
labels: Mapping[str, str]| None = null,
project: str | None = null,
domain: str | None = null
) - > [App](app.md?sid=flyte_remote__app_app)
Replace an existing app's that matches the given name, with a new spec and optionally labels.
Parameters
| Name | Type | Description |
|---|
| name | str | Name of the new app |
| updated_app_spec | app_definition_pb2.Spec | Updated app spec |
| reason | str | The reason for replacing the application specification |
| labels | `Mapping[str, str] | None` = null |
| project | `str | None` = null |
| domain | `str | None` = null |
Returns
| Type | Description |
|---|
[App](app.md?sid=flyte_remote__app_app) | A new app |
get()
@classmethod
def get(
name: str,
project: str | None = null,
domain: str | None = null
) - > [App](app.md?sid=flyte_remote__app_app)
Get an app by name.
Parameters
| Name | Type | Description |
|---|
| name | str | The name of the app. |
| project | `str | None` = null |
| domain | `str | None` = null |
Returns
| Type | Description |
|---|
[App](app.md?sid=flyte_remote__app_app) | The app remote object. |
listall()
@classmethod
def listall(
created_by_subject: str | None = null,
sort_by: Tuple[str, Literal["asc", "desc"]]| None = null,
limit: int = 100
) - > AsyncIterator[[App](app.md?sid=flyte_remote__app_app)]
Retrieves a paginated list of applications based on filters and sorting criteria.
Parameters
| Name | Type | Description |
|---|
| created_by_subject | `str | None` = null |
| sort_by | `Tuple[str, Literal["asc", "desc"]] | None` = null |
| limit | int = 100 | The maximum number of applications to return |
Returns
| Type | Description |
|---|
AsyncIterator[[App](app.md?sid=flyte_remote__app_app)] | An asynchronous iterator yielding App instances |
create()
@classmethod
def create(
app: app_definition_pb2.App
) - > [App](app.md?sid=flyte_remote__app_app)
Creates a new application or updates an existing one if a conflict occurs.
Parameters
| Name | Type | Description |
|---|
| app | app_definition_pb2.App | The application definition to be deployed |
Returns
| Type | Description |
|---|
[App](app.md?sid=flyte_remote__app_app) | The newly created or updated application instance |