expose the execution backend and MicroVM settings in the admin UI

Configuring where a transform runs meant setting values through the API, with
nothing in the UI to say the choice exists.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

.clj-kondo/config/modules/config.edn

@@ -3765,6 +3765,7 @@
    :model-imports #{:model/Database
                     :model/Field
                     :model/Table
+                    :model/Transform
                     :model/TransformRun}}
 
   enterprise/upload-management

.github/scripts/setup-python-runner.sh

@@ -18,8 +18,12 @@ while [[ $# -gt 0 ]]; do
   esac
 done
 
-echo "Cloning python-runner repository..."
-git clone --depth 1 https://${METABASE_AUTOMATION_USER_TOKEN}@github.com/metabase/python-runner-container.git python-runner
+# The runner branch this Metabase branch expects. Point at `main` once the MicroVM work merges.
+PYTHON_RUNNER_BRANCH="${PYTHON_RUNNER_BRANCH:-microvm-executor}"
+
+echo "Cloning python-runner repository (branch: ${PYTHON_RUNNER_BRANCH})..."
+git clone --depth 1 --branch "${PYTHON_RUNNER_BRANCH}" \
+  https://${METABASE_AUTOMATION_USER_TOKEN}@github.com/metabase/python-runner-container.git python-runner
 
 if [ "$SKIP_START" = false ]; then
   echo "Starting python-runner..."

enterprise/backend/src/metabase_enterprise/transforms_python/api.clj

@@ -1,8 +1,8 @@
 (ns metabase-enterprise.transforms-python.api
   (:require
    [clojure.string :as str]
+   [metabase-enterprise.transforms-python.executor :as executor]
    [metabase-enterprise.transforms-python.models.python-library :as python-library]
-   [metabase-enterprise.transforms-python.python-runner :as python-runner]
    [metabase-enterprise.transforms-python.settings :as transforms-python.settings]
    [metabase.api.common :as api]
    [metabase.api.macros :as api.macros]
@@ -10,6 +10,7 @@
    [metabase.api.util.handlers :as handlers]
    [metabase.permissions.core :as perms]
    [metabase.transforms-base.util :as transforms-base.u]
+   [metabase.transforms.core :as transforms.core]
    [metabase.util.i18n :as i18n]
    [metabase.util.malli.schema :as ms]
    [toucan2.core :as t2]))
@@ -58,22 +59,35 @@
    _
    {:keys [code
            source_tables
+           ingestion
+           transform_id
            output_row_limit
            per_input_row_limit]
     :or   {output_row_limit    100
            per_input_row_limit 100}}
    :- [:map
        [:code                                 :string]
-       [:source_tables                        [:sequential {:min 1} ::transforms-base.u/source-table-entry]]
+       ;; an ingestion transform has none: it materializes its own input
+       [:source_tables                        [:sequential ::transforms-base.u/source-table-entry]]
+       [:ingestion           {:optional true} [:maybe :boolean]]
+       ;; when the draft belongs to a saved transform, its stored secrets are used for the preview
+       [:transform_id        {:optional true} [:maybe ms/PositiveInt]]
        [:output_row_limit    {:optional true} [:and :int [:> 1] [:<= 100]]]
        [:per_input_row_limit {:optional true} [:and :int [:> 1] [:<= 100]]]]]
-  (let [db-ids (t2/select-fn-set :db_id [:model/Table :db_id] :id [:in (map :table_id source_tables)])]
-    (api/check-400 (= (count db-ids) 1) (i18n/deferred-tru "All source tables must belong to the same database."))
-    (api/check-403 (perms/has-db-transforms-permission? api/*current-user-id* (first db-ids))))
+  (if (empty? source_tables)
+    (api/check-403 (perms/has-any-transforms-permission? api/*current-user-id*))
+    (let [db-ids (t2/select-fn-set :db_id [:model/Table :db_id] :id [:in (map :table_id source_tables)])]
+      (api/check-400 (= (count db-ids) 1) (i18n/deferred-tru "All source tables must belong to the same database."))
+      (api/check-403 (perms/has-db-transforms-permission? api/*current-user-id* (first db-ids)))))
   ;; NOTE: we do not test database support, as there is no write target.
-  (let [result (python-runner/execute-and-read-output!
+  (let [secrets (when transform_id
+                  (api/write-check :model/Transform transform_id)
+                  (transforms.core/secrets-for-run transform_id))
+        result (executor/execute-and-read-output!
                 {:code            code
                  :source-tables   source_tables
+                 :ingestion?      (boolean ingestion)
+                 :secrets         secrets
                  :per-input-limit per_input_row_limit
                  :row-limit       output_row_limit
                  :timeout-secs    (transforms-python.settings/python-runner-test-run-timeout-seconds)})

enterprise/backend/src/metabase_enterprise/transforms_python/base.clj

@@ -8,9 +8,9 @@
    [clojure.core.async :as a]
    [clojure.java.io :as io]
    [clojure.string :as str]
+   [metabase-enterprise.transforms-python.executor :as executor]
    [metabase-enterprise.transforms-python.python-runner :as python-runner]
    [metabase-enterprise.transforms-python.s3 :as s3]
-   [metabase-enterprise.transforms-python.settings :as transforms-python.settings]
    [metabase.driver :as driver]
    [metabase.driver.sql :as driver.sql]
    [metabase.driver.sql.query-processor :as sql.qp]
@@ -18,10 +18,12 @@
    [metabase.transforms-base.interface :as transforms-base.i]
    [metabase.transforms-base.schema :as transforms-base.schema]
    [metabase.transforms-base.util :as transforms-base.u]
+   [metabase.transforms.core :as transforms.core]
    [metabase.transforms.instrumentation :as transforms.instrumentation]
    [metabase.util :as u]
    [metabase.util.format :as u.format]
    [metabase.util.i18n :as i18n]
+   [metabase.util.json :as json]
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
    [toucan2.core :as t2])
@@ -238,14 +240,6 @@
       :else
       (transfer-with-drop-create-fallback-strategy! driver db-id table-name metadata indexes data-source))))
 
-;;; ------------------------------------------------- Cancellation -------------------------------------------------
-
-(defn- start-cancellation-process!
-  "Starts a core.async process that sends a cancellation request to the python executor if cancel-chan receives a value."
-  [server-url run-id cancel-chan]
-  (a/go (when (a/<! cancel-chan)
-          (python-runner/cancel-python-code-http-call! server-url run-id))))
-
 ;;; ------------------------------------------------- Core Execution -------------------------------------------------
 
 (defn- count-jsonl-rows
@@ -264,9 +258,10 @@
   [{:keys [source] :as transform} db run-id cancel-chan message-log {:keys [with-stage-timing-fn source-range-params]}]
   ;; Resolve name-based source table refs to table IDs (throws if any not found)
   (let [resolved-source-tables (transforms-base.u/resolve-source-tables (:source-tables source))]
-    (with-open [shared-storage-ref (s3/open-shared-storage! resolved-source-tables)]
+    (with-open [shared-storage-ref (s3/open-shared-storage! resolved-source-tables)
+                _registered        (executor/register-run-closeable!
+                                    run-id {:ingestion? (transforms-base.u/ingestion-transform? transform)})]
       (let [driver          (:engine db)
-            server-url      (transforms-python.settings/python-runner-url)
             _               (python-runner/copy-tables-to-s3! {:run-id              run-id
                                                                :shared-storage      @shared-storage-ref
                                                                :source              (assoc source :source-tables resolved-source-tables)
@@ -274,14 +269,19 @@
                                                                :limit               (:limit source)
                                                                :transform-id        (:id transform)
                                                                :source-range-params source-range-params})
-            _               (start-cancellation-process! server-url run-id cancel-chan)
+            ingestion?      (transforms-base.u/ingestion-transform? transform)
             {:keys [status body] :as response}
-            (python-runner/execute-python-code-http-call!
-             {:server-url     server-url
-              :code           (:body source)
+            (executor/run-python-code!
+             {:code           (:body source)
               :run-id         run-id
+              :ingestion?     ingestion?
               :source-tables  resolved-source-tables
-              :shared-storage @shared-storage-ref})
+              :shared-storage @shared-storage-ref
+              :cancel-chan    cancel-chan
+              ;; secrets are never on the instance; fetched explicitly for the run
+              :secrets        (some-> (:id transform) transforms.core/secrets-for-run)
+              :state          (when ingestion?
+                                (some-> (:incremental_state transform) json/decode))})
 
             output-manifest (python-runner/read-output-manifest @shared-storage-ref)
             events          (python-runner/read-events @shared-storage-ref)]
@@ -313,6 +313,12 @@
                     (with-stage-timing-fn run-id [:import :file-to-dwh] do-transfer)
                     (do-transfer))
                   (transforms.instrumentation/record-data-transfer! run-id :file-to-dwh file-size rows-written)
+                  ;; Persist the sync state only after the data is durably in the target table,
+                  ;; so a failed transfer re-runs from the previous state.
+                  (when-some [new-state (and ingestion? (:state output-manifest))]
+                    (when-let [transform-id (:id transform)]
+                      (t2/update! :model/Transform transform-id
+                                  {:incremental_state (json/encode new-state)})))
                   (assoc response :rows-affected rows-written))
                 (finally
                   (.delete temp-file))))

enterprise/backend/src/metabase_enterprise/transforms_python/execute.clj

@@ -3,7 +3,7 @@
    [clojure.core.async :as a]
    [clojure.string :as str]
    [metabase-enterprise.transforms-python.base :as base]
-   [metabase-enterprise.transforms-python.python-runner :as python-runner]
+   [metabase-enterprise.transforms-python.executor :as executor]
    [metabase.app-db.core :as app-db]
    [metabase.driver :as driver]
    [metabase.driver.connection :as driver.conn]
@@ -45,7 +45,7 @@
         (log/debug "Message update loop interrupted")
         (do (let [sleep-ms (.toMillis python-message-loop-sleep-duration)]
               (when (pos? sleep-ms) (Thread/sleep sleep-ms)))
-            (let [{:keys [status body]} (python-runner/get-logs run-id)]
+            (let [{:keys [status body]} (executor/fetch-logs run-id)]
               (cond
                 (<= 200 status 299)
                 (let [{:keys [execution_id events]} body]

enterprise/backend/src/metabase_enterprise/transforms_python/executor.clj

@@ -0,0 +1,196 @@
+(ns metabase-enterprise.transforms-python.executor
+  "Where a python transform runs: the shared `:http-runner` container, or a per-run `:microvm`.
+
+  Both speak the same HTTP protocol over the same presigned-S3 data plane; they differ in
+  lifecycle and isolation."
+  (:require
+   [clojure.core.async :as a]
+   [clojure.java.io :as io]
+   [clojure.string :as str]
+   [metabase-enterprise.transforms-python.microvm :as microvm]
+   [metabase-enterprise.transforms-python.python-runner :as python-runner]
+   [metabase-enterprise.transforms-python.s3 :as s3]
+   [metabase-enterprise.transforms-python.settings :as transforms-python.settings]
+   [metabase.util :as u]
+   [metabase.util.i18n :as i18n]
+   [metabase.util.json :as json]
+   [metabase.util.log :as log])
+  (:import
+   (java.io Closeable)))
+
+(set! *warn-on-reflection* true)
+
+(declare backend-for)
+
+(defmulti run-python-code!
+  "Execute python `:code` on the backend chosen for this run, blocking until it completes.
+
+  ctx keys: `:code`, `:run-id`/`:request-id`, `:source-tables`, `:shared-storage`,
+  `:timeout-secs`, `:cancel-chan`, `:secrets`, `:state`.
+
+  Returns `{:status <int> :body {:exit_code ..., :timeout ...}}`, 200 meaning success."
+  {:arglists '([ctx])}
+  ;; dispatch through the var so a reload picks up changes (defmulti itself is idempotent)
+  #'backend-for)
+
+(def ^:private known-backends
+  "Backends a setting may name."
+  #{:http-runner :microvm})
+
+(defn- validated-backend [backend setting-name]
+  (if (contains? known-backends backend)
+    backend
+    (do (log/warnf "Unknown %s %s; falling back to :http-runner" setting-name backend)
+        :http-runner)))
+
+(defn ingestion-run?
+  "Whether this run belongs to a transform declared as ingestion."
+  [{:keys [ingestion?]}]
+  (boolean ingestion?))
+
+(defn backend-for
+  "The execution backend for a run; ingestion runs are routed separately."
+  [ctx]
+  (if (ingestion-run? ctx)
+    (validated-backend (transforms-python.settings/python-ingestion-execution-backend)
+                       "python-ingestion-execution-backend")
+    (validated-backend (transforms-python.settings/python-execution-backend)
+                       "python-execution-backend")))
+
+;;; ------------------------------------------------- Live log polling -------------------------------------------------
+
+;; The log poller has only a run id, so each run registers where its logs can be read from.
+(defonce ^:private live-runs (atom {}))
+
+(defn- register-run! [run-id info]
+  (when run-id
+    (swap! live-runs assoc run-id info)))
+
+(defn- unregister-run! [run-id]
+  (when run-id
+    (swap! live-runs dissoc run-id)))
+
+(defn register-run-closeable!
+  "Register a run for log polling, unregistering it on close (for `with-open`)."
+  ^Closeable [run-id ctx]
+  (register-run! run-id {:backend (backend-for ctx)})
+  (reify Closeable
+    (close [_] (unregister-run! run-id))))
+
+(defn- set-run-vm!
+  "Record where a MicroVM run's logs can be read from, once its VM is up. The endpoint alone is
+  not enough: the proxy in front of the VM rejects an unauthenticated request."
+  [run-id endpoint auth-headers]
+  (when run-id
+    (swap! live-runs update run-id assoc :endpoint endpoint :auth-headers auth-headers)))
+
+(defmulti fetch-logs
+  "Events logged so far by an in-flight run, shaped like the runner's `/logs` response:
+  `{:status <int> :body {:execution_id <run-id> :events [{:message ...} ...]}}`. A 404 means
+  \"nothing yet\" and the caller keeps polling."
+  {:arglists '([run-id])}
+  (fn [run-id] (get-in @live-runs [run-id :backend] :http-runner)))
+
+(defmethod fetch-logs :http-runner
+  [run-id]
+  (python-runner/get-logs run-id))
+
+(defmethod fetch-logs :microvm
+  [run-id]
+  (let [{:keys [endpoint auth-headers]} (get @live-runs run-id)]
+    (if endpoint
+      (microvm/fetch-logs endpoint auth-headers)
+      ;; the VM isn't up yet
+      {:status 404 :body {}})))
+
+;;; ------------------------------------------------- Backends -------------------------------------------------
+
+(defmethod run-python-code! :http-runner
+  [{:keys [run-id request-id cancel-chan] :as ctx}]
+  (let [server-url (transforms-python.settings/python-runner-url)
+        req-id     (or request-id run-id)]
+    (when cancel-chan
+      (a/go (when (a/<! cancel-chan)
+              (python-runner/cancel-python-code-http-call! server-url req-id))))
+    (python-runner/execute-python-code-http-call! (assoc ctx :server-url server-url))))
+
+(defmethod run-python-code! :microvm
+  [{:keys [run-id cancel-chan timeout-secs] :as ctx}]
+  (let [timeout-secs (or timeout-secs (transforms-python.settings/python-runner-timeout-seconds))
+        payload      (python-runner/execute-payload ctx)]
+    (microvm/with-microvm
+      {:run-id run-id}
+      (fn [{:keys [endpoint auth-headers terminate!]}]
+        (set-run-vm! run-id endpoint auth-headers)
+        (when cancel-chan
+          (a/go (when (a/<! cancel-chan)
+                  (log/infof "Cancelling run %s by terminating its MicroVM" run-id)
+                  (terminate!))))
+        (microvm/run-job! endpoint payload timeout-secs auth-headers)))))
+
+;;; ------------------------------------------------- Ad-hoc execution -------------------------------------------------
+
+(defn execute-and-read-output!
+  "Execute Python code via the configured backend and return output rows without persisting to a
+   database. Used for dry-run/preview/test-run scenarios.
+
+   Args:
+     :code          - Python code to execute
+     :source-tables - Sequential of source-table entries [{:alias ... :table_id ...} ...]
+     :ingestion?    - Whether the code fetches its own data, which selects the isolated backend
+     :secrets       - Secrets to expose to the code, keyed by name
+     :row-limit     - Max rows to return (also limits input rows)
+     :timeout-secs  - Optional timeout override
+
+   Returns:
+     {:status  :succeeded/:failed
+      :cols    [{:name ...} ...]      ; on success
+      :rows    [[...] ...]            ; on success, values in column order
+      :logs    [{:message ...} ...]   ; events from Python execution
+      :message \"error message\"}     ; on failure
+"
+  [{:keys [code source-tables ingestion? secrets per-input-limit row-limit timeout-secs]}]
+  (with-open [shared-storage-ref (s3/open-shared-storage! source-tables)]
+    (let [_ (python-runner/copy-tables-to-s3! {:shared-storage @shared-storage-ref
+                                               :source         {:source-tables source-tables}
+                                               :limit          (or per-input-limit row-limit)})
+          {:keys [status body]}
+          (run-python-code!
+           {:code           code
+            :request-id     (u/generate-nano-id)
+            :source-tables  source-tables
+            :ingestion?     ingestion?
+            :secrets        secrets
+            :timeout-secs   timeout-secs
+            :shared-storage @shared-storage-ref})
+          events (python-runner/read-events @shared-storage-ref)]
+      (cond
+        (:timeout body)
+        {:status  :failed
+         :logs    events
+         :message (i18n/deferred-tru "Python execution timed out")}
+
+        (not= 200 status)
+        {:status  :failed
+         :logs    events
+         :message (i18n/deferred-tru "Python execution failure (exit code {0})" (:exit_code body "?"))}
+
+        :else
+        (let [output-manifest (python-runner/read-output-manifest @shared-storage-ref)
+              {:keys [fields]} output-manifest]
+          (with-open [in  (python-runner/open-output @shared-storage-ref)
+                      rdr (io/reader in)]
+            (let [cols (mapv (fn [c]
+                               {:name      (:name c)
+                                :base_type (some-> c :base_type keyword)})
+                             fields)
+                  rows (into []
+                             (comp
+                              (remove str/blank?)
+                              (take row-limit)
+                              (map json/decode))
+                             (line-seq rdr))]
+              {:status :succeeded
+               :cols   cols
+               :rows   rows
+               :logs   events})))))))

enterprise/backend/src/metabase_enterprise/transforms_python/microvm.clj

@@ -0,0 +1,251 @@
+(ns metabase-enterprise.transforms-python.microvm
+  "Running one transform in one AWS Lambda MicroVM: launched for the job, terminated after it."
+  (:require
+   [clj-http.client :as http]
+   [clojure.string :as str]
+   [metabase-enterprise.transforms-python.settings :as transforms-python.settings]
+   [metabase.util :as u]
+   [metabase.util.json :as json]
+   [metabase.util.log :as log])
+  (:import
+   (java.net URI)
+   (java.util.function Consumer)
+   (software.amazon.awssdk.auth.credentials AwsSessionCredentials DefaultCredentialsProvider)
+   (software.amazon.awssdk.http ContentStreamProvider SdkHttpMethod SdkHttpRequest SdkHttpRequest$Builder)
+   (software.amazon.awssdk.http.auth.aws.signer AwsV4HttpSigner)
+   (software.amazon.awssdk.http.auth.spi.signer SignRequest$Builder SignedRequest)
+   (software.amazon.awssdk.identity.spi AwsCredentialsIdentity AwsSessionCredentialsIdentity)))
+
+(set! *warn-on-reflection* true)
+
+(def ^:private status-poll-interval-ms 2000)
+
+;;; ------------------------------------------------- The VM's HTTP protocol -------------------------------------------------
+
+(defn- vm-url
+  "RunMicrovm returns a bare hostname; the local stand-in is configured as a full URL."
+  [endpoint path]
+  (str (if (str/includes? endpoint "://") endpoint (str "https://" endpoint)) "/v1" path))
+
+(defn- vm-request
+  "Call the VM's HTTP surface. `:auth-headers` is the map from `CreateMicrovmAuthToken`, checked
+  by AWS before traffic reaches the VM."
+  [endpoint method path {:keys [auth-headers] :as opts}]
+  (http/request (merge {:method           method
+                        :url              (vm-url endpoint path)
+                        :content-type     :json
+                        :accept           :json
+                        :as               :json
+                        :throw-exceptions false}
+                       (dissoc opts :auth-headers)
+                       (when (seq auth-headers)
+                         {:headers auth-headers}))))
+
+(defn fetch-logs
+  "Events logged so far by the job running in the VM at `endpoint`."
+  ([endpoint] (fetch-logs endpoint nil))
+  ([endpoint auth-headers]
+   (vm-request endpoint :get "/logs" {:auth-headers auth-headers})))
+
+(defn run-job!
+  "Submit `payload` to the VM at `endpoint` and block until the job finishes, returning the
+  runner-shaped `{:status <int> :body {:exit_code ..., :timeout ...}}`.
+
+  `/execute` returns 202 and the result is polled from `/status`; the endpoint proxy has no
+  documented request timeout to hold a connection open against."
+  ([endpoint payload timeout-secs] (run-job! endpoint payload timeout-secs nil))
+  ([endpoint payload timeout-secs auth-headers]
+   (let [{:keys [status body]} (vm-request endpoint :post "/execute"
+                                           {:body (json/encode payload) :auth-headers auth-headers})]
+     (if-not (= 202 status)
+       {:status (if (= 200 status) 500 status)
+        :body   (if (map? body) body {:error (str body)})}
+       ;; grace beyond the job timeout: the VM still has to upload its events log
+       (let [deadline-ms (* 1000 (+ timeout-secs 60))
+             result      (u/poll {:thunk       (fn []
+                                                 (let [{:keys [body]} (vm-request endpoint :get "/status"
+                                                                                  {:auth-headers auth-headers})]
+                                                   (when (= "finished" (:state body))
+                                                     (:result body))))
+                                  :done?       some?
+                                  :interval-ms status-poll-interval-ms
+                                  :timeout-ms  deadline-ms})]
+         (cond
+           (nil? result)
+           {:status 504
+            :body   {:timeout true
+                     :error   (str "Timed out waiting for the MicroVM to finish after "
+                                   timeout-secs "s (+60s grace)")}}
+
+           (zero? (:exit_code result))
+           {:status 200 :body result}
+
+           :else
+           {:status 500 :body result}))))))
+
+;;; ------------------------------------------------- Control plane -------------------------------------------------
+
+(defmulti ^:private launch!
+  "Start a MicroVM for this run. Returns `{:endpoint <base-url> :auth-headers <map-or-nil>
+  :id <opaque>}`."
+  {:arglists '([opts])}
+  (fn [_opts] (transforms-python.settings/python-microvm-control-plane)))
+
+(defmulti ^:private terminate!
+  "Destroy the MicroVM. Must be safe to call twice (cancellation races normal completion)."
+  {:arglists '([vm])}
+  (fn [_vm] (transforms-python.settings/python-microvm-control-plane)))
+
+;;; --- local: a stand-in container, for development without AWS
+
+(defmethod launch! :local
+  [_opts]
+  (let [endpoint (transforms-python.settings/python-microvm-endpoint)]
+    (when-not endpoint
+      (throw (ex-info "python-microvm-endpoint must be set to use the local MicroVM control plane"
+                      {:error-type :configuration-error})))
+    (log/infof "Using local MicroVM stand-in at %s" endpoint)
+    {:endpoint endpoint, :id :local}))
+
+(defmethod terminate! :local
+  [_vm]
+  nil)
+
+;;; --- aws
+
+(defn- signing-identity
+  "Credentials for signing control-plane calls: the configured static keys, else the AWS default
+  chain. Temporary credentials must keep their session token or the signature is rejected."
+  ^AwsCredentialsIdentity []
+  (let [access-key (transforms-python.settings/python-microvm-access-key)
+        secret-key (transforms-python.settings/python-microvm-secret-key)]
+    (if (and access-key secret-key)
+      (AwsCredentialsIdentity/create access-key secret-key)
+      (let [creds (.resolveCredentials (DefaultCredentialsProvider/create))]
+        (if (instance? AwsSessionCredentials creds)
+          (AwsSessionCredentialsIdentity/create (.accessKeyId creds)
+                                                (.secretAccessKey creds)
+                                                (.sessionToken ^AwsSessionCredentials creds))
+          (AwsCredentialsIdentity/create (.accessKeyId creds) (.secretAccessKey creds)))))))
+
+(defn- sigv4-headers
+  [{:keys [method url body region]}]
+  (let [identity (signing-identity)
+        ^SdkHttpRequest$Builder builder (-> (SdkHttpRequest/builder)
+                                            (.method (SdkHttpMethod/fromValue (u/upper-case-en (name method))))
+                                            (.uri (URI/create url)))
+        _        (.putHeader builder "Content-Type" "application/json")
+        request  (.build builder)
+        payload  (some-> ^String body ContentStreamProvider/fromUtf8String)
+        ^SignedRequest signed
+        (.sign (AwsV4HttpSigner/create)
+               (reify Consumer
+                 (accept [_ b]
+                   (let [^SignRequest$Builder b b]
+                     (.identity b identity)
+                     (.request b request)
+                     (when payload (.payload b payload))
+                     (.putProperty b AwsV4HttpSigner/SERVICE_SIGNING_NAME "lambda")
+                     (.putProperty b AwsV4HttpSigner/REGION_NAME region)))))]
+    ;; keep every value of a multi-valued header: the signature was computed over all of them
+    (into {} (map (fn [[k vs]] [k (if (= 1 (count vs)) (first vs) (vec vs))])) (.headers (.request signed)))))
+
+(defn- control-plane-call!
+  [method path body]
+  (let [region   (or (transforms-python.settings/python-microvm-region)
+                     (throw (ex-info "python-microvm-region must be set to use the MicroVM backend"
+                                     {:error-type :configuration-error})))
+        base-url (or (transforms-python.settings/python-microvm-control-plane-endpoint)
+                     (str "https://lambda." region ".amazonaws.com"))
+        url      (str base-url path)
+        body-str (some-> body json/encode)
+        headers  (sigv4-headers {:method method :url url :body body-str :region region})
+        {:keys [status body]} (http/request (cond-> {:method           method
+                                                     :url              url
+                                                     :headers          headers
+                                                     :as               :json
+                                                     :throw-exceptions false}
+                                              body-str (assoc :body body-str)))]
+    (if (<= 200 status 299)
+      body
+      (throw (ex-info "MicroVM control plane call failed"
+                      {:status status, :path path, :body (str body)})))))
+
+(defn- managed-connector
+  "ARN of an AWS-managed network connector, e.g. ALL_INGRESS or INTERNET_EGRESS."
+  [region name]
+  (str "arn:aws:lambda:" region ":aws:network-connector:aws-network-connector:" name))
+
+(def ^:private launch-timeout-ms 120000)
+
+(defn- await-running!
+  "RunMicrovm returns while the VM is still PENDING, and its endpoint answers 502 until the
+  application inside is serving. Block until it is."
+  [vm-id]
+  (let [deadline (+ (System/currentTimeMillis) launch-timeout-ms)]
+    (loop []
+      (let [{:keys [state stateReason]} (control-plane-call! :get (str "/2025-09-09/microvms/" vm-id) nil)]
+        (cond
+          (= state "RUNNING")
+          state
+
+          (contains? #{"SUSPENDING" "SUSPENDED" "TERMINATING" "TERMINATED"} state)
+          (throw (ex-info "MicroVM stopped before it could run the job"
+                          {:microvm-id vm-id :state state :reason stateReason}))
+
+          (> (System/currentTimeMillis) deadline)
+          (throw (ex-info "Timed out waiting for the MicroVM to start"
+                          {:microvm-id vm-id :state state :reason stateReason}))
+
+          :else
+          (do (Thread/sleep 1000) (recur)))))))
+
+(defmethod launch! :aws
+  [{:keys [run-id]}]
+  (let [image (transforms-python.settings/python-microvm-image)
+        _     (when-not image
+                (throw (ex-info "python-microvm-image must be set to use the lambda MicroVM backend"
+                                {:error-type :configuration-error})))
+        region (transforms-python.settings/python-microvm-region)
+        ;; No execution role: the VM reaches everything by presigned URL, and user code inside it
+        ;; can read whatever role it is given.
+        max-secs (transforms-python.settings/python-runner-timeout-seconds)
+        vm    (control-plane-call! :post "/2025-09-09/microvms"
+                                   {:imageIdentifier          image
+                                    :maximumDurationInSeconds max-secs
+                                    :ingressNetworkConnectors [(or (transforms-python.settings/python-microvm-ingress-connector)
+                                                                   (managed-connector region "ALL_INGRESS"))]
+                                    :egressNetworkConnectors  [(or (transforms-python.settings/python-microvm-egress-connector)
+                                                                   (managed-connector region "INTERNET_EGRESS"))]})
+        vm-id (:microvmId vm)
+        _     (await-running! vm-id)
+        ;; a map of header name -> value
+        ;; the token has to outlive the VM: it authenticates the status and log polling too
+        token (:authToken (control-plane-call! :post (str "/2025-09-09/microvms/" vm-id "/auth-token")
+                                               {:expirationInMinutes (+ 5 (quot max-secs 60))
+                                                :allowedPorts        [{:port 8080}]}))]
+    (log/infof "Launched MicroVM %s for run %s" vm-id run-id)
+    {:endpoint (:endpoint vm), :auth-headers token, :id vm-id}))
+
+(defmethod terminate! :aws
+  [{:keys [id]}]
+  (when (and id (not= :local id))
+    (try
+      (control-plane-call! :delete (str "/2025-09-09/microvms/" id) nil)
+      (log/infof "Terminated MicroVM %s" id)
+      (catch Exception e
+        ;; an undestroyed VM bills until its maximum duration
+        (log/errorf e "Failed to terminate MicroVM %s" id)))))
+
+(defn with-microvm
+  "Launch a MicroVM, call `f` with `{:endpoint :auth-headers :terminate!}`, and terminate it no
+  matter how `f` ends. `terminate!` is idempotent, so the caller may also end the VM early."
+  [opts f]
+  (let [vm          (launch! opts)
+        terminated? (atom false)
+        terminate!* (fn [] (when (compare-and-set! terminated? false true)
+                             (terminate! vm)))]
+    (try
+      (f (assoc vm :terminate! terminate!*))
+      (finally
+        (terminate!*)))))

enterprise/backend/src/metabase_enterprise/transforms_python/python_runner.clj

@@ -3,7 +3,6 @@
    [clj-http.client :as http]
    [clojure.core.async :as a]
    [clojure.java.io :as io]
-   [clojure.string :as str]
    [medley.core :as m]
    [metabase-enterprise.transforms-python.s3 :as s3]
    [metabase-enterprise.transforms-python.settings :as transforms-python.settings]
@@ -16,7 +15,6 @@
    [metabase.query-processor.pipeline :as qp.pipeline]
    [metabase.transforms-base.util :as transforms-base.u]
    [metabase.transforms.instrumentation :as transforms.instrumentation]
-   [metabase.util :as u]
    [metabase.util.i18n :as i18n]
    [metabase.util.json :as json]
    [metabase.util.log :as log]
@@ -197,10 +195,10 @@
      [~job-run-id]
      (^:once fn* [] ~@body)))
 
-(defn execute-python-code-http-call!
-  "Calls the /execute endpoint of the python runner. Blocks until the run either succeeds or fails and returns
-  the response from the server."
-  [{:keys [server-url code request-id run-id source-tables shared-storage timeout-secs]}]
+(defn execute-payload
+  "The `/execute` request body, shared by every execution backend: user code, library files,
+  timeout, the presigned URLs for inputs and outputs, plus secrets and sync state."
+  [{:keys [code request-id run-id source-tables shared-storage timeout-secs secrets state]}]
   (let [{:keys [objects]} shared-storage
         {:keys [output output-manifest events]} objects
         url-for-path             (fn [path] (:url (get objects path)))
@@ -209,18 +207,27 @@
                                        source-tables)
         table-name->manifest-url (into {} (map (fn [{:keys [alias table_id]}]
                                                  [alias (url-for-path [:table table_id :manifest])]))
-                                       source-tables)
-        payload                  {:code                code
-                                  :library             (t2/select-fn->fn :path :source :model/PythonLibrary)
-                                  :timeout             (or timeout-secs (transforms-python.settings/python-runner-timeout-seconds))
-                                  :request_id          (or request-id run-id)
-                                  :output_url          (:url output)
-                                  :output_manifest_url (:url output-manifest)
-                                  :events_url          (:url events)
-                                  :table_mapping       table-name->url
-                                  :manifest_mapping    table-name->manifest-url}
-        response                 (with-python-api-timing [run-id]
-                                   (python-runner-request server-url :post "/execute" {:body (json/encode payload)}))]
+                                       source-tables)]
+    (cond-> {:code                code
+             :library             (t2/select-fn->fn :path :source :model/PythonLibrary)
+             :timeout             (or timeout-secs (transforms-python.settings/python-runner-timeout-seconds))
+             :request_id          (or request-id run-id)
+             :output_url          (:url output)
+             :output_manifest_url (:url output-manifest)
+             :events_url          (:url events)
+             :table_mapping       table-name->url
+             :manifest_mapping    table-name->manifest-url}
+      ;; decrypted from the transform row; keys are keywordized by the model transform
+      (seq secrets) (assoc :secrets (update-keys secrets name))
+      (some? state) (assoc :state state))))
+
+(defn execute-python-code-http-call!
+  "Calls the /execute endpoint of the python runner. Blocks until the run either succeeds or fails and returns
+  the response from the server."
+  [{:keys [server-url run-id] :as ctx}]
+  (let [payload  (execute-payload ctx)
+        response (with-python-api-timing [run-id]
+                   (python-runner-request server-url :post "/execute" {:body (json/encode payload)}))]
     ;; when a 500 is returned we observe a string in the body (despite the python returning json)
     ;; always try to parse the returned string as json before yielding (could tighten this up at some point)
     (update response :body (fn [string-if-error]
@@ -334,72 +341,3 @@
         (finally
           (safe-delete tmp-data-file)
           (safe-delete tmp-meta-file))))))
-
-(defn execute-and-read-output!
-  "Execute Python code and return output rows without persisting to a database.
-   Used for dry-run/preview/test-run scenarios.
-
-   Args:
-     :code          - Python code to execute
-     :source-tables - Sequential of source-table entries [{:alias ... :table_id ...} ...]
-     :row-limit     - Max rows to return (also limits input rows)
-     :timeout-secs  - Optional timeout override
-
-   Returns:
-     {:status  :succeeded/:failed
-      :cols    [{:name ...} ...]      ; on success
-      :rows    [[...] ...]            ; on success, values in column order
-      :logs    [{:message ...} ...]   ; events from Python execution
-      :message \"error message\"}     ; on failure
-"
-  [{:keys [code source-tables per-input-limit row-limit timeout-secs]}]
-  (with-open [shared-storage-ref (s3/open-shared-storage! source-tables)]
-    (let [server-url (transforms-python.settings/python-runner-url)
-          _          (copy-tables-to-s3! {:shared-storage @shared-storage-ref
-                                          :source         {:source-tables source-tables}
-                                          :limit          (or per-input-limit row-limit)})
-          {:keys [status body]}
-          (execute-python-code-http-call!
-           {:server-url     server-url
-            :code           code
-            :request-id     (u/generate-nano-id)
-            :source-tables  source-tables
-            :timeout-secs   timeout-secs
-            :shared-storage @shared-storage-ref})
-          events (read-events @shared-storage-ref)]
-      (cond
-        (:timeout body)
-        {:status  :failed
-         :logs    events
-         :message (i18n/deferred-tru "Python execution timed out")}
-
-        (not= 200 status)
-        {:status  :failed
-         :logs    events
-         :message (i18n/deferred-tru "Python execution failure (exit code {0})" (:exit_code body "?"))}
-
-        :else
-        (let [output-manifest (read-output-manifest @shared-storage-ref)
-              {:keys [fields]} output-manifest]
-          ;; TODO (Chris 2026-01-27) -- Disabled this check to match behavior in master, but *real* execution does it.
-          ;;      It seems we added the check as part of DRY-ing up transforms code to reuse with workspaces.
-          #_(if-not (seq fields)
-              {:status  :failed
-               :logs    events
-               :message (i18n/deferred-tru "No fields in output metadata")})
-          (with-open [in  (open-output @shared-storage-ref)
-                      rdr (io/reader in)]
-            (let [cols (mapv (fn [c]
-                               {:name      (:name c)
-                                :base_type (some-> c :base_type keyword)})
-                             fields)
-                  rows (into []
-                             (comp
-                              (remove str/blank?)
-                              (take row-limit)
-                              (map json/decode))
-                             (line-seq rdr))]
-              {:status :succeeded
-               :cols   cols
-               :rows   rows
-               :logs   events})))))))

enterprise/backend/src/metabase_enterprise/transforms_python/settings.clj

@@ -128,6 +128,122 @@
   :encryption :no
   :audit      :getter)
 
+(setting/defsetting python-execution-backend
+  (deferred-tru "Backend used to execute python transforms that read source tables: http-runner (the python-runner container service) or microvm.")
+  :type       :keyword
+  :visibility :admin
+  :default    :http-runner
+  :feature    :transforms-python
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)
+
+(setting/defsetting python-ingestion-execution-backend
+  (deferred-tru "Backend used to execute ingestion python transforms: those with no source tables, which fetch their own data over the network.")
+  :type       :keyword
+  :visibility :admin
+  :default    :http-runner
+  :feature    :transforms-python
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)
+
+(setting/defsetting python-microvm-control-plane
+  (deferred-tru "How MicroVMs are created: aws, or local to target a stand-in container during development.")
+  :type       :keyword
+  :visibility :admin
+  :default    (if config/is-prod? :aws :local)
+  :feature    :transforms-python
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)
+
+(setting/defsetting python-microvm-endpoint
+  (deferred-tru "Base URL of the MicroVM stand-in used by the local control plane.")
+  :type       :string
+  :visibility :admin
+  :default    (when (not config/is-prod?) "http://localhost:8085")
+  :feature    :transforms-python
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)
+
+(setting/defsetting python-microvm-image
+  (deferred-tru "ARN of the MicroVM image transforms are executed from.")
+  :type       :string
+  :visibility :admin
+  :feature    :transforms-python
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)
+
+(setting/defsetting python-microvm-control-plane-endpoint
+  (deferred-tru "Endpoint override for the MicroVM control plane API. Leave empty for the regional AWS endpoint.")
+  :type       :string
+  :visibility :admin
+  :feature    :transforms-python
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)
+
+(setting/defsetting python-microvm-region
+  (deferred-tru "AWS region MicroVMs run in. Independent of where object storage lives: MicroVMs are only offered in a few regions, so the bucket may well be elsewhere.")
+  :type       :string
+  :visibility :admin
+  :feature    :transforms-python
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)
+
+(setting/defsetting python-microvm-access-key
+  (deferred-tru "AWS access key ID for the MicroVM control plane. Leave empty to use ambient credentials (instance profile, IRSA, environment), which is the recommended setup.")
+  :type       :string
+  :visibility :admin
+  :sensitive? true
+  :feature    :transforms-python
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)
+
+(setting/defsetting python-microvm-secret-key
+  (deferred-tru "AWS secret access key for the MicroVM control plane. Leave empty to use ambient credentials.")
+  :type       :string
+  :visibility :admin
+  :sensitive? true
+  :feature    :transforms-python
+  :doc        false
+  :export?    false
+  :encryption :when-encryption-key-set
+  :audit      :never)
+
+(setting/defsetting python-microvm-ingress-connector
+  (deferred-tru "ARN of the network connector governing inbound traffic to a MicroVM. Leave empty for the AWS-managed ALL_INGRESS connector, which is what lets Metabase reach the VM.")
+  :type       :string
+  :visibility :admin
+  :feature    :transforms-python
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)
+
+(setting/defsetting python-microvm-egress-connector
+  (deferred-tru "ARN of the network connector governing MicroVM outbound traffic. Leave empty for the AWS-managed INTERNET_EGRESS connector; point it at a VPC connector to confine ingestion traffic.")
+  :type       :string
+  :visibility :admin
+  :feature    :transforms-python
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)
+
 (setting/defsetting python-runner-test-run-timeout-seconds
   (deferred-tru "Timeout in seconds for Python script test runs. Defaults to 1 minute (60 seconds).")
   :type :integer

enterprise/backend/test/metabase_enterprise/transforms/incremental_test.clj

@@ -166,12 +166,12 @@
 
 (defn- get-checkpoint-value [transform-id]
   (let [transform (t2/select-one :model/Transform transform-id)
-        {:keys [last_checkpoint_value]} transform]
-    (when last_checkpoint_value
+        {:keys [incremental_state]} transform]
+    (when incremental_state
       (let [source-incremental-strategy (get-in transform [:source :source-incremental-strategy])
             field-id (:checkpoint-filter-field-id source-incremental-strategy)
             field (t2/select-one :model/Field field-id)]
-        (#'transforms-base.u/parse-checkpoint-value (:base_type field) last_checkpoint_value)))))
+        (#'transforms-base.u/parse-checkpoint-value (:base_type field) incremental_state)))))
 
 (defn- compare-checkpoint-values
   "Compare two checkpoint values with type-appropriate logic. "
@@ -540,7 +540,7 @@
                 (is (=? {:name target-table} (t2/select-one :model/Table :name target-table))))
               (testing "checkpoint is set from source table MAX even though query returned no rows"
                 (let [transform (t2/select-one :model/Transform (:id transform))]
-                  (is (some? (:last_checkpoint_value transform))
+                  (is (some? (:incremental_state transform))
                       "Checkpoint is computed from source table, not query output"))))))))))
 
 (deftest checkpoint-field-does-not-exist-test

enterprise/backend/test/metabase_enterprise/transforms/merge_test.clj

@@ -305,7 +305,7 @@
                                                              (sql.u/quote-name driver/*driver* :table schema tbl)
                                                              (q "order_id") order-id)]
                                     (driver/execute-raw-queries! driver/*driver* spec [[sql]])))
-                  watermark    (fn [id] (t2/select-one-fn :last_checkpoint_value :model/Transform id))]
+                  watermark    (fn [id] (t2/select-one-fn :incremental_state :model/Transform id))]
               (mt/with-temp [:model/Transform transform payload]
                 (testing "first run builds the full target and records the watermark"
                   (transforms.execute/execute! transform {:run-method :manual})
@@ -407,7 +407,7 @@
                   (transforms.tu/wait-for-table (:name target-table) 10000)
                   (is (= {:by-order {1 "created", 2 "created", 3 "created"} :count 3}
                          (read-target)))
-                  (let [wm (t2/select-one-fn :last_checkpoint_value :model/Transform (:id transform))]
+                  (let [wm (t2/select-one-fn :incremental_state :model/Transform (:id transform))]
                     (is (re-find #"T" (str wm))
                         "the watermark encodes as a temporal value, not the raw epoch number")))
                 (testing "a late-arriving epoch inside the lookback window is upserted on rerun"

enterprise/backend/test/metabase_enterprise/transforms_python/base_test.clj

@@ -3,6 +3,7 @@
    [clojure.core.async :as a]
    [clojure.test :refer :all]
    [metabase-enterprise.transforms-python.base :as base]
+   [metabase-enterprise.transforms-python.executor :as executor]
    [metabase-enterprise.transforms-python.python-runner :as python-runner]
    [metabase-enterprise.transforms-python.s3 :as s3]
    [metabase.transforms-base.util :as transforms-base.u]
@@ -59,13 +60,12 @@
         {#'transforms-base.u/resolve-source-tables             (constantly [])
          #'s3/open-shared-storage!                             (constantly stub-storage)
          #'python-runner/copy-tables-to-s3!                    (constantly nil)
-         #'python-runner/execute-python-code-http-call!        (constantly {:status 200 :body {:exit_code 0}})
+         #'executor/run-python-code!                           (constantly {:status 200 :body {:exit_code 0}})
          #'python-runner/read-output-manifest                  (constantly {:fields [{:name "a" :base_type "Integer"}]})
          #'python-runner/read-events                           (constantly [])
          #'python-runner/open-output                           (fn [_] (fake-output-stream))
          #'transforms.instrumentation/record-data-transfer!    (fn [_run-id _stage _bytes rows]
                                                                  (reset! recorded-rows rows))
-         #'base/start-cancellation-process!                    (constantly nil)
          #'base/transfer-file-to-db                            (constantly nil)}
         (fn []
           (let [response (#'base/run-python-transform-impl!

enterprise/frontend/src/metabase-enterprise/transforms-python/components/PythonTransformEditor/PythonDataPicker/PythonDataPicker.tsx

@@ -3,7 +3,16 @@ import { t } from "ttag";
 
 import { skipToken, useListTablesQuery } from "metabase/api";
 import { LoadingAndErrorWrapper } from "metabase/common/components/LoadingAndErrorWrapper";
-import { Box, Button, Icon, Stack, Text } from "metabase/ui";
+import {
+  Box,
+  Button,
+  Group,
+  Icon,
+  Stack,
+  Switch,
+  Text,
+  Tooltip,
+} from "metabase/ui";
 import type {
   DatabaseId,
   PythonTransformTableAliases,
@@ -26,11 +35,13 @@ type PythonDataPickerProps = {
   disabled?: boolean;
   tables: PythonTransformTableAliases;
   readOnly?: boolean;
+  isIngestion: boolean;
   onChange: (
     database: number,
     tables: PythonTransformTableAliases,
     tableInfo: Table[],
   ) => void;
+  onIngestionChange: (isIngestion: boolean) => void;
 };
 
 export function PythonDataPicker({
@@ -38,7 +49,9 @@ export function PythonDataPicker({
   disabled,
   tables,
   readOnly,
+  isIngestion,
   onChange,
+  onIngestionChange,
 }: PythonDataPickerProps) {
   const [tableSelections, setTableSelections] = useState<TableSelection[]>(
     getInitialTableSelections(tables),
@@ -109,10 +122,6 @@ export function PythonDataPicker({
     handleChange(newSelections);
   };
 
-  if (!database) {
-    return null;
-  }
-
   return (
     <Stack
       p="md"
@@ -123,7 +132,24 @@ export function PythonDataPicker({
       className={S.dataPicker}
       data-testid="python-data-picker"
     >
-      {database && (
+      <Group gap="xs" align="center" wrap="nowrap">
+        <Switch
+          checked={isIngestion}
+          disabled={disabled || readOnly}
+          size="sm"
+          label={t`Fetches its own data`}
+          wrapperProps={{ "data-testid": "python-ingestion-switch" }}
+          onChange={(event) => onIngestionChange(event.currentTarget.checked)}
+        />
+        <Tooltip
+          multiline
+          w={260}
+          label={t`Gets its data over the network instead of reading source tables. It can store credentials and keeps its own incremental state.`}
+        >
+          <Icon name="info" c="text-secondary" size={14} />
+        </Tooltip>
+      </Group>
+      {database && !isIngestion && (
         <Box>
           <Text fw="bold">{t`Pick tables and alias them`}</Text>
           <Text size="sm" c="text-disabled" mb="sm">

enterprise/frontend/src/metabase-enterprise/transforms-python/components/PythonTransformEditor/PythonTransformEditor.tsx

@@ -3,6 +3,10 @@ import { useEffect } from "react";
 import { usePrevious } from "react-use";
 
 import type { PythonTransformEditorProps } from "metabase/plugins";
+import {
+  INGESTION_PYTHON_BODY,
+  STARTER_PYTHON_BODY,
+} from "metabase/transforms/constants";
 import { Flex, Stack } from "metabase/ui";
 import type {
   DatabaseId,
@@ -18,7 +22,7 @@ import { PythonEditorResults } from "./PythonEditorResults";
 import S from "./PythonTransformEditor.module.css";
 import { PythonTransformTopBar } from "./PythonTransformTopBar";
 import { useTestPythonTransform } from "./hooks";
-import { updateTransformSignature } from "./utils";
+import { setIngestionSignature, updateTransformSignature } from "./utils";
 
 export function PythonTransformEditor({
   source,
@@ -34,7 +38,7 @@ export function PythonTransformEditor({
   onRun,
 }: PythonTransformEditorProps) {
   const { isRunning, cancel, run, executionResult, isDirty } =
-    useTestPythonTransform(source);
+    useTestPythonTransform(source, transform?.id);
 
   useEffect(() => {
     const errMsg = [executionResult?.error?.message, executionResult?.logs]
@@ -84,6 +88,35 @@ export function PythonTransformEditor({
     onChangeSource(newSource);
   };
 
+  // The backend rejects an ingestion transform that also has source tables, so turning the
+  // toggle on drops the selections (and the parameters they added to the script signature).
+  const handleIngestionChange = (isIngestion: boolean) => {
+    // An untouched template is swapped for the other one; a script the user has written keeps its
+    // body and only has its signature adapted.
+    const isUntouched = (template: string) =>
+      source.body.trim() === template.trim();
+
+    if (!isIngestion) {
+      onChangeSource({
+        ...source,
+        ingestion: false,
+        body: isUntouched(INGESTION_PYTHON_BODY)
+          ? STARTER_PYTHON_BODY
+          : updateTransformSignature(source.body, [], []),
+      });
+      return;
+    }
+
+    onChangeSource({
+      ...source,
+      ingestion: true,
+      "source-tables": [],
+      body: isUntouched(STARTER_PYTHON_BODY)
+        ? INGESTION_PYTHON_BODY
+        : setIngestionSignature(source.body),
+    });
+  };
+
   const handleRun = () => {
     if (onRun) {
       onRun();
@@ -139,7 +172,9 @@ export function PythonTransformEditor({
             disabled={uiOptions?.readOnly}
             database={source["source-database"]}
             tables={source["source-tables"]}
+            isIngestion={source.ingestion ?? false}
             onChange={handleDataChange}
+            onIngestionChange={handleIngestionChange}
           />
         )}
         <Stack w="100%" h="100%" gap={0}>

enterprise/frontend/src/metabase-enterprise/transforms-python/components/PythonTransformEditor/hooks.ts

@@ -6,6 +6,7 @@ import { useExecutePythonMutation } from "metabase-enterprise/api/transform-pyth
 import type {
   PythonTransformSourceDraft,
   TestPythonTransformResponse,
+  TransformId,
 } from "metabase-types/api";
 
 type TestPythonScriptState = {
@@ -18,6 +19,7 @@ type TestPythonScriptState = {
 
 export function useTestPythonTransform(
   source: PythonTransformSourceDraft,
+  transformId?: TransformId,
 ): TestPythonScriptState {
   const [
     executePython,
@@ -35,6 +37,8 @@ export function useTestPythonTransform(
     const request = executePython({
       code: source.body,
       source_tables: source["source-tables"],
+      ingestion: source.ingestion ?? false,
+      transform_id: transformId,
     });
 
     abort.current = () => request.abort();

enterprise/frontend/src/metabase-enterprise/transforms-python/components/PythonTransformEditor/utils.ts

@@ -138,3 +138,12 @@ ${tableAliases
 
   return script + functionTemplate;
 }
+
+const TRANSFORM_SIGNATURE_REGEX = /^def\s+transform\s*\([^)]*\)\s*:/m;
+
+export function setIngestionSignature(script: string): string {
+  return script.replace(
+    TRANSFORM_SIGNATURE_REGEX,
+    "def transform(secrets=None, state=None):",
+  );
+}

enterprise/frontend/src/metabase-enterprise/transforms-python/pages/PythonRunnerSettingsPage/PythonRunnerSettingsPage.tsx

@@ -77,6 +77,71 @@ export function PythonRunnerSettingsPage() {
           inputType="boolean"
         />
       </SettingsSection>
+      <SettingsSection
+        title={t`Execution`}
+        description={t`Ingestion transforms fetch their own data over the network, so they can run in a MicroVM created for that one run instead of the shared runner.`}
+      >
+        <AdminSettingInput
+          name="python-ingestion-execution-backend"
+          title={t`Ingestion transform backend`}
+          description={t`Where transforms that fetch their own data run: "microvm", or "http-runner" to use the shared Python runner.`}
+          placeholder="http-runner"
+          inputType="text"
+        />
+        <AdminSettingInput
+          name="python-execution-backend"
+          title={t`Transform backend`}
+          description={t`Where every other Python transform runs.`}
+          placeholder="http-runner"
+          inputType="text"
+        />
+      </SettingsSection>
+      <SettingsSection title={t`MicroVM Configuration`}>
+        <AdminSettingInput
+          name="python-microvm-control-plane"
+          title={t`Control plane`}
+          description={t`"aws" to create real MicroVMs, or "local" to target a stand-in container during development.`}
+          placeholder="aws"
+          inputType="text"
+        />
+        <AdminSettingInput
+          name="python-microvm-image"
+          title={t`Image ARN`}
+          description={t`ARN of the MicroVM image transforms are executed from.`}
+          inputType="text"
+        />
+        <AdminSettingInput
+          name="python-microvm-region"
+          title={t`Region`}
+          description={t`AWS region MicroVMs run in. Independent of where object storage lives.`}
+          inputType="text"
+        />
+        <AdminSettingInput
+          name="python-microvm-access-key"
+          title={t`Access Key ID`}
+          description={t`Leave empty to use ambient credentials, which is the recommended setup.`}
+          inputType="password"
+        />
+        <AdminSettingInput
+          name="python-microvm-secret-key"
+          title={t`Secret Access Key`}
+          description={t`Leave empty to use ambient credentials.`}
+          inputType="password"
+        />
+        <AdminSettingInput
+          name="python-microvm-egress-connector"
+          title={t`Egress connector ARN (Optional)`}
+          description={t`Governs outbound traffic from a MicroVM. Leave empty to allow unrestricted internet access.`}
+          inputType="text"
+        />
+        <AdminSettingInput
+          name="python-microvm-endpoint"
+          title={t`Stand-in endpoint (Optional)`}
+          description={t`Base URL of the MicroVM stand-in used by the local control plane.`}
+          placeholder="http://localhost:8085"
+          inputType="text"
+        />
+      </SettingsSection>
     </SettingsPageWrapper>
   );
 }

enterprise/frontend/src/metabase-enterprise/transforms-python/utils.ts

@@ -20,7 +20,9 @@ export function getPythonSourceValidationResult(
     };
   }
 
-  if (source["source-tables"].length === 0) {
+  // An ingestion transform fetches its own data over the network, so it must not have any
+  // source tables; every other python transform needs at least one.
+  if (!source.ingestion && source["source-tables"].length === 0) {
     return {
       isValid: false,
       errorMessage: t`Select at least one table to alias`,

frontend/src/metabase-types/api/settings.ts

@@ -830,6 +830,17 @@ export interface EnterpriseSettings extends Settings {
   "python-storage-s-3-container-endpoint"?: string | null;
   "python-storage-s-3-path-style-access"?: boolean | null;
   "python-runner-timeout-seconds"?: number | null;
+  "python-execution-backend"?: string | null;
+  "python-ingestion-execution-backend"?: string | null;
+  "python-microvm-control-plane"?: string | null;
+  "python-microvm-endpoint"?: string | null;
+  "python-microvm-image"?: string | null;
+  "python-microvm-region"?: string | null;
+  "python-microvm-access-key"?: string | null;
+  "python-microvm-secret-key"?: string | null;
+  "python-microvm-control-plane-endpoint"?: string | null;
+  "python-microvm-ingress-connector"?: string | null;
+  "python-microvm-egress-connector"?: string | null;
   "python-runner-test-run-timeout-seconds"?: number | null;
   "llm-metabot-provider"?: string | null;
   "llm-anthropic-api-key"?: string | null;

frontend/src/metabase-types/api/transform.ts

@@ -56,7 +56,12 @@ export type Transform = {
   owner_email?: string | null;
   owner?: TransformOwner | null;
 
-  last_checkpoint_value?: string | null;
+  // where the last run left off: a checkpoint watermark, or the opaque JSON state an ingestion
+  // transform returned from its own code
+  incremental_state?: string | null;
+
+  // names of configured secret env vars (values never leave the server)
+  secret_keys?: string[];
 
   // set by the job transforms endpoint on transforms pulled into the plan
   // only as dependencies (not tagged for the job); `scheduled` says whether
@@ -120,6 +125,12 @@ export type PythonTransformSourceDraft = {
   body: string;
   "source-database": DatabaseId | undefined;
   "source-tables": PythonTransformTableAliases;
+  /**
+   * Declares that the transform fetches its own data over the network instead of reading source
+   * tables. Such a transform may hold secrets and manages its own incremental sync state. The backend
+   * rejects it when `source-tables` is non-empty.
+   */
+  ingestion?: boolean;
 };
 
 export type PythonTransformSource = {
@@ -127,6 +138,7 @@ export type PythonTransformSource = {
   body: string;
   "source-database": DatabaseId;
   "source-tables": PythonTransformTableAliases;
+  ingestion?: boolean;
   "source-incremental-strategy"?: SourceIncrementalStrategy;
 };
 
@@ -297,6 +309,8 @@ export type CreateTransformRequest = {
   collection_id?: number | null;
   owner_user_id?: UserId | null;
   owner_email?: string | null;
+  // write-only, map of secret name -> value
+  secrets?: Record<string, string>;
 };
 
 export type UpdateTransformRequest = {
@@ -309,6 +323,8 @@ export type UpdateTransformRequest = {
   collection_id?: number | null;
   owner_user_id?: UserId | null;
   owner_email?: string | null;
+  // write-only, per-key merge: string sets/replaces, null removes
+  secrets?: Record<string, string | null>;
 };
 
 export type CreateTransformJobRequest = {
@@ -466,6 +482,9 @@ export type ListTransformGraphRunsResponse = {
 export type TestPythonTransformRequest = {
   code: string;
   source_tables: PythonTransformTableAliases;
+  ingestion?: boolean;
+  // a saved transform lends its stored secrets to the preview run
+  transform_id?: TransformId;
 };
 
 export type TestPythonTransformResponse = {

frontend/src/metabase/transforms/components/IncrementalTransform/IncrementalTransformSettings.tsx

@@ -8,7 +8,10 @@ import { PLUGIN_REMOTE_SYNC } from "metabase/plugins";
 import { useSelector } from "metabase/redux";
 import { getMetadata } from "metabase/selectors/metadata";
 import { SOURCE_STRATEGY_OPTIONS } from "metabase/transforms/constants";
-import { getLibQuery } from "metabase/transforms/utils";
+import {
+  getLibQuery,
+  hasCodeManagedSyncState,
+} from "metabase/transforms/utils";
 import {
   Anchor,
   Box,
@@ -64,6 +67,7 @@ export const IncrementalTransformSettings = ({
 
   const isMultiTablePythonTransform =
     getIsPythonTransformWithMultipleTables(source);
+  const isCodeManagedCursor = hasCodeManagedSyncState(source);
   const isNativeWithoutTableTags = getIsNativeWithoutTableVariables(
     libQuery,
     transformType,
@@ -83,7 +87,7 @@ export const IncrementalTransformSettings = ({
       if (isNativeWithoutTableTags) {
         return t`Incremental transforms for native queries require a table variable.`;
       }
-      if (!hasCheckpointOptions) {
+      if (!hasCheckpointOptions && !isCodeManagedCursor) {
         return t`Incremental transforms require at least one numeric or temporal source field.`;
       }
       return t`Only process new data`;
@@ -91,7 +95,7 @@ export const IncrementalTransformSettings = ({
 
     const transformHasIssues =
       isNativeWithoutTableTags ||
-      !hasCheckpointOptions ||
+      (!hasCheckpointOptions && !isCodeManagedCursor) ||
       isMultiTablePythonTransform;
 
     const switchContent = (
@@ -152,15 +156,19 @@ export const IncrementalTransformSettings = ({
         <Group p="lg">{renderIncrementalSwitch()}</Group>
         {incremental && (
           <>
-            <Divider />
-            <Group p="lg">
-              <SourceStrategyFields
-                source={source}
-                query={libQuery}
-                transformType={transformType}
-                readOnly={readOnly}
-              />
-            </Group>
+            {!isCodeManagedCursor && (
+              <>
+                <Divider />
+                <Group p="lg">
+                  <SourceStrategyFields
+                    source={source}
+                    query={libQuery}
+                    transformType={transformType}
+                    readOnly={readOnly}
+                  />
+                </Group>
+              </>
+            )}
             {extraActions && (
               <>
                 <Divider />
@@ -189,11 +197,13 @@ export const IncrementalTransformSettings = ({
       </Box>
       {incremental && (
         <>
-          <SourceStrategyFields
-            source={source}
-            query={libQuery}
-            transformType={transformType}
-          />
+          {!isCodeManagedCursor && (
+            <SourceStrategyFields
+              source={source}
+              query={libQuery}
+              transformType={transformType}
+            />
+          )}
           <TargetStrategyFields
             variant={variant}
             targetTableId={targetTableId}
@@ -285,18 +295,22 @@ function SourceStrategyFields({
               disabled={readOnly}
             />
           )}
-          {transformType === "python" && "source-tables" in source && (
-            <PythonKeysetColumnSelect
-              name="checkpointFilterFieldId"
-              label={t`Field to check for new values`}
-              placeholder={t`Pick a field`}
-              description={t`Pick the input field we should scan to determine which records are new or changed`}
-              descriptionProps={{ lh: "1rem" }}
-              sourceTables={source["source-tables"]}
-              disabled={readOnly}
-            />
+          {transformType === "python" &&
+            "source-tables" in source &&
+            !hasCodeManagedSyncState(source) && (
+              <PythonKeysetColumnSelect
+                name="checkpointFilterFieldId"
+                label={t`Field to check for new values`}
+                placeholder={t`Pick a field`}
+                description={t`Pick the input field we should scan to determine which records are new or changed`}
+                descriptionProps={{ lh: "1rem" }}
+                sourceTables={source["source-tables"]}
+                disabled={readOnly}
+              />
+            )}
+          {!hasCodeManagedSyncState(source) && (
+            <LookbackField readOnly={readOnly} />
           )}
-          <LookbackField readOnly={readOnly} />
         </>
       )}
     </>

frontend/src/metabase/transforms/components/IncrementalTransform/IncrementalTransformSettings.unit.spec.tsx

@@ -13,6 +13,7 @@ jest.mock("./useHasCheckpointOptions", () => ({
 }));
 
 jest.mock("metabase/transforms/utils", () => ({
+  ...jest.requireActual("metabase/transforms/utils"),
   getLibQuery: () => null,
   isMbqlQuery: () => false,
 }));

frontend/src/metabase/transforms/components/IncrementalTransform/form.ts

@@ -37,12 +37,21 @@ export type IncrementalSettingsFormValues = {
 export const VALIDATION_SCHEMA = Yup.object({
   incremental: Yup.boolean().required(),
   sourceStrategy: Yup.mixed<"checkpoint">().oneOf(["checkpoint"]).required(),
+  // `$hasCodeManagedSyncState` is threaded in via `FormProvider`'s `validationContext`;
+  // python transforms without source tables manage the sync state in code, so no
+  // checkpoint field is required.
   checkpointFilterFieldId: Yup.string()
     .nullable()
     .defined()
-    .when(["incremental", "sourceStrategy"], {
-      is: (incremental: boolean, sourceStrategy: "checkpoint") =>
-        incremental && sourceStrategy === "checkpoint",
+    .when(["incremental", "sourceStrategy", "$hasCodeManagedSyncState"], {
+      is: (
+        incremental: boolean,
+        sourceStrategy: "checkpoint",
+        hasCodeManagedSyncState: boolean | undefined,
+      ) =>
+        incremental &&
+        sourceStrategy === "checkpoint" &&
+        !hasCodeManagedSyncState,
       then: (schema) => schema.required(Errors.required),
       otherwise: (schema) => schema.nullable().defined(),
     }),

frontend/src/metabase/transforms/components/SecretNameInput/SecretNameInput.tsx

@@ -0,0 +1,33 @@
+import { t } from "ttag";
+
+import { TextInput, type TextInputProps } from "metabase/ui";
+
+import { MAX_SECRET_NAME_LENGTH, isValidSecretName } from "../../utils";
+
+type SecretNameInputProps = Omit<TextInputProps, "value" | "onChange"> & {
+  value: string;
+  onChange: (value: string) => void;
+};
+
+export function SecretNameInput({
+  value,
+  onChange,
+  ...props
+}: SecretNameInputProps) {
+  const trimmedName = value.trim();
+  const error =
+    trimmedName.length > 0 && !isValidSecretName(trimmedName)
+      ? t`Use uppercase letters, digits, and underscores, starting with a letter (max ${MAX_SECRET_NAME_LENGTH} characters).`
+      : undefined;
+
+  return (
+    <TextInput
+      label={t`Name`}
+      placeholder="MY_API_TOKEN"
+      {...props}
+      value={value}
+      error={error}
+      onChange={(event) => onChange(event.currentTarget.value)}
+    />
+  );
+}

frontend/src/metabase/transforms/components/SecretNameInput/index.ts

@@ -0,0 +1 @@
+export * from "./SecretNameInput";

frontend/src/metabase/transforms/constants.ts

@@ -29,3 +29,47 @@ export const TARGET_STRATEGY_OPTIONS = [
     },
   },
 ];
+
+export const STARTER_PYTHON_BODY = `# Write your Python transformation script here
+import common
+import pandas as pd
+
+def transform():
+    """
+    Your transformation function.
+
+    Select tables above to add them as function parameters.
+
+    Returns:
+        DataFrame to write to the destination table
+    """
+    # Your transformation logic here
+    return pd.DataFrame([{"message": "Hello from Python transform!"}])`;
+
+export const INGESTION_PYTHON_BODY = `# Fetch data from an external service and load it into the target table
+import common
+import pandas as pd
+import dlt
+
+def transform(secrets=None, state=None):
+    """
+    Your ingestion function.
+
+    Args:
+        secrets: credentials configured on this transform, keyed by name
+        state:   only used to sync incrementally -- whatever you returned last
+                 run, or None on the first run. Ignore it to refetch everything.
+
+    Returns:
+        the DataFrame to write, or (DataFrame, state) to sync incrementally
+    """
+    since = (state or {}).get("since")
+
+    rows = [{"id": 1, "updated_at": "2026-01-01T00:00:00Z"}]
+    df = pd.DataFrame(rows)
+
+    # Return just df to refetch everything on every run. Returning state as well
+    # resumes from where this run left off; pair that with a merge key on the
+    # target so re-fetched rows update in place.
+    next_since = df["updated_at"].max() if len(df) else since
+    return df, {"since": next_since}`;

frontend/src/metabase/transforms/pages/NewTransformPage/CreateTransformModal/CreateTransformModal.tsx

@@ -20,6 +20,10 @@ import {
   FormTextInput,
 } from "metabase/forms";
 import { IncrementalTransformSettings } from "metabase/transforms/components/IncrementalTransform/IncrementalTransformSettings";
+import {
+  hasCodeManagedSyncState,
+  isIngestionSource,
+} from "metabase/transforms/utils";
 import { Box, Button, Group, Modal, Stack } from "metabase/ui";
 import type {
   SchemaName,
@@ -30,7 +34,8 @@ import type {
 import { SchemaFormSelect } from "../../../components/SchemaFormSelect";
 import { TargetNameInput } from "../../../components/TargetNameInput";
 
-import type { NewTransformValues } from "./form";
+import { SecretsFields } from "./SecretsFields";
+import type { NewTransformSecret, NewTransformValues } from "./form";
 import { useCreateTransform } from "./hooks";
 
 export type ValidationSchemaExtension = Record<string, Yup.AnySchema>;
@@ -102,8 +107,11 @@ export function CreateTransformModal({
   );
 
   const validationContext = useMemo(
-    () => ({ supportsSchemas: Boolean(supportsSchemas) }),
-    [supportsSchemas],
+    () => ({
+      supportsSchemas: Boolean(supportsSchemas),
+      hasCodeManagedSyncState: hasCodeManagedSyncState(source),
+    }),
+    [supportsSchemas, source],
   );
 
   if (isLoading || error != null) {
@@ -169,6 +177,10 @@ function CreateTransformForm({
     setFieldValue("incremental", value);
   };
 
+  const handleSecretsChange = (secrets: NewTransformSecret[]) => {
+    setFieldValue("secrets", secrets);
+  };
+
   return (
     <Form>
       <Stack gap="lg" mt="sm">
@@ -199,6 +211,12 @@ function CreateTransformForm({
             onIncrementalChange={handleIncrementalChange}
           />
         )}
+        {isIngestionSource(source) && (
+          <SecretsFields
+            secrets={values.secrets}
+            onChange={handleSecretsChange}
+          />
+        )}
         <Group>
           <Box flex={1}>
             <FormErrorMessage />

frontend/src/metabase/transforms/pages/NewTransformPage/CreateTransformModal/SecretsFields.tsx

@@ -0,0 +1,95 @@
+import { t } from "ttag";
+
+import { SecretNameInput } from "metabase/transforms/components/SecretNameInput";
+import {
+  ActionIcon,
+  Box,
+  Button,
+  Group,
+  Icon,
+  PasswordInput,
+  Stack,
+  Text,
+  Tooltip,
+} from "metabase/ui";
+
+import type { NewTransformSecret } from "./form";
+
+const SECRET_NAME_WIDTH = "12rem";
+
+type SecretsFieldsProps = {
+  secrets: NewTransformSecret[];
+  onChange: (secrets: NewTransformSecret[]) => void;
+};
+
+export function SecretsFields({ secrets, onChange }: SecretsFieldsProps) {
+  const handleAdd = () => {
+    onChange([...secrets, { name: "", value: "" }]);
+  };
+
+  const handleRemove = (index: number) => {
+    onChange(secrets.filter((_, secretIndex) => secretIndex !== index));
+  };
+
+  const handleChange = (index: number, secret: NewTransformSecret) => {
+    onChange(
+      secrets.map((prevSecret, secretIndex) =>
+        secretIndex === index ? secret : prevSecret,
+      ),
+    );
+  };
+
+  return (
+    <Stack gap="sm" data-testid="transform-secrets-fields">
+      <Group gap="xs">
+        <Text fw="bold">{t`Secrets`}</Text>
+        <Tooltip
+          label={t`Secrets are passed to your Python code in the secrets argument, keyed by name. Values can't be viewed after saving.`}
+        >
+          <Icon name="info" c="text-secondary" size={14} />
+        </Tooltip>
+      </Group>
+      {secrets.map((secret, index) => (
+        <Group key={index} gap="sm" wrap="nowrap" align="flex-start">
+          <SecretNameInput
+            label={undefined}
+            aria-label={t`Secret name`}
+            w={SECRET_NAME_WIDTH}
+            value={secret.name}
+            onChange={(name) => handleChange(index, { ...secret, name })}
+          />
+          <PasswordInput
+            aria-label={t`Secret value`}
+            placeholder={t`Value`}
+            flex={1}
+            value={secret.value}
+            onChange={(event) =>
+              handleChange(index, {
+                ...secret,
+                value: event.currentTarget.value,
+              })
+            }
+          />
+          <ActionIcon
+            aria-label={t`Remove secret`}
+            mt="xs"
+            onClick={() => handleRemove(index)}
+          >
+            <Icon name="trash" />
+          </ActionIcon>
+        </Group>
+      ))}
+      <Box>
+        <Button
+          variant="subtle"
+          size="compact-md"
+          p={0}
+          leftSection={<Icon name="add" size={12} />}
+          onClick={handleAdd}
+        >
+          {t`Add secret`}
+        </Button>
+      </Box>
+    </Stack>
+  );
+}

frontend/src/metabase/transforms/pages/NewTransformPage/CreateTransformModal/form.ts

@@ -1,3 +1,4 @@
+import { t } from "ttag";
 import * as Yup from "yup";
 
 import * as Errors from "metabase/utils/errors";
@@ -13,6 +14,18 @@ import {
   buildIncrementalTarget,
   getInitialValues as incrementalTransformGetInitialValues,
 } from "../../../components/IncrementalTransform";
+import { isIngestionSource, isValidSecretName } from "../../../utils";
+
+const SECRET_SCHEMA = Yup.object({
+  name: Yup.string().default("").defined(),
+  value: Yup.string().default("").defined(),
+});
+
+export type NewTransformSecret = Yup.InferType<typeof SECRET_SCHEMA>;
+
+/** Half-filled rows are dropped on submit, so they must not block validation. */
+const isFilledSecret = ({ name, value }: NewTransformSecret) =>
+  name.trim() !== "" && value !== "";
 
 export const VALIDATION_SCHEMA = Yup.object({
   name: Yup.string().required(Errors.required),
@@ -26,6 +39,19 @@ export const VALIDATION_SCHEMA = Yup.object({
       then: (schema) => schema.required(Errors.required),
     }),
   collection_id: Yup.number().nullable().defined(),
+  secrets: Yup.array()
+    .of(SECRET_SCHEMA)
+    .default([])
+    .defined()
+    .test(
+      "secret-names",
+      () => t`Check the names of your secrets.`,
+      (secrets: NewTransformSecret[] = []) =>
+        secrets.every(
+          (secret) =>
+            !isFilledSecret(secret) || isValidSecretName(secret.name.trim()),
+        ),
+    ),
 }).concat(INCREMENTAL_TRANSFORM_VALIDATION_SCHEMA);
 
 export type NewTransformValues = Yup.InferType<typeof VALIDATION_SCHEMA>;
@@ -43,9 +69,28 @@ export const getInitialValues = (
       ? slugify(defaultValues.name)
       : "",
   collection_id: null,
+  secrets: [],
   ...incrementalTransformGetInitialValues(defaultValues),
 });
 
+const buildSecrets = (
+  source: TransformSource,
+  secrets: NewTransformSecret[],
+): Record<string, string> | undefined => {
+  if (!isIngestionSource(source)) {
+    return undefined;
+  }
+
+  const filledSecrets = secrets.filter(isFilledSecret);
+  if (filledSecrets.length === 0) {
+    return undefined;
+  }
+
+  return Object.fromEntries(
+    filledSecrets.map(({ name, value }) => [name.trim(), value]),
+  );
+};
+
 export const convertTransformFormToCreateRequest = (
   source: TransformSource,
   values: NewTransformValues,
@@ -61,10 +106,13 @@ export const convertTransformFormToCreateRequest = (
     values,
   );
 
+  const secrets = buildSecrets(source, values.secrets);
+
   return {
     name: values.name,
     source: transformSource,
     target: transformTarget,
     collection_id: values.collection_id,
+    ...(secrets != null ? { secrets } : {}),
   };
 };

frontend/src/metabase/transforms/pages/NewTransformPage/CreateTransformModal/form.unit.spec.ts

@@ -1,4 +1,10 @@
-import { VALIDATION_SCHEMA } from "./form";
+import type { PythonTransformSource } from "metabase-types/api";
+
+import {
+  type NewTransformValues,
+  VALIDATION_SCHEMA,
+  convertTransformFormToCreateRequest,
+} from "./form";
 
 // Full valid form payload; individual tests override only the fields they care about.
 const baseValues = {
@@ -12,6 +18,27 @@ const baseValues = {
   uniqueKey: "",
 };
 
+const DATABASE_ID = 1;
+
+const createPythonSource = (
+  opts: Partial<PythonTransformSource> = {},
+): PythonTransformSource => ({
+  type: "python",
+  body: "def transform(): ...",
+  "source-database": DATABASE_ID,
+  "source-tables": [],
+  ...opts,
+});
+
+const createFormValues = (
+  secrets: NewTransformValues["secrets"],
+): NewTransformValues => ({
+  ...baseValues,
+  lookbackValue: null,
+  lookbackUnit: "day",
+  secrets,
+});
+
 describe("CreateTransformModal VALIDATION_SCHEMA (GDGT-2144)", () => {
   describe("when the database supports schemas", () => {
     const context = { supportsSchemas: true };
@@ -59,4 +86,73 @@ describe("CreateTransformModal VALIDATION_SCHEMA (GDGT-2144)", () => {
       ).resolves.toBeTruthy();
     });
   });
+
+  describe("secrets", () => {
+    const context = { supportsSchemas: true };
+
+    it("accepts half-filled rows, which are dropped on submit", async () => {
+      await expect(
+        VALIDATION_SCHEMA.validate(
+          {
+            ...baseValues,
+            secrets: [
+              { name: "", value: "abc" },
+              { name: "GITHUB_TOKEN", value: "" },
+            ],
+          },
+          { context },
+        ),
+      ).resolves.toBeTruthy();
+    });
+
+    it("rejects a filled row with an invalid name", async () => {
+      await expect(
+        VALIDATION_SCHEMA.validate(
+          {
+            ...baseValues,
+            secrets: [{ name: "github token", value: "abc" }],
+          },
+          { context },
+        ),
+      ).rejects.toThrow();
+    });
+  });
+});
+
+describe("convertTransformFormToCreateRequest secrets", () => {
+  const ingestionSource = createPythonSource({ ingestion: true });
+
+  it("includes only the fully filled secrets", () => {
+    const request = convertTransformFormToCreateRequest(
+      ingestionSource,
+      createFormValues([
+        { name: " GITHUB_TOKEN ", value: "ghp_1" },
+        { name: "", value: "orphan" },
+        { name: "API_KEY", value: "" },
+      ]),
+      DATABASE_ID,
+    );
+
+    expect(request.secrets).toEqual({ GITHUB_TOKEN: "ghp_1" });
+  });
+
+  it("omits secrets when there is no valid pair", () => {
+    const request = convertTransformFormToCreateRequest(
+      ingestionSource,
+      createFormValues([{ name: "API_KEY", value: "" }]),
+      DATABASE_ID,
+    );
+
+    expect(request).not.toHaveProperty("secrets");
+  });
+
+  it("omits secrets for non-ingestion sources", () => {
+    const request = convertTransformFormToCreateRequest(
+      createPythonSource(),
+      createFormValues([{ name: "API_KEY", value: "abc" }]),
+      DATABASE_ID,
+    );
+
+    expect(request).not.toHaveProperty("secrets");
+  });
 });

frontend/src/metabase/transforms/pages/NewTransformPage/utils.ts

@@ -1,3 +1,4 @@
+import { STARTER_PYTHON_BODY } from "metabase/transforms/constants";
 import Question from "metabase-lib/v1/Question";
 import type {
   Card,
@@ -29,21 +30,7 @@ export function getInitialPythonSource(): PythonTransformSourceDraft {
     type: "python",
     "source-database": undefined,
     "source-tables": [],
-    body: `# Write your Python transformation script here
-import common
-import pandas as pd
-
-def transform():
-    """
-    Your transformation function.
-
-    Select tables above to add them as function parameters.
-
-    Returns:
-        DataFrame to write to the destination table
-    """
-    # Your transformation logic here
-    return pd.DataFrame([{"message": "Hello from Python transform!"}])`,
+    body: STARTER_PYTHON_BODY,
   };
 }
 

frontend/src/metabase/transforms/pages/TransformListPage/utils.ts

@@ -5,7 +5,10 @@ import { getCollectionIcon } from "metabase/common/collections/utils";
 import { PLUGIN_REMOTE_SYNC } from "metabase/plugins";
 import { useSelector } from "metabase/redux";
 import { getMetadata } from "metabase/selectors/metadata";
-import { getLibQuery } from "metabase/transforms/utils";
+import {
+  getLibQuery,
+  hasCodeManagedSyncState,
+} from "metabase/transforms/utils";
 import type { ColorName } from "metabase/ui/colors/types";
 import * as Lib from "metabase-lib";
 import type Metadata from "metabase-lib/v1/metadata/Metadata";
@@ -48,6 +51,12 @@ export function getIncrementalWarning(
     }
   }
 
+  // Python transforms without source tables manage the sync state in code,
+  // so no checkpoint field is needed.
+  if (hasCodeManagedSyncState(transform.source)) {
+    return undefined;
+  }
+
   const checkpointFieldId =
     transform.source["source-incremental-strategy"]?.[
       "checkpoint-filter-field-id"

frontend/src/metabase/transforms/pages/TransformListPage/utils.unit.spec.ts

@@ -35,6 +35,7 @@ const dirtyEntity = (
   });
 
 jest.mock("metabase/transforms/utils", () => ({
+  ...jest.requireActual("metabase/transforms/utils"),
   getLibQuery: jest.fn(),
 }));
 

frontend/src/metabase/transforms/pages/TransformSettingsPage/TransformSettingsSection/ResetCheckpointSection.tsx

@@ -10,10 +10,43 @@ import {
 import { ConfirmModal } from "metabase/common/components/ConfirmModal";
 import { useMetadataToasts } from "metabase/metadata/hooks";
 import { CheckpointValue } from "metabase/transforms/components/CheckpointValue";
-import { isTransformRunning } from "metabase/transforms/utils";
+import {
+  hasCodeManagedSyncState,
+  isTransformRunning,
+} from "metabase/transforms/utils";
 import { Box, Button, Code, Group, Icon, Text } from "metabase/ui";
 import type { Transform } from "metabase-types/api";
 
+function SyncCursorSection({
+  transform,
+  onReset,
+  isResetting,
+}: {
+  transform: Transform;
+  onReset: () => void;
+  isResetting: boolean;
+}) {
+  const labelId = useId();
+
+  return (
+    <Group gap="md" align="center">
+      <Box c="text-secondary" role="group" aria-labelledby={labelId}>
+        <span id={labelId}>{t`Sync state`}: </span>
+        <Code bg="background_page-tertiary" style={{ whiteSpace: "pre-wrap" }}>
+          {transform.incremental_state}
+        </Code>
+      </Box>
+      <Button
+        leftSection={<Icon name="revert" aria-hidden />}
+        disabled={isTransformRunning(transform) || isResetting}
+        onClick={onReset}
+      >
+        {t`Reset state`}
+      </Button>
+    </Group>
+  );
+}
+
 export function ResetCheckpointSection({
   transform,
 }: {
@@ -43,7 +76,34 @@ export function ResetCheckpointSection({
     }
   };
 
-  if (transform.last_checkpoint_value == null) {
+  // An ingestion transform has no checkpoint field: its state is whatever its code returned,
+  // so show that instead. Resetting it is the same endpoint — it clears both.
+  const isCodeManaged =
+    transform.source != null && hasCodeManagedSyncState(transform.source);
+  if (isCodeManaged) {
+    if (transform.incremental_state == null) {
+      return null;
+    }
+    return (
+      <>
+        <SyncCursorSection
+          transform={transform}
+          onReset={openModal}
+          isResetting={isLoading}
+        />
+        <ConfirmModal
+          title={t`Reset the sync state?`}
+          message={t`The next run will start from scratch instead of continuing from the stored position. Depending on the transform, this may re-fetch everything from the source.`}
+          opened={isModalOpen}
+          onClose={closeModal}
+          onConfirm={handleConfirm}
+          confirmButtonText={t`Reset state`}
+        />
+      </>
+    );
+  }
+
+  if (transform.incremental_state == null) {
     return null;
   }
 
@@ -62,7 +122,7 @@ export function ResetCheckpointSection({
         <span id={labelId}>{label}: </span>
         <Text component="span" fw="bold" c="text-primary">
           <CheckpointValue
-            value={transform.last_checkpoint_value}
+            value={transform.incremental_state}
             checkpointField={checkpointField}
           />
         </Text>

frontend/src/metabase/transforms/pages/TransformSettingsPage/TransformSettingsSection/SecretsSection.tsx

@@ -0,0 +1,156 @@
+import { useState } from "react";
+import { t } from "ttag";
+
+import { useUpdateTransformMutation } from "metabase/api";
+import { TitleSection } from "metabase/common/data-studio/components/TitleSection";
+import { useMetadataToasts } from "metabase/metadata/hooks";
+import { SecretNameInput } from "metabase/transforms/components/SecretNameInput";
+import { isValidSecretName } from "metabase/transforms/utils";
+import {
+  ActionIcon,
+  Button,
+  Code,
+  Divider,
+  Group,
+  Icon,
+  PasswordInput,
+  Stack,
+  Text,
+} from "metabase/ui";
+import type { Transform } from "metabase-types/api";
+
+type SecretsSectionProps = {
+  transform: Transform;
+  readOnly?: boolean;
+};
+
+export function SecretsSection({ transform, readOnly }: SecretsSectionProps) {
+  const secretKeys = transform.secret_keys ?? [];
+
+  return (
+    <TitleSection
+      label={t`Secrets`}
+      description={t`Secrets are passed to this transform's Python code in the secrets argument, keyed by name. Values can't be viewed after saving.`}
+      data-testid="transform-secrets-section"
+    >
+      {secretKeys.length > 0 ? (
+        <Stack gap={0}>
+          {secretKeys.map((name) => (
+            <SecretRow
+              key={name}
+              transform={transform}
+              name={name}
+              readOnly={readOnly}
+            />
+          ))}
+        </Stack>
+      ) : (
+        <Group p="lg">
+          <Text c="text-secondary">{t`No secrets configured yet.`}</Text>
+        </Group>
+      )}
+      {!readOnly && (
+        <>
+          <Divider />
+          <AddSecretForm transform={transform} secretKeys={secretKeys} />
+        </>
+      )}
+    </TitleSection>
+  );
+}
+
+type SecretRowProps = {
+  transform: Transform;
+  name: string;
+  readOnly?: boolean;
+};
+
+function SecretRow({ transform, name, readOnly }: SecretRowProps) {
+  const [updateTransform, { isLoading }] = useUpdateTransformMutation();
+  const { sendSuccessToast, sendErrorToast } = useMetadataToasts();
+
+  const handleRemove = async () => {
+    const { error } = await updateTransform({
+      id: transform.id,
+      secrets: { [name]: null },
+    });
+    if (error) {
+      sendErrorToast(t`Failed to remove secret ${name}`);
+    } else {
+      sendSuccessToast(t`Secret ${name} removed`);
+    }
+  };
+
+  return (
+    <Group
+      justify="space-between"
+      wrap="nowrap"
+      px="lg"
+      py="md"
+      data-testid="transform-secret-row"
+    >
+      <Group gap="sm" wrap="nowrap">
+        <Code fz="md">{name}</Code>
+        <Text c="text-secondary" fz="sm">{t`Value hidden`}</Text>
+      </Group>
+      {!readOnly && (
+        <ActionIcon
+          aria-label={t`Remove secret ${name}`}
+          disabled={isLoading}
+          onClick={handleRemove}
+        >
+          <Icon name="trash" />
+        </ActionIcon>
+      )}
+    </Group>
+  );
+}
+
+type AddSecretFormProps = {
+  transform: Transform;
+  secretKeys: string[];
+};
+
+function AddSecretForm({ transform, secretKeys }: AddSecretFormProps) {
+  const [name, setName] = useState("");
+  const [value, setValue] = useState("");
+  const [updateTransform, { isLoading }] = useUpdateTransformMutation();
+  const { sendSuccessToast, sendErrorToast } = useMetadataToasts();
+
+  const trimmedName = name.trim();
+  const isNameValid = isValidSecretName(trimmedName);
+  const isReplacing = secretKeys.includes(trimmedName);
+  const canSave = isNameValid && value.length > 0 && !isLoading;
+
+  const handleSave = async () => {
+    const { error } = await updateTransform({
+      id: transform.id,
+      secrets: { [trimmedName]: value },
+    });
+    if (error) {
+      sendErrorToast(t`Failed to save secret ${trimmedName}`);
+    } else {
+      sendSuccessToast(t`Secret ${trimmedName} saved`);
+      setName("");
+      setValue("");
+    }
+  };
+
+  return (
+    <Stack p="lg" gap="md">
+      <Group align="flex-end" wrap="nowrap">
+        <SecretNameInput maw="16rem" value={name} onChange={setName} />
+        <PasswordInput
+          label={t`Value`}
+          placeholder={t`Secret value`}
+          value={value}
+          flex={1}
+          onChange={(event) => setValue(event.currentTarget.value)}
+        />
+        <Button disabled={!canSave} loading={isLoading} onClick={handleSave}>
+          {isReplacing ? t`Replace secret` : t`Add secret`}
+        </Button>
+      </Group>
+    </Stack>
+  );
+}

frontend/src/metabase/transforms/pages/TransformSettingsPage/TransformSettingsSection/TransformSettingsSection.tsx

@@ -19,8 +19,13 @@ import { Button, Divider, Group, Icon, Loader, Stack, Text } from "metabase/ui";
 import * as Urls from "metabase/urls";
 import type { IconName, Transform, UserId } from "metabase-types/api";
 
-import { isTransformRunning, sourceDatabaseId } from "../../../utils";
+import {
+  isIngestionSource,
+  isTransformRunning,
+  sourceDatabaseId,
+} from "../../../utils";
 
+import { SecretsSection } from "./SecretsSection";
 import { UpdateIncrementalSettings } from "./UpdateIncrementalSettings";
 import { UpdateTargetModal } from "./UpdateTargetModal";
 
@@ -60,6 +65,9 @@ export const TransformSettingsSection = ({
         )}
       </TitleSection>
       <UpdateIncrementalSettings transform={transform} readOnly={readOnly} />
+      {isIngestionSource(transform.source) && (
+        <SecretsSection transform={transform} readOnly={readOnly} />
+      )}
     </Stack>
   );
 };

frontend/src/metabase/transforms/pages/TransformSettingsPage/TransformSettingsSection/UpdateIncrementalSettings.tsx

@@ -1,6 +1,6 @@
 import { useDisclosure } from "@mantine/hooks";
 import { useFormikContext } from "formik";
-import { useCallback, useRef } from "react";
+import { useCallback, useMemo, useRef } from "react";
 import { t } from "ttag";
 import _ from "underscore";
 
@@ -12,6 +12,7 @@ import {
   IncrementalTransformSettings,
   useUpdateIncrementalSettings,
 } from "metabase/transforms/components/IncrementalTransform";
+import { hasCodeManagedSyncState } from "metabase/transforms/utils";
 import type { Transform } from "metabase-types/api";
 
 import { ResetCheckpointSection } from "./ResetCheckpointSection";
@@ -37,6 +38,8 @@ const IncrementalTransformSettingsWrapper = ({
     }
   };
 
+  const hasResettableState = transform.incremental_state != null;
+
   return (
     <IncrementalTransformSettings
       source={transform.source}
@@ -46,7 +49,7 @@ const IncrementalTransformSettingsWrapper = ({
       readOnly={readOnly}
       targetTableId={transform.table?.id}
       extraActions={
-        !readOnly && transform.last_checkpoint_value != null ? (
+        !readOnly && hasResettableState ? (
           <ResetCheckpointSection transform={transform} />
         ) : undefined
       }
@@ -97,10 +100,18 @@ export function UpdateIncrementalSettings({
   const { initialValues, validationSchema, updateIncrementalSettings } =
     useUpdateIncrementalSettings(transform);
 
+  const validationContext = useMemo(
+    () => ({
+      hasCodeManagedSyncState: hasCodeManagedSyncState(transform.source),
+    }),
+    [transform.source],
+  );
+
   return (
     <FormProvider
       initialValues={initialValues}
       validationSchema={validationSchema}
+      validationContext={validationContext}
       onSubmit={_.noop}
       enableReinitialize
     >
@@ -136,7 +147,9 @@ function useCheckpointChangeInterceptor(transform: Transform) {
       const waitingForCheckpointSelection =
         values.incremental &&
         values.sourceStrategy === "checkpoint" &&
-        values.checkpointFilterFieldId == null;
+        values.checkpointFilterFieldId == null &&
+        // No checkpoint field is expected when the transform code manages the sync state.
+        !hasCodeManagedSyncState(transform.source);
 
       if (waitingForCheckpointSelection) {
         return false;
@@ -147,7 +160,7 @@ function useCheckpointChangeInterceptor(transform: Transform) {
         values.checkpointFilterFieldId != null &&
         String(currentFieldId) !== values.checkpointFilterFieldId;
 
-      if (fieldChanged && transform.last_checkpoint_value != null) {
+      if (fieldChanged && transform.incremental_state != null) {
         return new Promise<boolean>((resolve) => {
           pendingResolve.current = resolve;
           openModal();

frontend/src/metabase/transforms/utils.ts

@@ -191,6 +191,27 @@ export function isSourceEmpty(
   return !nativeQuery?.trim();
 }
 
+/**
+ * An ingestion transform fetches its own data over the network instead of reading source tables.
+ * It is the only kind that can hold secrets and that manages its own incremental state.
+ */
+export function isIngestionSource(source: DraftTransformSource): boolean {
+  return source.type === "python" && source.ingestion === true;
+}
+
+export const SECRET_NAME_REGEX = /^[A-Z][A-Z0-9_]*$/;
+export const MAX_SECRET_NAME_LENGTH = 64;
+
+/** Secret names follow the usual env var shape: uppercase letters, digits and underscores. */
+export function isValidSecretName(name: string): boolean {
+  return SECRET_NAME_REGEX.test(name) && name.length <= MAX_SECRET_NAME_LENGTH;
+}
+
+/** Ingestion transforms manage their own sync state in code, not via a checkpoint field. */
+export function hasCodeManagedSyncState(source: DraftTransformSource): boolean {
+  return isIngestionSource(source);
+}
+
 export function isCompleteSource(
   source: DraftTransformSource,
 ): source is TransformSource {

resources/migrations/064/20260724_transform_secrets_incremental_state.yaml

@@ -0,0 +1,52 @@
+## Ingestion support for python transforms: `secrets` holds an encrypted JSON map of env vars
+## (e.g. API tokens) injected into the python runner subprocess; the resume point a run leaves
+## behind lives in `incremental_state`, shared with the checkpoint watermark.
+
+databaseChangeLog:
+  - changeSet:
+      id: v64.2026-07-24T10:00:00
+      author: bronsa
+      comment: Add transform.secrets
+      changes:
+        - addColumn:
+            tableName: transform
+            columns:
+              - column:
+                  name: secrets
+                  type: ${text.type}
+                  remarks: Encrypted JSON map of secret env vars passed to python transform runs
+                  constraints:
+                    nullable: true
+
+  ## A transform resumes from either a checkpoint watermark or the state its own code returned,
+  ## never both, so the two share one column.
+  - changeSet:
+      id: v64.2026-07-30T10:00:00
+      author: bronsa
+      comment: Rename transform.last_checkpoint_value to transform.incremental_state
+      changes:
+        - renameColumn:
+            tableName: transform
+            oldColumnName: last_checkpoint_value
+            newColumnName: incremental_state
+            columnDataType: ${text.type}
+
+  ## liquibase's setColumnRemarks silently drops the comment on mysql, so the SQL is written out
+  - changeSet:
+      id: v64.2026-07-30T10:00:01
+      author: bronsa
+      comment: Describe what transform.incremental_state now holds
+      changes:
+        - sql:
+            dbms: h2,postgresql
+            sql: "COMMENT ON COLUMN transform.incremental_state IS 'Where the last run left off: a checkpoint watermark, or opaque JSON state returned by the transform code'"
+        - sql:
+            dbms: mysql,mariadb
+            sql: "ALTER TABLE transform MODIFY incremental_state ${text.type} COMMENT 'Where the last run left off: a checkpoint watermark, or opaque JSON state returned by the transform code'"
+      rollback:
+        - sql:
+            dbms: h2,postgresql
+            sql: "COMMENT ON COLUMN transform.incremental_state IS 'Serialized checkpoint watermark value for incremental transforms, temporal as ISO8601'"
+        - sql:
+            dbms: mysql,mariadb
+            sql: "ALTER TABLE transform MODIFY incremental_state ${text.type} COMMENT 'Serialized checkpoint watermark value for incremental transforms, temporal as ISO8601'"

src/metabase/app_db/encryption.clj

@@ -9,7 +9,8 @@
 
 (set! *warn-on-reflection* true)
 
-;; All columns encrypted via `mi/transform-encrypted-json`. The on-disk format of such a column is
+;; All columns encrypted via `mi/transform-encrypted-json` (or, for write-only columns such as
+;; `transform.secrets`, via `mi/encrypted-json-in` alone). The on-disk format of such a column is
 ;; `encrypt(json-string)`, so rotating the key only requires decrypting the raw value with the current key and
 ;; re-encrypting the resulting string. We list raw table names (not models) so this also works for enterprise models
 ;; that aren't loaded in every edition.
@@ -20,7 +21,8 @@
    [:metabase_database :admin_details]
    [:core_user :settings]
    [:channel :details]
-   [:auth_identity :credentials]])
+   [:auth_identity :credentials]
+   [:transform :secrets]])
 
 ;; Older versions of dump-to-h2 and key rotation only processed `metabase_database.details` (plus settings and
 ;; secrets), skipping every other encrypted JSON column. A dump or rotation from such a version left the skipped

src/metabase/transforms/core.clj

@@ -40,6 +40,8 @@
  [metabase.transforms.util
   add-source-readable
   is-temp-transform-table?]
+ [metabase.transforms.models.transform
+  secrets-for-run]
  [metabase.transforms.crud
   check-database-feature
   check-feature-enabled!

src/metabase/transforms/crud.clj

@@ -99,6 +99,13 @@
             (api/check-400 (contains? #{"day" "week" "month" "quarter" "year"} unit)
                            (deferred-tru "A lookback window on a date checkpoint column must use days or a coarser unit."))))))))
 
+(defn validate-ingestion-source!
+  "An ingestion transform materializes its own input, so it must declare no source tables."
+  [{:keys [source] :as transform}]
+  (when (transforms-base.u/ingestion-transform? transform)
+    (api/check-400 (empty? (:source-tables source))
+                   (deferred-tru "An ingestion transform fetches its own data and cannot have source tables."))))
+
 (defn validate-incremental-table-tag!
   "Reject a table-incremental native-query transform whose source query has no table template tag for
   the incremental range filter to target.
@@ -151,6 +158,8 @@
         (u/update-some :last_run transforms-base.u/present-run)
         (assoc :table target-table)
         (assoc :requestable_indexes (requestable-indexes transform))
+        ;; names only — secret values never leave the server
+        (assoc :secret_keys (vec (sort (map name (keys (transform.model/secrets-for-run id))))))
         transforms.u/add-source-readable)))
 
 (defn create-transform!
@@ -162,6 +171,7 @@
    (when (transforms-base.u/query-transform? body)
      (validate-transform-query! body))
    (validate-target-schema! body)
+   (validate-ingestion-source! body)
    (validate-incremental-table-tag! body)
    (validate-lookback! body)
    (let [creator-id (or creator-id api/*current-user-id*)
@@ -172,7 +182,7 @@
                                             (or (:owner_user_id body) creator-id))
                             transform     (t2/insert-returning-instance!
                                            :model/Transform
-                                           (assoc (select-keys body [:name :description :source :target :run_trigger
+                                           (assoc (select-keys body [:name :description :source :target :run_trigger :secrets
                                                                      :collection_id :owner_email])
                                                   :creator_id creator-id
                                                   :owner_user_id owner-user-id))]
@@ -200,7 +210,8 @@
                       (when (contains? body :target)
                         (validate-target-schema! new))
                       (when (contains? body :source)
-                        (validate-incremental-column-type! new))
+                        (validate-incremental-column-type! new)
+                        (validate-ingestion-source! new))
                       (when (or (contains? body :source) (contains? body :target))
                         (validate-incremental-table-tag! new)
                         (validate-lookback! new))
@@ -213,7 +224,15 @@
                                            (transforms-base.u/target-table-exists? new)))
                                  403
                                  (deferred-tru "A table with that name already exists.")))
-                    (t2/update! :model/Transform id (dissoc body :tag_ids))
+                    ;; :secrets is a per-key merge (nil value = remove) so callers can edit
+                    ;; individual secrets without knowing the other values
+                    (t2/update! :model/Transform id
+                                (cond-> (dissoc body :tag_ids)
+                                  (contains? body :secrets)
+                                  (assoc :secrets (->> (merge (update-keys (transform.model/secrets-for-run id) name)
+                                                              (:secrets body))
+                                                       (into {} (remove (comp nil? val)))
+                                                       not-empty))))
                     ;; Update tag associations if provided
                     (when (contains? body :tag_ids)
                       (transform.model/update-transform-tags! id (:tag_ids body)))

src/metabase/transforms/models/transform.clj

@@ -134,6 +134,9 @@
    :target             mi/transform-json
    ;; nil round-trips as NULL
    :table_dependencies {:in #(some-> % mi/json-in), :out mi/json-out-with-keywordization}
+   ;; write-only here: after-select drops the column, so there's no point decrypting it on reads
+   ;; (see [[secrets-for-run]] for the explicit read path)
+   :secrets            {:in mi/encrypted-json-in}
    :run_trigger        mi/transform-keyword})
 
 (defmethod collection/allowed-namespaces :model/Transform
@@ -204,20 +207,31 @@
       target-changed?
       (assoc :target_db_id target-db-id)
 
-      ;; Reset checkpoint when the incremental filter field changes
+      ;; Reset the resume point when the incremental filter field changes, so the next run rescans
+      ;; the source range from scratch rather than appending from a watermark that no longer applies.
       (let [old-field-id (get-in (t2/original transform) [:source :source-incremental-strategy :checkpoint-filter-field-id])
             new-field-id (get-in transform [:source :source-incremental-strategy :checkpoint-filter-field-id])]
         (and old-field-id (not= old-field-id new-field-id)))
-      (assoc :last_checkpoint_value nil)
+      (assoc :incremental_state nil)
 
       true
       resolve-merge-key-field-ids)))
 
 (t2/define-after-select :model/Transform
   [{:keys [source] :as transform}]
-  (if source
-    (assoc transform :source_type (transforms-base.u/transform-source-type source))
-    transform))
+  ;; Secrets never leave the model on a select; execution fetches them via [[secrets-for-run]].
+  (let [transform (dissoc transform :secrets)]
+    (if source
+      (assoc transform :source_type (transforms-base.u/transform-source-type source))
+      transform)))
+
+(defn secrets-for-run
+  "Decrypted secrets env-var map for a transform. The only way to read secrets back: selects the raw
+  column, since instances returned by the model never include it."
+  [transform-id]
+  (some-> (t2/query-one {:select [:secrets], :from [:transform], :where [:= :id transform-id]})
+          :secrets
+          mi/encrypted-json-out))
 
 (defn- hydrate-permission
   "Batched-hydrate helper: attach a permission under `k` to each transform by calling `pred`
@@ -277,12 +291,12 @@
         (let [{:keys [status checkpoint_hi_value] :as last-run} (get last-runs transform-id)
               transform (assoc transform :last_run (dissoc last-run :last_heartbeat))]
           (if (and (= status :succeeded) checkpoint_hi_value)
-            ;; ensure consistency of last_checkpoint_value with last_run
-            (if (:last_checkpoint_value transform)
-              (assoc transform :last_checkpoint_value checkpoint_hi_value)
+            ;; ensure consistency of incremental_state with last_run
+            (if (:incremental_state transform)
+              (assoc transform :incremental_state checkpoint_hi_value)
               ;; latest transform value wins, could be reset
-              (assoc transform :last_checkpoint_value
-                     (t2/select-one-fn :last_checkpoint_value [:model/Transform :last_checkpoint_value] transform-id)))
+              (assoc transform :incremental_state
+                     (t2/select-one-fn :incremental_state [:model/Transform :incremental_state] transform-id)))
             transform))))))
 
 (methodical/defmethod t2/batched-hydrate [:model/Transform :transform_tag_ids]
@@ -470,7 +484,8 @@
 (defmethod serdes/make-spec "Transform"
   [_model-name opts]
   {:copy      [:name :description :entity_id :owner_email]
-   :skip      [:source_type :target_db_id :target_table_id :last_checkpoint_value :table_dependencies]
+   :skip      [:source_type :target_db_id :target_table_id :incremental_state :table_dependencies
+               :secrets]
    :transform {:created_at         (serdes/date)
                :creator_id         (serdes/fk :model/User)
                :owner_user_id      (serdes/fk :model/User)

src/metabase/transforms/schema.clj

@@ -40,6 +40,9 @@
      [:source-tables   [:sequential ::transforms-base.u/source-table-entry]]
      [:type {:decode/normalize lib.schema.common/normalize-keyword} [:= :python]]
      [:body :string]
+     ;; Declares that the transform fetches its own data over the network: it may hold secrets,
+     ;; manages its own incremental sync state, and is executed in isolation from other transforms.
+     [:ingestion {:optional true} :boolean]
      [:source-incremental-strategy {:optional true} ::source-incremental-strategy]]]])
 
 (mr/def ::append-config

src/metabase/transforms_base/util.clj

@@ -71,6 +71,14 @@
   [transform]
   (type-is? (:source transform) :python))
 
+(defn ingestion-transform?
+  "True if `transform` declares that it fetches its own data over the network, rather than reading
+  source tables. Such a transform may hold secrets, keeps its own incremental sync state, and runs
+  isolated from other transforms."
+  [transform]
+  (boolean (and (python-transform? transform)
+                (get-in transform [:source :ingestion]))))
+
 (defn table-target?
   "True if `transform` writes to a plain table, recreated on every run."
   [transform]
@@ -100,7 +108,7 @@
 
 (defn full-incremental-run?
   "True when an incremental transform should drop-and-recreate the target rather than append.
-  Fires when `last_checkpoint_value` is nil (first run, or after the before-update hook clears the watermark on
+  Fires when `incremental_state` is nil (first run, or after the before-update hook clears the resume point on
   a `checkpoint-filter-field-id` change), or when pending index changes require rebuilding the table to apply
   physical index state.
 
@@ -111,7 +119,7 @@
   (if (contains? transform :full-incremental-run?)
     (:full-incremental-run? transform)
     (and (incremental-target? transform)
-         (or (nil? (:last_checkpoint_value transform))
+         (or (nil? (:incremental_state transform))
              (table-index/pending-changes-for-transform? id)))))
 
 (defn full-create-run?
@@ -266,12 +274,16 @@
       (some? hi-value)           (assoc :transform/checkpoint-hi (encode-checkpoint-value hi-value)))))
 
 (defn save-watermark!
-  "Commits the incremental transforms :hi watermark value to the appdb."
+  "Commits the incremental transforms :hi watermark value to the appdb.
+
+  A run with no source range has no watermark to commit, and must leave `incremental_state` alone:
+  an ingestion transform stores the state its own code returned in that same column."
   [transform-id source-range-params]
-  (let [hi-value (:value (:hi source-range-params))]
-    (t2/update! :model/Transform
-                transform-id
-                {:last_checkpoint_value (some-> hi-value encode-checkpoint-value)})))
+  (when source-range-params
+    (let [hi-value (:value (:hi source-range-params))]
+      (t2/update! :model/Transform
+                  transform-id
+                  {:incremental_state (some-> hi-value encode-checkpoint-value)}))))
 
 (defn save-run-checkpoint-range!
   "Persist the checkpoint range (lo/hi) on a transform run record.
@@ -346,7 +358,9 @@
   native query without a table variable to inject the filter into, or no checkpoint field
   selected."
   [{:keys [source] :as transform}]
-  (when (incremental-target? transform)
+  (when (and (incremental-target? transform)
+             ;; an ingestion transform keeps its own sync state, so it needs no checkpoint field
+             (not (ingestion-transform? transform)))
     (when (and (native-query-transform? transform)
                (not (some (fn [tag] (#{:table "table"} (:type tag)))
                           (lib/template-tags (:query source)))))
@@ -415,15 +429,15 @@
   (let [{:keys [checkpoint-filter-field-id lookback]} (:source-incremental-strategy source)]
     (validate-incremental-source! transform)
     (when checkpoint-filter-field-id
-      (let [{:keys [last_checkpoint_value]} (cond-> transform
-                                              (full-incremental-run? transform)
-                                              (assoc :last_checkpoint_value nil))
+      (let [{:keys [incremental_state]} (cond-> transform
+                                          (full-incremental-run? transform)
+                                          (assoc :incremental_state nil))
             db-id             (transforms-base.i/target-db-id transform)
             metadata-provider (lib-be/application-database-metadata-provider db-id)
             column            (checkpoint-column metadata-provider checkpoint-filter-field-id)
             base-type         (lib.types.isa/column-type column)
             ;; `checkpoint-lo` is the stored watermark; `lo` is the scan bound, pushed back by any lookback.
-            checkpoint-lo     (when last_checkpoint_value (parse-checkpoint-value base-type last_checkpoint_value))
+            checkpoint-lo     (when incremental_state (parse-checkpoint-value base-type incremental_state))
             lo                (cond-> checkpoint-lo
                                 (and checkpoint-lo lookback) (apply-lookback column lookback))
 

src/metabase/transforms_rest/api/transform.clj

@@ -100,7 +100,10 @@
    [:owner_user_id {:optional true} [:maybe pos-int?]]
    [:owner_email {:optional true} [:maybe :string]]
    [:owner {:optional true} [:maybe OwnerResponse]]
-   [:last_checkpoint_value {:optional true} [:maybe :string]]
+   [:incremental_state {:optional true} [:maybe :string]]
+   ;; :secrets never appears: model selects strip it (see transform model after-select).
+   ;; GET /:id carries the configured names as :secret_keys.
+   [:secret_keys {:optional true} [:sequential :string]]
    [:can_read {:optional true} :boolean]
    [:can_write {:optional true} :boolean]
    [:can_execute {:optional true} :boolean]
@@ -177,7 +180,9 @@
             [:tag_ids {:optional true} [:sequential ms/PositiveInt]]
             [:collection_id {:optional true} [:maybe ms/PositiveInt]]
             [:owner_user_id {:optional true} [:maybe ms/PositiveInt]]
-            [:owner_email {:optional true} [:maybe :string]]]]
+            [:owner_email {:optional true} [:maybe :string]]
+            ;; env-var map for python transforms; write-only (responses expose :secret_keys)
+            [:secrets {:optional true} [:maybe [:map-of ms/NonBlankString :string]]]]]
   (transforms.core/check-feature-enabled! body)
   (api/create-check :model/Transform body)
   (transforms.core/check-database-feature body)
@@ -303,7 +308,10 @@
             [:tag_ids {:optional true} [:sequential ms/PositiveInt]]
             [:collection_id {:optional true} [:maybe ms/PositiveInt]]
             [:owner_user_id {:optional true} [:maybe ms/PositiveInt]]
-            [:owner_email {:optional true} [:maybe :string]]]]
+            [:owner_email {:optional true} [:maybe :string]]
+            ;; env-var map for python transforms; write-only and merged per key: a nil value
+            ;; removes that secret, unmentioned keys are kept (responses expose :secret_keys)
+            [:secrets {:optional true} [:maybe [:map-of ms/NonBlankString [:maybe :string]]]]]]
   (api/write-check :model/Transform id)
   (transforms.core/update-transform! id body))
 
@@ -339,7 +347,7 @@
   "Reset the stored checkpoint for an incremental transform."
   [{:keys [id]} :- [:map [:id ms/PositiveInt]]]
   (api/write-check :model/Transform id)
-  (t2/update! :model/Transform id {:last_checkpoint_value nil})
+  (t2/update! :model/Transform id {:incremental_state nil})
   nil)
 
 (defn- check-feature-and-lock!

test/metabase/indexes_rest/api_test.clj

@@ -167,7 +167,7 @@
 
 (deftest incremental-mutations-force-next-run-to-rebuild-test
   (mt/with-temp [:model/Transform {transform-id :id} (assoc (temp-incremental-transform-spec)
-                                                            :last_checkpoint_value "100")]
+                                                            :incremental_state "100")]
     (letfn [(full-run? []
               (transforms-base.u/full-incremental-run? (t2/select-one :model/Transform transform-id)))]
       (testing "no index changes: the watermark alone decides, so the next run appends"
@@ -175,7 +175,7 @@
       (testing "POST leaves the watermark alone; the pending row forces the rebuild"
         (let [created (mt/user-http-request :crowberto :post 200 "index/request"
                                             {:transform_id transform-id :structured btree})]
-          (is (= "100" (t2/select-one-fn :last_checkpoint_value :model/Transform transform-id)))
+          (is (= "100" (t2/select-one-fn :incremental_state :model/Transform transform-id)))
           (is (full-run?))
           (testing "PUT keeps forcing it (update-pending)"
             (mt/user-http-request :crowberto :put 200 (str "index/request/" (:id created))

test/metabase/transforms/models_test.clj

@@ -327,10 +327,10 @@
                                :schema "public"
                                :name "test_incr"
                                :db_id db-id}
-                      :last_checkpoint_value "42"}]
+                      :incremental_state "42"}]
         (testing "checkpoint is present before update"
           (let [t (t2/select-one :model/Transform transform-id)]
-            (is (= "42" (:last_checkpoint_value t)))))
+            (is (= "42" (:incremental_state t)))))
         (testing "changing checkpoint-filter-field-id resets checkpoint"
           (t2/update! :model/Transform transform-id
                       {:source {:type "query"
@@ -341,11 +341,11 @@
                                 :source-incremental-strategy {:type "checkpoint"
                                                               :checkpoint-filter-field-id 200}}})
           (let [t (t2/select-one :model/Transform transform-id)]
-            (is (nil? (:last_checkpoint_value t)))))
+            (is (nil? (:incremental_state t)))))
         (testing "updating without changing checkpoint-filter-field-id preserves checkpoint"
           ;; Set checkpoint again
           (t2/update! :model/Transform transform-id
-                      {:last_checkpoint_value "99"})
+                      {:incremental_state "99"})
           (t2/update! :model/Transform transform-id
                       {:source {:type "query"
                                 :query {:database db-id
@@ -355,7 +355,7 @@
                                 :source-incremental-strategy {:type "checkpoint"
                                                               :checkpoint-filter-field-id 200}}})
           (let [t (t2/select-one :model/Transform transform-id)]
-            (is (= "99" (:last_checkpoint_value t)))))))))
+            (is (= "99" (:incremental_state t)))))))))
 
 (deftest lookback-round-trips-with-keyword-keys-test
   (testing "the nested lookback map comes back from the app DB with keyword keys"

test/metabase/transforms/util_test.clj

@@ -545,7 +545,7 @@
     (testing "First incremental run (no watermark) → {full-incremental-run=true}, both types bumped from the same scan"
       (analytics/clear! :metabase-transforms/incremental-rows)
       (run-cancelable-with-mocks!
-       {:id 1 :target {:type "table-incremental"} :last_checkpoint_value nil}
+       {:id 1 :target {:type "table-incremental"} :incremental_state nil}
        {:checkpoint-filter-field-id 42 :rows-available 1000}
        {:rows-affected 1000})
       (is (== 1 (:count (mt/metric-value system :metabase-transforms/incremental-rows
@@ -559,7 +559,7 @@
     (testing "Subsequent incremental run (watermark present) → {full-incremental-run=false}, attrition surfaces as sum mismatch"
       (analytics/clear! :metabase-transforms/incremental-rows)
       (run-cancelable-with-mocks!
-       {:id 1 :target {:type "table-incremental"} :last_checkpoint_value "42"}
+       {:id 1 :target {:type "table-incremental"} :incremental_state "42"}
        {:checkpoint-filter-field-id 42 :rows-available 500}
        {:rows-affected 120})
       (is (== 1 (:count (mt/metric-value system :metabase-transforms/incremental-rows
@@ -573,7 +573,7 @@
     (testing "Driver result missing :rows-affected (defensive contract) → no emission on either side"
       (analytics/clear! :metabase-transforms/incremental-rows)
       (run-cancelable-with-mocks!
-       {:id 1 :target {:type "table-incremental"} :last_checkpoint_value "42"}
+       {:id 1 :target {:type "table-incremental"} :incremental_state "42"}
        {:checkpoint-filter-field-id 42 :rows-available 999}
        {:some-other-shape "no rows-affected here"})
       (is (== 0 (:count (mt/metric-value system :metabase-transforms/incremental-rows
@@ -583,7 +583,7 @@
     (testing "Python-shaped driver result (clj-http response augmented with :rows-affected by run-python-transform-impl!) emits the metric"
       (analytics/clear! :metabase-transforms/incremental-rows)
       (run-cancelable-with-mocks!
-       {:id 1 :target {:type "table-incremental"} :last_checkpoint_value "42"}
+       {:id 1 :target {:type "table-incremental"} :incremental_state "42"}
        {:checkpoint-filter-field-id 42 :rows-available 800}
        {:status 200 :body {:exit_code 0} :rows-affected 750})
       (is (== 1 (:count (mt/metric-value system :metabase-transforms/incremental-rows
@@ -597,7 +597,7 @@
     (testing "Incremental run whose source-range-params lacks :rows-available → no emission on either side"
       (analytics/clear! :metabase-transforms/incremental-rows)
       (run-cancelable-with-mocks!
-       {:id 1 :target {:type "table-incremental"} :last_checkpoint_value "42"}
+       {:id 1 :target {:type "table-incremental"} :incremental_state "42"}
        {:checkpoint-filter-field-id 42}
        {:rows-affected 100})
       (is (== 0 (:count (mt/metric-value system :metabase-transforms/incremental-rows
@@ -609,7 +609,7 @@
       (try
         (analytics/clear! :metabase-transforms/incremental-rows)
         (run-cancelable-with-mocks!
-         {:id 1 :target {:type "table-incremental" :schema "x" :name "tgt"} :last_checkpoint_value nil}
+         {:id 1 :target {:type "table-incremental" :schema "x" :name "tgt"} :incremental_state nil}
          {:checkpoint-filter-field-id 42 :rows-available 1000}
          {:rows-affected 0})
         (is (== 0 (:count (mt/metric-value system :metabase-transforms/incremental-rows
@@ -625,7 +625,7 @@
       (try
         (analytics/clear! :metabase-transforms/incremental-rows)
         (run-cancelable-with-mocks!
-         {:id 1 :target {:type "table-incremental"} :last_checkpoint_value "42"}
+         {:id 1 :target {:type "table-incremental"} :incremental_state "42"}
          {:checkpoint-filter-field-id 42 :rows-available 500}
          {:rows-affected 120})
         (is (== 120 (:sum (mt/metric-value system :metabase-transforms/incremental-rows
@@ -639,7 +639,7 @@
                                   (fn [& _] (throw (ex-info "Synthetic emission failure" {})))]
         (is (= {:status :succeeded :result {:rows-affected 1000}}
                (run-cancelable-with-mocks!
-                {:id 1 :target {:type "table-incremental"} :last_checkpoint_value nil}
+                {:id 1 :target {:type "table-incremental"} :incremental_state nil}
                 {:checkpoint-filter-field-id 42 :rows-available 1000}
                 {:rows-affected 1000}))
             "run-cancelable-transform! returns the success envelope; the emission throw is swallowed.")))))

test/metabase/transforms_base/util_test.clj

@@ -88,18 +88,18 @@
 (deftest ^:parallel full-incremental-run?-test
   (testing "true for an incremental transform with no checkpoint yet"
     (is (true? (transforms-base.u/full-incremental-run?
-                {:target {:type "table-incremental"} :last_checkpoint_value nil}))))
+                {:target {:type "table-incremental"} :incremental_state nil}))))
   (testing "false for an incremental transform that has already recorded a watermark"
     (is (false? (transforms-base.u/full-incremental-run?
-                 {:target {:type "table-incremental"} :last_checkpoint_value "42"}))))
+                 {:target {:type "table-incremental"} :incremental_state "42"}))))
   (testing "false for non-incremental targets regardless of checkpoint value"
     (is (false? (transforms-base.u/full-incremental-run?
-                 {:target {:type "table"} :last_checkpoint_value nil})))
+                 {:target {:type "table"} :incremental_state nil})))
     (is (false? (transforms-base.u/full-incremental-run?
-                 {:target {:type :table} :last_checkpoint_value nil}))))
+                 {:target {:type :table} :incremental_state nil}))))
   (testing "accepts both string and keyword target types"
     (is (true? (transforms-base.u/full-incremental-run?
-                {:target {:type :table-incremental} :last_checkpoint_value nil})))))
+                {:target {:type :table-incremental} :incremental_state nil})))))
 
 (deftest ^:parallel supported-checkpoint-column?-test
   (testing "numeric and date/datetime columns are supported"