Skip to content

omnipy.hub.runtime

Runtime configuration and the global Omnipy runtime singleton.

The runtime wires together configuration, engines, serializers, registries, and user-interface integration for the current process. Most user code works with the module-level runtime object rather than constructing a Runtime manually.

ATTRIBUTE DESCRIPTION
runtime

Global Runtime singleton for the current Python process. It owns shared configuration and runtime objects, and it auto-initializes UI integration outside the test environment.

TYPE: IsRuntime

CLASS DESCRIPTION
Runtime

Application-level runtime object coordinating shared Omnipy services.
RuntimeConfig

Configuration tree owned by a Runtime instance.
RuntimeObjects

Concrete services and registries owned by a Runtime instance.

runtime module-attribute 

runtime: IsRuntime = Runtime()

Runtime

Bases: DataPublisher


              flowchart BT
              omnipy.hub.runtime.Runtime[Runtime]
              omnipy.util.publisher.DataPublisher[DataPublisher]

                              omnipy.util.publisher.DataPublisher --> omnipy.hub.runtime.Runtime
                                omnipy.util.pydantic.BaseModel --> omnipy.util.publisher.DataPublisher
                



              click omnipy.hub.runtime.Runtime href "" "omnipy.hub.runtime.Runtime"
              click omnipy.util.publisher.DataPublisher href "" "omnipy.util.publisher.DataPublisher"

Application-level runtime object coordinating shared Omnipy services.

A Runtime keeps configuration and runtime objects in sync, selects the active execution engine, and initializes display-related integration for the current environment. Most applications should use the module-level runtime singleton.

ATTRIBUTE DESCRIPTION
config

Runtime configuration object.

TYPE: IsRuntimeConfig

objects

Runtime service and registry container.

TYPE: IsRuntimeObjects

Used by:
METHOD DESCRIPTION
__init__

Initialize runtime data, UI integration, and subscriptions.
reset_backlinks

Set parent backlinks from runtime sub-objects to this runtime.
reset_subscriptions

