Function initialize

src/zenml/zen_server/zen_server_api.py:194–226

Initialize the ZenML server.

()

@app.on_event("startup")
async def initialize() -> None:
    """Initialize the ZenML server."""
    cfg = server_config()
    # Set the maximum number of worker threads
    to_thread.current_default_thread_limiter().total_tokens = (
        cfg.thread_pool_size
    )
    # IMPORTANT: these need to be run before the fastapi app starts, to avoid
    # race conditions
    await initialize_request_manager()
    initialize_zen_store()
    # Instrument the SQL store with OpenTelemetry after it has been initialized.
    instrument_sqlalchemy_store(store=zen_store())
    initialize_resource_pool_store()
    service_connector_registry.register_builtin_service_connectors()
    initialize_rbac()
    initialize_feature_gate()
    initialize_workload_manager()
    initialize_resource_pool_store()
    initialize_snapshot_executor()
    await initialize_snapshot_run_dispatcher()
    initialize_artifact_store_cache()
    await initialize_streaming()
    initialize_secure_headers()
    if cfg.deployment_type == ServerDeploymentType.CLOUD:
        # Send a workspace status update to the Cloud API to indicate that the
        # ZenML server is running or to update the version and server URL.
        send_pro_workspace_status_update()

    if logger.isEnabledFor(logging.DEBUG):
        start_event_loop_lag_monitor()

    await register_event_handlers()


@app.on_event("shutdown")