Skip to content

App

AppState

OpsML application state object. This is typically used in API workflows where you wish create a shared state to share among all requests. The OpsML app state provides a convenient way to load and store artifacts. Most notably, it provides an integration with Scouter so that you can load a ServiceCard along with a ScouterQueue for drift detection. Future iterations of this class may include other convenience methods that simplify common API tasks.

Source code in python/opsml/app/_app.pyi 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class AppState:
    """OpsML application state object. This is typically used in API
    workflows where you wish create a shared state to share among all requests.
    The OpsML app state provides a convenient way to load and store artifacts.
    Most notably, it provides an integration with Scouter so that you can load a `ServiceCard`
    along with a `ScouterQueue` for drift detection. Future iterations of this class may
    include other convenience methods that simplify common API tasks.
    """

    @staticmethod
    def from_path(
        path: Path,
        transport_config: Optional[
            Union[
                KafkaConfig,
                RabbitMQConfig,
                RedisConfig,
                HTTPConfig,
            ]
        ] = None,
        reload_config: Optional[ReloadConfig] = None,
        load_kwargs: Optional[Dict[str, Dict[str, Any]]] = None,
    ) -> "AppState":
        """
        Load the AppState from a file path.

        Args:
            path (str):
                The file path to load the AppState from. This is typically the path
                pointing to the directory containing the `ServiceCard`.

            transport_config (KafkaConfig | RabbitMQConfig | RedisConfig | HTTPConfig | None):
                Optional transport configuration for the queue publisher
                Can be KafkaConfig, RabbitMQConfig RedisConfig, or HTTPConfig. If not provided,
                the queue will not be initialized.

            load_kwargs (Dict[str, Dict[str, Any]]):
                Optional kwargs for loading cards. Expected format:
                {
                    "card_alias": {
                        "interface": interface_object,
                        "load_kwargs": DataLoadKwargs | ModelLoadKwargs
                    }
                }

        Example:
            ```python
            from opsml.app import AppState
            from opsml.scouter import KafkaConfig

            app_state = AppState.from_path(
                "/path/to/service",
                transport_config=KafkaConfig(),
                )

            # Access the service card and queue
            service = app_state.service
            queue = app_state.queue
            ```

        Returns:
            AppState: The loaded AppState.
        """

    @property
    def service(self) -> ServiceCard:
        """Get the service card."""

    @property
    def queue(self) -> ScouterQueue:
        """Get the Scouter queue."""

    @property
    def reloader_running(self) -> bool:
        """Check if the ServiceReloader is currently running."""

    def reload(self) -> None:
        """Forces `AppState` to check for new `ServiceCards` and reload if necessary."""

    def start_reloader(self) -> None:
        """Starts the `AppState` reloader."""

    def stop_reloader(self) -> None:
        """Stops the `AppState` reloader."""

    def shutdown(self) -> None:
        """Shuts down the `AppState` `ScouterQueue` and reloader if running.
        This is a destructive operation and will attempt to close all background threads
        associated with the `ScouterQueue` and reloader. Only use this method with graceful
        shutdown procedures in mind.
        """

queue property

Get the Scouter queue.

reloader_running property

Check if the ServiceReloader is currently running.

service property

Get the service card.

from_path(path, transport_config=None, reload_config=None, load_kwargs=None) staticmethod

Load the AppState from a file path.

Parameters:

Name Type Description Default
path str

The file path to load the AppState from. This is typically the path pointing to the directory containing the ServiceCard.

 required
transport_config KafkaConfig | RabbitMQConfig | RedisConfig | HTTPConfig | None

Optional transport configuration for the queue publisher Can be KafkaConfig, RabbitMQConfig RedisConfig, or HTTPConfig. If not provided, the queue will not be initialized.

 None
load_kwargs Dict[str, Dict[str, Any]]

Optional kwargs for loading cards. Expected format: { "card_alias": { "interface": interface_object, "load_kwargs": DataLoadKwargs | ModelLoadKwargs } }

 None
Example 
from opsml.app import AppState
from opsml.scouter import KafkaConfig

app_state = AppState.from_path(
    "/path/to/service",
    transport_config=KafkaConfig(),
    )

# Access the service card and queue
service = app_state.service
queue = app_state.queue

Returns:

Name Type Description
AppState AppState