Reset runtime subscriptions between config and runtime objects.
Source code in src/omnipy/hub/runtime.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
class Runtime(DataPublisher):
    """Application-level runtime object coordinating shared Omnipy services.

    A ``Runtime`` keeps configuration and runtime objects in sync, selects the
    active execution engine, and initializes display-related integration for
    the current environment. Most applications should use the module-level
    ``runtime`` singleton.

    Attributes:
        config: Runtime configuration object.
        objects: Runtime service and registry container.

    """

    config: IsRuntimeConfig = pyd.Field(default_factory=RuntimeConfig)
    objects: IsRuntimeObjects = pyd.Field(default_factory=RuntimeObjects)

    def __init__(self, **data: Any) -> None:
        """Initialize runtime data, UI integration, and subscriptions.

        Args:
            **data: Optional initialization data passed to ``DataPublisher``.
        """

        super().__init__(**data)
        detect_and_setup_user_interface(self)
        self.reset_subscriptions()

    def reset_subscriptions(self) -> None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISRUNTIME_RESET_SUBSCRIPTIONS_SUMMARY}}
        #
        # {{ISRUNTIME_RESET_SUBSCRIPTIONS_DETAILS}}
        """Reset runtime subscriptions between config and runtime objects.

        This method rebuilds the callback graph that keeps configuration, engines, registries,
        logging, and reactive UI objects synchronized. Call it after replacing runtime subobjects
        manually.

        Raises:
            AssertionError: If a Jupyter UI is detected but reactive objects are unexpectedly
                missing.
        """

        self.reset_backlinks()

        self.config.unsubscribe_all()
        self.objects.unsubscribe_all()

        # Makes sure that the config references in the runtime objects always refer to the related
        # config in runtime, even when:
        #
        # 1. The runtime config is replaced with a new config object, or
        # 2. The runtime object is replaced with a new runtime object, or
        # 3. Both of the above
        #
        # This might e.g. happen due to the use of  mock objects for testing, or if the config is
        # reloaded from a file.

        self.config.subscribe_attr('data', self.objects.data_class_creator.set_config)
        self.config.subscribe_attr('job', self.objects.job_creator.set_config)
        self.config.subscribe_attr('root_log', self.objects.root_log.set_config)

        self.config.data.ui.subscribe_attr('detected_type', self.objects.setup_reactive)

        if UserInterfaceType.is_jupyter_in_browser(self.config.data.ui.detected_type):
            assert self.objects.reactive is not None
            self.config.data.ui.subscribe_attr('jupyter',
                                               self.objects.reactive.jupyter_ui_config.set)
            self.config.data.ui.subscribe_attr('text', self.objects.reactive.text_config.set)
            self.config.data.ui.subscribe_attr('layout', self.objects.reactive.layout_config.set)

        self.config.engine.subscribe_attr('local', self.objects.local.set_config)
        self.config.engine.subscribe_attr('prefect', self.objects.prefect.set_config)

        self.objects.subscribe_attr(
            'reactive',
            self.objects.data_class_creator.set_reactive_objects,
        )

        # Makes sure that the registry references in the job runners always refer to the registry
        # object in runtime, even when one or both of the objects are replaced with new objects.
        self.objects.subscribe_attr('registry', self.objects.local.set_registry)
        self.objects.subscribe_attr('registry', self.objects.prefect.set_registry)

        # Makes sure that the engine used by the job creator is always the one specified in the
        # 'engine' config item in runtime
        self.config.engine.subscribe_attr('local', self._update_job_creator_engine)
        self.config.engine.subscribe_attr('prefect', self._update_job_creator_engine)
        self.config.engine.subscribe_attr('choice', self._update_job_creator_engine)

        # Makes sure that the local and prefect engine configs are always reloaded when the local
        # or prefect engine objects are replaced with new engine objects. This is necessary because
        # the `get_config_cls()` method of the engine objects define the classes of the respective
        # engine config objects to be used. If an engine object is replaced with a new engine that
        # require a different engine config class than is currently used, the config object will be
        # replaced with a new default config object of the correct type.
        self.objects.subscribe_attr('local', self._update_local_runner_config)
        self.objects.subscribe_attr('prefect', self._update_prefect_engine_config)

    def reset_backlinks(self) -> None:
        """Set parent backlinks from runtime sub-objects to this runtime."""

        self.config._back = self  # type: ignore[attr-defined]
        self.objects._back = self  # type: ignore[attr-defined]

    def _get_engine_config(self, choice: EngineChoice.Literals) -> IsEngineConfig:
        """Return the engine configuration object for a selected engine.

        Args:
            choice: Engine identifier whose configuration should be returned.

        Returns:
            IsEngineConfig: Configuration object for the selected engine.

        Raises:
            AttributeError: If ``choice`` does not exist on ``config.engine``.
        """

        return getattr(self.config.engine, choice)

    def _set_engine_config(
        self,
        choice: EngineChoice.Literals,
        engine_config: IsJobRunnerConfig,
    ) -> None:
        """Set the configuration object for a selected engine.

        Args:
            choice: Engine identifier whose configuration should be updated.
            engine_config: New engine configuration object.

        Raises:
            AttributeError: If ``choice`` does not exist on ``config.engine``.
        """

        setattr(self.config.engine, choice, engine_config)

    def _get_engine(self, choice: EngineChoice.Literals) -> IsEngine:
        """Return the engine object for a selected engine choice.

        Args:
            choice: Engine identifier whose runtime engine should be returned.

        Returns:
            IsEngine: Runtime engine object for ``choice``.

        Raises:
            AttributeError: If ``choice`` does not exist on ``objects``.
        """

        return getattr(self.objects, choice)

    def _new_engine_config_if_new_cls(
        self,
        engine: IsEngine,
        choice: EngineChoice.Literals,
    ) -> None:
        """Replace engine config when an engine requires a new config class.

        Args:
            engine: Engine instance to inspect for config class changes.
            choice: Engine identifier whose config may need replacement.
        """

        # TODO: when parsing config from file is implemented, make sure that the new engine
        #       config classes here reparse the config files
        engine_config_cls = engine.get_config_cls()
        if self._get_engine_config(choice).__class__ is not engine_config_cls:
            self._set_engine_config(choice, engine_config_cls())

    def _update_local_runner_config(self, local_runner: IsEngine) -> None:
        """Refresh local engine config when the local engine object changes.

        Args:
            local_runner: New local engine instance.
        """

        self._new_engine_config_if_new_cls(local_runner, EngineChoice.LOCAL)

    def _update_prefect_engine_config(self, prefect_engine: IsEngine) -> None:
        """Refresh Prefect engine config when the Prefect engine changes.

        Args:
            prefect_engine: New Prefect engine instance.
        """

        self._new_engine_config_if_new_cls(prefect_engine, EngineChoice.PREFECT)

    def _update_job_creator_engine(self, _item_changed: Any) -> None:
        """Sync the job creator with the currently selected runtime engine.

        Args:
            _item_changed: Subscription payload for the triggering change.

        Raises:
            AttributeError: If the configured engine choice has no matching
                runtime engine object.
        """

        self.objects.job_creator.set_engine(self._get_engine(self.config.engine.choice))

config class-attribute instance-attribute 

config: IsRuntimeConfig = pyd.Field(default_factory=RuntimeConfig)

objects class-attribute instance-attribute 

objects: IsRuntimeObjects = pyd.Field(default_factory=RuntimeObjects)

__init__ 

__init__(**data: Any) -> None

Initialize runtime data, UI integration, and subscriptions.

PARAMETER DESCRIPTION
**data

Optional initialization data passed to DataPublisher.

TYPE: Any DEFAULT: {}

Source code in src/omnipy/hub/runtime.py 
194
195
196
197
198
199
200
201
202
203
def __init__(self, **data: Any) -> None:
    """Initialize runtime data, UI integration, and subscriptions.

    Args:
        **data: Optional initialization data passed to ``DataPublisher``.
    """

    super().__init__(**data)
    detect_and_setup_user_interface(self)
    self.reset_subscriptions()

reset_backlinks() -> None

Set parent backlinks from runtime sub-objects to this runtime.

Source code in src/omnipy/hub/runtime.py 
277
278
279
280
281
def reset_backlinks(self) -> None:
    """Set parent backlinks from runtime sub-objects to this runtime."""

    self.config._back = self  # type: ignore[attr-defined]
    self.objects._back = self  # type: ignore[attr-defined]

reset_subscriptions 

reset_subscriptions() -> None

Reset runtime subscriptions between config and runtime objects.

This method rebuilds the callback graph that keeps configuration, engines, registries, logging, and reactive UI objects synchronized. Call it after replacing runtime subobjects manually.

RAISES DESCRIPTION
AssertionError

If a Jupyter UI is detected but reactive objects are unexpectedly missing.
Source code in src/omnipy/hub/runtime.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def reset_subscriptions(self) -> None:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISRUNTIME_RESET_SUBSCRIPTIONS_SUMMARY}}
    #
    # {{ISRUNTIME_RESET_SUBSCRIPTIONS_DETAILS}}
    """Reset runtime subscriptions between config and runtime objects.

    This method rebuilds the callback graph that keeps configuration, engines, registries,
    logging, and reactive UI objects synchronized. Call it after replacing runtime subobjects
    manually.

    Raises:
        AssertionError: If a Jupyter UI is detected but reactive objects are unexpectedly
            missing.
    """

    self.reset_backlinks()

    self.config.unsubscribe_all()
    self.objects.unsubscribe_all()

    # Makes sure that the config references in the runtime objects always refer to the related
    # config in runtime, even when:
    #
    # 1. The runtime config is replaced with a new config object, or
    # 2. The runtime object is replaced with a new runtime object, or
    # 3. Both of the above
    #
    # This might e.g. happen due to the use of  mock objects for testing, or if the config is
    # reloaded from a file.

    self.config.subscribe_attr('data', self.objects.data_class_creator.set_config)
    self.config.subscribe_attr('job', self.objects.job_creator.set_config)
    self.config.subscribe_attr('root_log', self.objects.root_log.set_config)

    self.config.data.ui.subscribe_attr('detected_type', self.objects.setup_reactive)

    if UserInterfaceType.is_jupyter_in_browser(self.config.data.ui.detected_type):
        assert self.objects.reactive is not None
        self.config.data.ui.subscribe_attr('jupyter',
                                           self.objects.reactive.jupyter_ui_config.set)
        self.config.data.ui.subscribe_attr('text', self.objects.reactive.text_config.set)
        self.config.data.ui.subscribe_attr('layout', self.objects.reactive.layout_config.set)

    self.config.engine.subscribe_attr('local', self.objects.local.set_config)
    self.config.engine.subscribe_attr('prefect', self.objects.prefect.set_config)

    self.objects.subscribe_attr(
        'reactive',
        self.objects.data_class_creator.set_reactive_objects,
    )

    # Makes sure that the registry references in the job runners always refer to the registry
    # object in runtime, even when one or both of the objects are replaced with new objects.
    self.objects.subscribe_attr('registry', self.objects.local.set_registry)
    self.objects.subscribe_attr('registry', self.objects.prefect.set_registry)

    # Makes sure that the engine used by the job creator is always the one specified in the
    # 'engine' config item in runtime
    self.config.engine.subscribe_attr('local', self._update_job_creator_engine)
    self.config.engine.subscribe_attr('prefect', self._update_job_creator_engine)
    self.config.engine.subscribe_attr('choice', self._update_job_creator_engine)

    # Makes sure that the local and prefect engine configs are always reloaded when the local
    # or prefect engine objects are replaced with new engine objects. This is necessary because
    # the `get_config_cls()` method of the engine objects define the classes of the respective
    # engine config objects to be used. If an engine object is replaced with a new engine that
    # require a different engine config class than is currently used, the config object will be
    # replaced with a new default config object of the correct type.
    self.objects.subscribe_attr('local', self._update_local_runner_config)
    self.objects.subscribe_attr('prefect', self._update_prefect_engine_config)