The loaded AppState.
Source code in python/opsml/app/_app.pyi 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@staticmethod
def from_path(
    path: Path,
    transport_config: Optional[
        Union[
            KafkaConfig,
            RabbitMQConfig,
            RedisConfig,
            HTTPConfig,
        ]
    ] = None,
    reload_config: Optional[ReloadConfig] = None,
    load_kwargs: Optional[Dict[str, Dict[str, Any]]] = None,
) -> "AppState":
    """
    Load the AppState from a file path.

    Args:
        path (str):
            The file path to load the AppState from. This is typically the path
            pointing to the directory containing the `ServiceCard`.

        transport_config (KafkaConfig | RabbitMQConfig | RedisConfig | HTTPConfig | None):
            Optional transport configuration for the queue publisher
            Can be KafkaConfig, RabbitMQConfig RedisConfig, or HTTPConfig. If not provided,
            the queue will not be initialized.

        load_kwargs (Dict[str, Dict[str, Any]]):
            Optional kwargs for loading cards. Expected format:
            {
                "card_alias": {
                    "interface": interface_object,
                    "load_kwargs": DataLoadKwargs | ModelLoadKwargs
                }
            }

    Example:
        ```python
        from opsml.app import AppState
        from opsml.scouter import KafkaConfig

        app_state = AppState.from_path(
            "/path/to/service",
            transport_config=KafkaConfig(),
            )

        # Access the service card and queue
        service = app_state.service
        queue = app_state.queue
        ```

    Returns:
        AppState: The loaded AppState.
    """

reload()

Forces AppState to check for new ServiceCards and reload if necessary.

Source code in python/opsml/app/_app.pyi 
120
121
def reload(self) -> None:
    """Forces `AppState` to check for new `ServiceCards` and reload if necessary."""

shutdown()

Shuts down the AppState ScouterQueue and reloader if running. This is a destructive operation and will attempt to close all background threads associated with the ScouterQueue and reloader. Only use this method with graceful shutdown procedures in mind.

Source code in python/opsml/app/_app.pyi 
129
130
131
132
133
134
def shutdown(self) -> None:
    """Shuts down the `AppState` `ScouterQueue` and reloader if running.
    This is a destructive operation and will attempt to close all background threads
    associated with the `ScouterQueue` and reloader. Only use this method with graceful
    shutdown procedures in mind.
    """

start_reloader()

Starts the AppState reloader.

Source code in python/opsml/app/_app.pyi 
123
124
def start_reloader(self) -> None:
    """Starts the `AppState` reloader."""

stop_reloader()

Stops the AppState reloader.

Source code in python/opsml/app/_app.pyi 
126
127
def stop_reloader(self) -> None:
    """Stops the `AppState` reloader."""

ReloadConfig

Reload configuation to use with an Opsml AppState object. Defines the reload logic for checking, downloading and reloading ServiceCards and ScouterQueues associated with an AppState

Source code in python/opsml/app/_app.pyi 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class ReloadConfig:
    """Reload configuation to use with an Opsml AppState object. Defines the reload logic
    for checking, downloading and reloading ServiceCards and ScouterQueues associated with
    an AppState
    """

    def __init__(
        self,
        cron: str,
        max_retries: int = 3,
        write_path: Optional[Path] = None,
    ):
        """Initialize the reload configuration.

        Args:
            cron (str):
                The cron expression for the reload schedule.
            max_retries (int):
                The maximum number of retries for loading the service card.
                Defaults to 3.
            write_path (Optional[Path]):
                The optional path to write the service card. Defaults to Path({current directory})/ service_reload
        """
        ...

    @property
    def cron(self) -> str:
        """Get the cron expression for the reload schedule."""

    @cron.setter
    def cron(self, value: str):
        """Set the cron expression for the reload schedule."""

cron property writable

Get the cron expression for the reload schedule.

__init__(cron, max_retries=3, write_path=None)

Initialize the reload configuration.

Parameters:

Name Type Description Default
cron str

The cron expression for the reload schedule.

 required
max_retries int

The maximum number of retries for loading the service card. Defaults to 3.

 3
write_path Optional[Path]

The optional path to write the service card. Defaults to Path({current directory})/ service_reload

 None
Source code in python/opsml/app/_app.pyi 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def __init__(
    self,
    cron: str,
    max_retries: int = 3,
    write_path: Optional[Path] = None,
):
    """Initialize the reload configuration.

    Args:
        cron (str):
            The cron expression for the reload schedule.
        max_retries (int):
            The maximum number of retries for loading the service card.
            Defaults to 3.
        write_path (Optional[Path]):
            The optional path to write the service card. Defaults to Path({current directory})/ service_reload
    """
    ...