RuntimeConfig

Bases: RuntimeEntryPublisher, ConfigBase


              flowchart BT
              omnipy.hub.runtime.RuntimeConfig[RuntimeConfig]
              omnipy.util.publisher.RuntimeEntryPublisher[RuntimeEntryPublisher]
              omnipy.config.ConfigBase[ConfigBase]
              omnipy.util.publisher.DataPublisher[DataPublisher]

                              omnipy.util.publisher.RuntimeEntryPublisher --> omnipy.hub.runtime.RuntimeConfig
                                omnipy.util.publisher.DataPublisher --> omnipy.util.publisher.RuntimeEntryPublisher
                                omnipy.util.pydantic.BaseModel --> omnipy.util.publisher.DataPublisher
                


                omnipy.config.ConfigBase --> omnipy.hub.runtime.RuntimeConfig
                                omnipy.util.publisher.DataPublisher --> omnipy.config.ConfigBase
                                omnipy.util.pydantic.BaseModel --> omnipy.util.publisher.DataPublisher
                




              click omnipy.hub.runtime.RuntimeConfig href "" "omnipy.hub.runtime.RuntimeConfig"
              click omnipy.util.publisher.RuntimeEntryPublisher href "" "omnipy.util.publisher.RuntimeEntryPublisher"
              click omnipy.config.ConfigBase href "" "omnipy.config.ConfigBase"
              click omnipy.util.publisher.DataPublisher href "" "omnipy.util.publisher.DataPublisher"

Configuration tree owned by a Runtime instance.

ATTRIBUTE DESCRIPTION
data

Data-related runtime settings.

TYPE: IsDataConfig

engine

Engine selection and per-engine configuration.

TYPE: IsEngineConfig

job

Job creation and execution settings.

TYPE: IsJobConfig

root_log

Root logger integration settings.

TYPE: IsRootLogConfig

Used by:
METHOD DESCRIPTION
reset_to_defaults

Reset all runtime configuration sections to their default values.
Source code in src/omnipy/hub/runtime.py 
 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
class RuntimeConfig(RuntimeEntryPublisher, ConfigBase):
    """Configuration tree owned by a ``Runtime`` instance.

    Attributes:
        data: Data-related runtime settings.
        engine: Engine selection and per-engine configuration.
        job: Job creation and execution settings.
        root_log: Root logger integration settings.

    """

    data: IsDataConfig = pyd.Field(default_factory=_data_config_factory)
    engine: IsEngineConfig = pyd.Field(default_factory=EngineConfig)
    job: IsJobConfig = pyd.Field(default_factory=_job_config_factory)
    root_log: IsRootLogConfig = pyd.Field(default_factory=RootLogConfig)

    def reset_to_defaults(self) -> None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISRUNTIMECONFIG_RESET_TO_DEFAULTS_SUMMARY}}
        #
        # {{ISRUNTIMECONFIG_RESET_TO_DEFAULTS_DETAILS}}
        """Reset all runtime configuration sections to their default values.

        Rebuilds the data, engine, job, and root-log config sections and then refreshes runtime
        subscriptions when the config is attached to a runtime object.
        """

        prev_back = self._back
        self._back = None

        self.data = cast(IsDataConfig, DataConfig())
        self.engine = cast(IsEngineConfig, EngineConfig())
        self.job = cast(IsJobConfig, JobConfig())
        self.root_log = cast(IsRootLogConfig, RootLogConfig())

        self._back = prev_back
        if self._back is not None:
            self._back.reset_subscriptions()

data class-attribute instance-attribute 

data: IsDataConfig = pyd.Field(default_factory=_data_config_factory)

engine class-attribute instance-attribute 

engine: IsEngineConfig = pyd.Field(default_factory=EngineConfig)

job class-attribute instance-attribute 

job: IsJobConfig = pyd.Field(default_factory=_job_config_factory)

root_log class-attribute instance-attribute 

root_log: IsRootLogConfig = pyd.Field(default_factory=RootLogConfig)

reset_to_defaults 

reset_to_defaults() -> None

Reset all runtime configuration sections to their default values.

Rebuilds the data, engine, job, and root-log config sections and then refreshes runtime subscriptions when the config is attached to a runtime object.

Source code in src/omnipy/hub/runtime.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def reset_to_defaults(self) -> None:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISRUNTIMECONFIG_RESET_TO_DEFAULTS_SUMMARY}}
    #
    # {{ISRUNTIMECONFIG_RESET_TO_DEFAULTS_DETAILS}}
    """Reset all runtime configuration sections to their default values.

    Rebuilds the data, engine, job, and root-log config sections and then refreshes runtime
    subscriptions when the config is attached to a runtime object.
    """

    prev_back = self._back
    self._back = None

    self.data = cast(IsDataConfig, DataConfig())
    self.engine = cast(IsEngineConfig, EngineConfig())
    self.job = cast(IsJobConfig, JobConfig())
    self.root_log = cast(IsRootLogConfig, RootLogConfig())

    self._back = prev_back
    if self._back is not None:
        self._back.reset_subscriptions()

RuntimeObjects

Bases: RuntimeEntryPublisher, DataPublisher


              flowchart BT
              omnipy.hub.runtime.RuntimeObjects[RuntimeObjects]
              omnipy.util.publisher.RuntimeEntryPublisher[RuntimeEntryPublisher]
              omnipy.util.publisher.DataPublisher[DataPublisher]

                              omnipy.util.publisher.RuntimeEntryPublisher --> omnipy.hub.runtime.RuntimeObjects
                                omnipy.util.publisher.DataPublisher --> omnipy.util.publisher.RuntimeEntryPublisher
                                omnipy.util.pydantic.BaseModel --> omnipy.util.publisher.DataPublisher
                


                omnipy.util.publisher.DataPublisher --> omnipy.hub.runtime.RuntimeObjects
                                omnipy.util.pydantic.BaseModel --> omnipy.util.publisher.DataPublisher
                



              click omnipy.hub.runtime.RuntimeObjects href "" "omnipy.hub.runtime.RuntimeObjects"
              click omnipy.util.publisher.RuntimeEntryPublisher href "" "omnipy.util.publisher.RuntimeEntryPublisher"
              click omnipy.util.publisher.DataPublisher href "" "omnipy.util.publisher.DataPublisher"

Concrete services and registries owned by a Runtime instance.

ATTRIBUTE DESCRIPTION
job_creator

Shared job creator configured by RuntimeConfig.job.

TYPE: IsJobConfigHolder

data_class_creator

Shared data creator configured by RuntimeConfig.data.

TYPE: IsDataClassCreator

reactive

Optional UI-reactive helper objects for notebook contexts.

TYPE: IsReactiveObjects | None

local

Local execution engine.

TYPE: IsEngine

prefect

Prefect execution engine.

TYPE: IsEngine

registry

Runtime run-state registry.

TYPE: IsRunStateRegistry

serializers

Dataset serializer registry.

TYPE: IsSerializerRegistry

root_log

Root logging integration objects.

TYPE: IsRootLogObjects

Used by:
METHOD DESCRIPTION
setup_reactive

Create or remove reactive UI helpers for the detected interface.
Source code in src/omnipy/hub/runtime.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class RuntimeObjects(RuntimeEntryPublisher, DataPublisher):
    """Concrete services and registries owned by a ``Runtime`` instance.

    Attributes:
        job_creator: Shared job creator configured by ``RuntimeConfig.job``.
        data_class_creator: Shared data creator configured by
            ``RuntimeConfig.data``.
        reactive: Optional UI-reactive helper objects for notebook contexts.
        local: Local execution engine.
        prefect: Prefect execution engine.
        registry: Runtime run-state registry.
        serializers: Dataset serializer registry.
        root_log: Root logging integration objects.

    """

    job_creator: IsJobConfigHolder = pyd.Field(default_factory=_job_creator_factory)
    data_class_creator: IsDataClassCreator = pyd.Field(default_factory=_data_class_creator_factory)
    reactive: IsReactiveObjects | None = None
    local: IsEngine = pyd.Field(default_factory=LocalRunner)
    prefect: IsEngine = pyd.Field(default_factory=PrefectEngine)
    registry: IsRunStateRegistry = pyd.Field(default_factory=RunStateRegistry)
    serializers: IsSerializerRegistry = pyd.Field(default_factory=SerializerRegistry)
    root_log: IsRootLogObjects = pyd.Field(default_factory=RootLogObjects)

    def setup_reactive(self, ui_type: UserInterfaceType.Literals) -> None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISRUNTIMEOBJECTS_SETUP_REACTIVE_SUMMARY}}
        #
        # {{ISRUNTIMEOBJECTS_SETUP_REACTIVE_DETAILS}}
        """Create or remove reactive UI helpers for the detected interface.

        Args:
            ui_type: Detected user-interface type for the current runtime.
        """

        if UserInterfaceType.is_jupyter_in_browser(ui_type):
            from omnipy.data._display.integrations.jupyter.helpers import ReactiveObjects

            if self.reactive is None:
                self.reactive = ReactiveObjects()
        else:
            if self.reactive is not None:
                self.reactive = None

data_class_creator class-attribute instance-attribute 

data_class_creator: IsDataClassCreator = pyd.Field(default_factory=_data_class_creator_factory)

job_creator class-attribute instance-attribute 

job_creator: IsJobConfigHolder = pyd.Field(default_factory=_job_creator_factory)

local class-attribute instance-attribute 

local: IsEngine = pyd.Field(default_factory=LocalRunner)

prefect class-attribute instance-attribute 

prefect: IsEngine = pyd.Field(default_factory=PrefectEngine)

reactive class-attribute instance-attribute 

reactive: IsReactiveObjects | None = None

registry class-attribute instance-attribute 

registry: IsRunStateRegistry = pyd.Field(default_factory=RunStateRegistry)

root_log class-attribute instance-attribute 

root_log: IsRootLogObjects = pyd.Field(default_factory=RootLogObjects)

serializers class-attribute instance-attribute 

serializers: IsSerializerRegistry = pyd.Field(default_factory=SerializerRegistry)

setup_reactive 

setup_reactive(ui_type: UserInterfaceType.Literals) -> None

Create or remove reactive UI helpers for the detected interface.

PARAMETER DESCRIPTION
ui_type

Detected user-interface type for the current runtime.

TYPE: UserInterfaceType.Literals

Source code in src/omnipy/hub/runtime.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def setup_reactive(self, ui_type: UserInterfaceType.Literals) -> None:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISRUNTIMEOBJECTS_SETUP_REACTIVE_SUMMARY}}
    #
    # {{ISRUNTIMEOBJECTS_SETUP_REACTIVE_DETAILS}}
    """Create or remove reactive UI helpers for the detected interface.

    Args:
        ui_type: Detected user-interface type for the current runtime.
    """

    if UserInterfaceType.is_jupyter_in_browser(ui_type):
        from omnipy.data._display.integrations.jupyter.helpers import ReactiveObjects

        if self.reactive is None:
            self.reactive = ReactiveObjects()
    else:
        if self.reactive is not None:
            self.reactive = None