fix permissions when running transform as background job

Closes QUE-2302

.clj-kondo/config.edn

@@ -1,7 +1,7 @@
 ;; -*- comment-column: 100; -*-
 {:config-paths ["macros" "src" "config"]
  :linters
- { ;;
+ {;;
   ;; enabled optional linters (these are disabled by default)
   ;;
 
@@ -187,7 +187,7 @@
   :metabase/disallow-hardcoded-driver-names-in-tests
   ;; don't include `:h2` since that will probably lead to a mountain of tests where we are specifically testing
   ;; middleware or compilation behavior with just H2 since it's the default.
-  {:drivers #{ ;; core drivers
+  {:drivers #{;; core drivers
               #_:h2
               :postgres
               :mysql
@@ -419,6 +419,7 @@
   metabase-enterprise.transforms.api-test/with-transform-tags                           clojure.core/let
   metabase-enterprise.transforms.api-test/with-worker-runs                              clojure.core/let
   metabase-enterprise.transforms.test-util/with-transform-cleanup!                      clojure.core/let
+  metabase-enterprise.transforms.execute/with-transform-lifecycle                       clojure.core/let
   metabase.actions.test-util/with-actions                                               clojure.core/let
   metabase.api.common/let-404                                                           clojure.core/let
   metabase.app-db.custom-migrations.util/with-temp-scheduler!                           clojure.core/fn
@@ -735,7 +736,9 @@
   ;; `metabase-enterprise.sandbox.query-processor*-test`
   ;;
   {:pattern "^metabase(?:((?:-enterprise\\.sandbox)?\\.query-processor)|(?:\\.driver\\.sql(?!server)(?!ite))).*-test$"
-   :name    general-qp-and-driver-test-namespaces}]
+   :name    general-qp-and-driver-test-namespaces}
+  {:pattern "^metabase\\.python-runner\\..*"
+   :name    python-runner-namespaces}]
 
  :config-in-ns
  {test-namespaces

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

@@ -884,6 +884,19 @@
            enterprise/advanced-permissions
            enterprise/sandbox}}
 
+  python-runner
+  {:team "Workflows"
+   :api  #{metabase.python-runner.api
+           metabase.python-runner.core
+           metabase.python-runner.init}
+   :uses #{api
+           driver
+           models
+           query-processor
+           request
+           util
+           enterprise/transforms}}
+
   queries
   {:team "Querying"
    :api  #{metabase.queries.api
@@ -1925,7 +1938,9 @@
   {:team "Querying"
    :api  #{metabase-enterprise.transforms.api
            metabase-enterprise.transforms.core
-           metabase-enterprise.transforms.init}
+           metabase-enterprise.transforms.init
+           ;; TODO: Remove this from API once we remove the python-runner callback to metabase
+           metabase-enterprise.transforms.settings}
    :uses #{api
            app-db
            config
@@ -1942,6 +1957,8 @@
            sync
            task
            task-history
+           ;; TODO: Remove this once we're using a better file format than CSV
+           upload
            util}}
 
   enterprise/upload-management

.gitignore

@@ -156,3 +156,4 @@ CLAUDE.local.md
 
 /logs
 /.clojure-mcp
+__pycache__/

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

@@ -28,9 +28,18 @@
 (set! *warn-on-reflection* true)
 
 (mr/def ::transform-source
-  [:map
-   [:type [:= "query"]]
-   [:query [:map [:database :int]]]])
+  [:multi {:dispatch (comp keyword :type)}
+   [:query
+    [:map
+     [:type [:= "query"]]
+     [:query [:map [:database :int]]]]]
+   [:python
+    [:map {:closed true}
+     [:source-database {:optional true} :int]
+     [:source-tables   [:map-of :string :int]]
+     [:target-database :int]
+     [:type [:= "python"]]
+     [:body :string]]]])
 
 (mr/def ::transform-target
   [:map
@@ -56,19 +65,22 @@
 
 (defn- source-database-id
   [transform]
-  (-> transform :source :query :database))
+  (if (transforms.util/python-transform? transform)
+    (-> transform :source :source-database)
+    (-> transform :source :query :database)))
 
 (defn- check-database-feature
   [transform]
-  (let [database (api/check-400 (t2/select-one :model/Database (source-database-id transform))
-                                (deferred-tru "The source database cannot be found."))
-        feature (transforms.util/required-database-feature transform)]
-    (api/check-400 (not (:is_sample database))
-                   (deferred-tru "Cannot run transforms on the sample database."))
-    (api/check-400 (not (:is_audit database))
-                   (deferred-tru "Cannot run transforms on audit databases."))
-    (api/check-400 (driver.u/supports? (:engine database) feature database)
-                   (deferred-tru "The database does not support the requested transform target type."))))
+  (when (transforms.util/query-transform? transform)
+    (let [database (api/check-400 (t2/select-one :model/Database (source-database-id transform))
+                                  (deferred-tru "The source database cannot be found."))
+          feature (transforms.util/required-database-feature transform)]
+      (api/check-400 (not (:is_sample database))
+                     (deferred-tru "Cannot run transforms on the sample database."))
+      (api/check-400 (not (:is_audit database))
+                     (deferred-tru "Cannot run transforms on audit databases."))
+      (api/check-400 (driver.u/supports? (:engine database) feature database)
+                     (deferred-tru "The database does not support the requested transform target type.")))))
 
 (api.macros/defendpoint :get "/"
   "Get a list of transforms."
@@ -156,9 +168,10 @@
           new (merge old body)
           target-fields #(-> % :target (select-keys [:schema :name]))]
       (check-database-feature new)
-      (when-let [{:keys [cycle-str]} (transforms.ordering/get-transform-cycle new)]
-        (throw (ex-info (str "Cyclic transform definitions detected: " cycle-str)
-                        {:status-code 400})))
+      (when (transforms.util/query-transform? old)
+        (when-let [{:keys [cycle-str]} (transforms.ordering/get-transform-cycle new)]
+          (throw (ex-info (str "Cyclic transform definitions detected: " cycle-str)
+                          {:status-code 400}))))
       (api/check (not (and (not= (target-fields old) (target-fields new))
                            (transforms.util/target-table-exists? new)))
                  403
@@ -206,9 +219,13 @@
   (api/check-superuser)
   (let [transform (api/check-404 (t2/select-one :model/Transform id))
         start-promise (promise)]
-    (u.jvm/in-virtual-thread*
-     (transforms.execute/run-mbql-transform! transform {:start-promise start-promise
-                                                        :run-method :manual}))
+    (if (transforms.util/python-transform? transform)
+      (u.jvm/in-virtual-thread*
+       (transforms.execute/execute-python-transform! transform {:start-promise start-promise
+                                                                :run-method :manual}))
+      (u.jvm/in-virtual-thread*
+       (transforms.execute/run-mbql-transform! transform {:start-promise start-promise
+                                                          :run-method :manual})))
     (when (instance? Throwable @start-promise)
       (throw @start-promise))
     (let [result @start-promise
@@ -218,6 +235,31 @@
                               :run_id run-id})
           (assoc :status 202)))))
 
+(api.macros/defendpoint :post "/test"
+  "Test Python code execution without creating a transform."
+  [_route-params
+   _query-params
+   body :- [:map
+            [:code :string]
+            [:tables [:map-of :string :int]]]]
+  (log/info "test python code execution")
+  (api/check-superuser)
+  (try
+    (let [result (transforms.execute/call-python-runner-api! (:code body) (:tables body))]
+      (log/info "Python test execution succeeded")
+      (-> (response/response {:message (deferred-tru "Python code executed successfully")
+                              :result result})
+          (assoc :status 200)))
+    (catch Exception e
+      (log/error e "Error executing Python test code")
+      (let [ex-data (ex-data e)]
+        (-> (response/response {:message (deferred-tru "Python code execution failed")
+                                :error (.getMessage e)
+                                :stdout (:stdout ex-data)
+                                :stderr (:stderr ex-data)
+                                :exit_code (:exit-code ex-data)})
+            (assoc :status (or (:status-code ex-data) 500)))))))
+
 (def ^{:arglists '([request respond raise])} routes
   "`/api/ee/transform` routes."
   (handlers/routes

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

@@ -1,17 +1,26 @@
 (ns metabase-enterprise.transforms.execute
   (:require
    [clojure.core.async :as a]
+   [clojure.java.io :as io]
+   [clojure.string :as str]
    [metabase-enterprise.transforms.canceling :as canceling]
    [metabase-enterprise.transforms.models.transform-run :as transform-run]
+   [metabase-enterprise.transforms.python-runner :as python-runner]
    [metabase-enterprise.transforms.settings :as transforms.settings]
    [metabase-enterprise.transforms.util :as transforms.util]
+   [metabase.api.common :as api]
    [metabase.driver :as driver]
+   [metabase.driver.ddl.interface :as ddl.i]
    [metabase.driver.util :as driver.u]
    [metabase.lib.schema.common :as schema.common]
    [metabase.query-processor.pipeline :as qp.pipeline]
+   [metabase.upload.core :as upload]
+   [metabase.util.json :as json]
    [metabase.util.log :as log]
    [metabase.util.malli.registry :as mr]
-   [toucan2.core :as t2]))
+   [toucan2.core :as t2])
+  (:import
+   (java.io File)))
 
 (set! *warn-on-reflection* true)
 
@@ -26,6 +35,24 @@
   [:map
    [:overwrite? :boolean]])
 
+(defmacro with-transform-lifecycle
+  "Macro to manage transform run lifecycle. Starts a run, executes body,
+   and handles success/failure appropriately.
+
+   Usage:
+   (with-transform-lifecycle [run-id transform-id opts]
+     ;; body that returns result on success or throws on failure
+     )"
+  [[run-id-sym [transform-id opts]] & body]
+  `(let [{~run-id-sym :id} (transform-run/start-run! ~transform-id ~opts)]
+     (try
+       (let [result# (do ~@body)]
+         (transform-run/succeed-started-run! ~run-id-sym)
+         result#)
+       (catch Throwable t#
+         (transform-run/fail-started-run! ~run-id-sym {:message (.getMessage t#)})
+         (throw t#)))))
+
 (defn- sync-target!
   ([transform-id run-id]
    (let [{:keys [source target]} (t2/select-one :model/Transform transform-id)
@@ -37,22 +64,35 @@
    (log/info "Syncing target" (pr-str target) "for transform")
    (transforms.util/activate-table-and-mark-computed! database target)))
 
-(defn run-transform!
+(defn run-cancelable-transform!
   "Run a compiled transform"
-  [run-id driver transform-details opts]
-  ;; local run is responsible for status
+  [run-id run-transform!]
+  ;; local run is responsible for status, using canceling lifecycle
   (try
     (canceling/chan-start-timeout-vthread! run-id (transforms.settings/transform-timeout))
-    (binding [qp.pipeline/*canceled-chan* (a/promise-chan)]
-      (canceling/chan-start-run! run-id qp.pipeline/*canceled-chan*)
-      (driver/run-transform! driver transform-details opts))
-    (transform-run/succeed-started-run! run-id)
+    (let [cancel-chan (a/promise-chan)
+          ret (binding [qp.pipeline/*canceled-chan* cancel-chan]
+                (canceling/chan-start-run! run-id cancel-chan)
+                (run-transform! cancel-chan))]
+      (transform-run/succeed-started-run! run-id)
+      ret)
     (catch Throwable t
       (transform-run/fail-started-run! run-id {:message (.getMessage t)})
       (throw t))
     (finally
       (canceling/chan-end-run! run-id))))
 
+(defn- try-start-unless-already-running [id run-method]
+  (try
+    (transform-run/start-run! id {:run_method run-method})
+    (catch java.sql.SQLException e
+      (if (= (.getSQLState e) "23505")
+        (throw (ex-info "Transform is already running"
+                        {:error :already-running
+                         :transform-id id}
+                        e))
+        (throw e)))))
+
 (defn run-mbql-transform!
   "Run `transform` and sync its target table.
 
@@ -74,19 +114,11 @@
          (throw (ex-info "The database does not support the requested transform target type."
                          {:driver driver, :database database, :feature feature})))
        ;; mark the execution as started and notify any observers
-       (let [{run-id :id} (try
-                            (transform-run/start-run! id {:run_method run-method})
-                            (catch java.sql.SQLException e
-                              (if (= (.getSQLState e) "23505")
-                                (throw (ex-info "Transform is already running"
-                                                {:error :already-running
-                                                 :transform-id id}
-                                                e))
-                                (throw e))))]
+       (let [{run-id :id} (try-start-unless-already-running id run-method)]
          (when start-promise
            (deliver start-promise [:started run-id]))
          (log/info "Executing transform" id "with target" (pr-str target))
-         (run-transform! run-id driver transform-details opts)
+         (run-cancelable-transform! run-id (fn [_cancel-chan] (driver/run-transform! driver transform-details opts)))
          (sync-target! target database run-id)))
      (catch Throwable t
        (log/error t "Error executing transform")
@@ -95,3 +127,85 @@
          ;; but we assume nobody would catch the exception anyway
          (deliver start-promise t))
        (throw t)))))
+
+(defn call-python-runner-api!
+  "Call the Python runner API endpoint to execute Python code.
+   Returns the result map or throws on error."
+  [code table-name->id]
+  ;; TODO probably don't need this hack anymore, double check
+  (update (python-runner/execute-python-code code table-name->id)
+          :body #(if (string? %) json/decode+kw %)))
+
+(defn- debug-info-str [{:keys [exit-code stdout stderr]}]
+  (str/join "\n"
+            [(format "exit code %d" exit-code)
+             "======"
+             "stdout"
+             "======"
+             stdout
+             "stderr"
+             "======"
+             stderr]))
+
+(defn- run-python-transform! [{:keys [schema name]} {:keys [source-tables body]} db]
+  (let [driver (:engine db)
+        {:keys [body status] :as result} (call-python-runner-api! body source-tables)]
+    (if (not= 200 status)
+      (throw (ex-info (debug-info-str body)
+                      {:status-code 400
+                       :error (or (:stderr body) (:error body))
+                       :stdout (:stdout body)
+                       :stderr (:stderr body)}))
+      (try
+        ;; TODO would be nice if we have create or replace in upload
+        ;; NOTE (chris) we do have a replace, so this would be easy! but we're gonna stop using csv
+        ;; TODO we can remove this hack once we move away from upload-csv
+        (binding [api/*current-user-permissions-set* (atom #{"/"})]
+          (when-let [table (t2/select-one :model/Table
+                                          :name (ddl.i/format-name driver name)
+                                          :schema (ddl.i/format-name driver schema)
+                                          :db_id (:id db))]
+            (upload/delete-upload! table)
+            ;; TODO shouldn't this be handled in upload?
+            (t2/delete! :model/Table (:id table)))
+          ;; Create temporary CSV file from output data
+          (let [temp-file (File/createTempFile "transform-output-" ".csv")
+                csv-data (:output body)]
+            (try
+              (with-open [writer (io/writer temp-file)]
+                (.write writer ^String csv-data))
+              (upload/create-from-csv-and-sync! {:db db
+                                                 :filename (.getName temp-file)
+                                                 :file temp-file
+                                                 :schema schema
+                                                 :table-name name})
+              (finally
+                ;; Clean up temp file
+                (.delete temp-file))))
+          result)
+        (catch Exception e
+          (log/error e "Failed to to create resulting table")
+          (throw (ex-info "Failed to create the resulting table"
+                          {:error (.getMessage e)})))))))
+
+(defn execute-python-transform!
+  "Execute a Python transform by calling the python runner.
+
+  This is executing synchronously, but supports being kicked off in the background
+  by delivering the `start-promise` just before the start when the beginning of the execution has been booked
+  in the database."
+  ([transform] (execute-python-transform! transform nil))
+  ([transform {:keys [run-method start-promise]}]
+   (when (transforms.util/python-transform? transform)
+     (try
+       (let [{:keys [source target] transform-id :id} transform
+             db (t2/select-one :model/Database (:target-database source))
+             {run-id :id} (try-start-unless-already-running transform-id run-method)]
+         (some-> start-promise (deliver [:started run-id]))
+         (log/info "Executing Python transform" transform-id "with target" (pr-str target))
+         (let [result (run-cancelable-transform! run-id (fn [_cancel-chan] (run-python-transform! target source db)))]
+           {:run_id run-id
+            :result result}))
+       (catch Throwable t
+         (log/error t "Error executing Python transform")
+         (throw t))))))

enterprise/backend/src/metabase_enterprise/transforms/jobs.clj

@@ -9,6 +9,7 @@
    [metabase-enterprise.transforms.models.transform-run :as transform-run]
    [metabase-enterprise.transforms.ordering :as transforms.ordering]
    [metabase-enterprise.transforms.settings :as transforms.settings]
+   [metabase-enterprise.transforms.util :as transforms.util]
    [metabase.task.core :as task]
    [metabase.util.log :as log]
    [toucan2.core :as t2]))
@@ -73,7 +74,9 @@
           :else
           (do
             (log/info "Executing job transform" (pr-str transform-id))
-            (transforms.execute/run-mbql-transform! current-transform {:run-method run-method})
+            (if (transforms.util/python-transform? current-transform)
+              (transforms.execute/execute-python-transform! current-transform {:run-method run-method})
+              (transforms.execute/run-mbql-transform! current-transform {:run-method run-method}))
             (transforms.job-run/add-run-activity! run-id)
             (recur (conj complete (:id current-transform)) nil)))))))
 

enterprise/backend/src/metabase_enterprise/transforms/ordering.clj

@@ -56,20 +56,29 @@
 
   The result is a map of transform id -> #{transform ids the transform depends on}. Dependencies are limited to just
   the transforms in the original list -- if a transform depends on some transform not in the list, the 'extra'
-  dependency is ignored."
+  dependency is ignored. Python transforms have no dependencies and run sequentially."
   [transforms]
-  (let [transforms-by-db (->> transforms
-                              (map (fn [transform]
-                                     {(get-in transform [:source :query :database]) [transform]}))
-                              (apply merge-with into))
+  (let [{python-transforms true
+         sql-transforms false} (group-by transforms.util/python-transform? transforms)
 
-        {:keys [output-tables dependencies]} (->> transforms-by-db
+        ;; TODO: figure out python transforms deps
+        python-deps (into {} (map (fn [transform] [(:id transform) #{}])) python-transforms)
+
+        sql-transforms-by-db (->> sql-transforms
+                                  (map (fn [transform]
+                                         {(get-in transform [:source :query :database]) [transform]}))
+                                  (apply merge-with into))
+
+        {:keys [output-tables dependencies]} (->> sql-transforms-by-db
                                                   (map (fn [[db-id db-transforms]]
                                                          (qp.store/with-metadata-provider db-id
                                                            {:output-tables (output-table-map db-transforms)
                                                             :dependencies (dependency-map db-transforms)})))
-                                                  (apply merge-with merge))]
-    (update-vals dependencies #(into #{} (keep output-tables) %))))
+                                                  (apply merge-with merge))
+
+        sql-deps (update-vals dependencies #(into #{} (keep output-tables) %))]
+    ;; Merge Python and SQL dependencies
+    (merge python-deps sql-deps)))
 
 (defn find-cycle
   "Finds a path containing a cycle in the directed graph `node->children`.

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

@@ -0,0 +1,131 @@
+(ns metabase-enterprise.transforms.python-runner
+  (:require
+   [clj-http.client :as http]
+   [clojure.java.io :as io]
+   [metabase-enterprise.transforms.settings :as transforms.settings]
+   [metabase.driver :as driver]
+   [metabase.query-processor.compile :as qp.compile]
+   ^{:clj-kondo/ignore [:discouraged-namespace]}
+   [metabase.query-processor.store :as qp.store]
+   [metabase.util.json :as json]
+   [toucan2.core :as t2])
+  (:import
+   (java.io BufferedWriter OutputStream OutputStreamWriter File)
+   (java.nio.charset StandardCharsets)))
+
+(set! *warn-on-reflection* true)
+
+(defn- safe-slurp
+  "Safely slurp a file, returning empty string on error."
+  [file]
+  (try (slurp file) (catch Exception _ "")))
+
+(defn- safe-delete
+  "Safely delete a file."
+  [^File file]
+  (try (.delete file) (catch Exception _)))
+
+(defn- write-to-stream! [^OutputStream os col-names reducible-rows]
+  (let [writer (-> os
+                   (OutputStreamWriter. StandardCharsets/UTF_8)
+                   (BufferedWriter.))]
+
+    (run! (fn [row]
+            (let [row-map (zipmap col-names row)]
+              (json/encode-to row-map writer {})
+              (.newLine writer)))
+          reducible-rows)
+
+    (doto writer
+      (.flush)
+      (.close))))
+
+(defn- execute-mbql-query
+  [driver db-id query respond]
+  (driver/with-driver driver
+    (let [native (qp.compile/compile {:type :query, :database db-id :query query})
+
+          query {:database db-id
+                 :type :native
+                 :native native}]
+      (qp.store/with-metadata-provider db-id
+        (driver/execute-reducible-query driver query {} respond)))))
+
+(defn- write-table-data! [id file]
+  (let [db-id (t2/select-one-fn :db_id (t2/table-name :model/Table) :id id)
+        driver (t2/select-one-fn :engine :model/Database db-id)
+        ;; TODO: limit
+        query {:source-table id}]
+    (execute-mbql-query driver db-id query
+                        (fn [{cols-meta :cols} reducible-rows]
+                          (with-open [os (io/output-stream file)]
+                            (write-to-stream! os (mapv :name cols-meta) reducible-rows))))))
+
+(defn execute-python-code
+  "Execute Python code using the Python execution server."
+  [code table-name->id]
+  (let [mount-path (transforms.settings/python-execution-mount-path)
+        work-dir-name (str "run-" (System/currentTimeMillis) "-" (rand-int 10000))
+        work-dir (str mount-path "/" work-dir-name)
+        work-dir-file (io/file work-dir)]
+
+    ;; Ensure mount base path exists
+    (.mkdirs (io/file mount-path))
+    ;; Create working directory
+    (.mkdirs work-dir-file)
+
+    (try
+      (let [server-url (transforms.settings/python-execution-server-url)
+            table-name->file (into {} (map (fn [[table-name id]]
+                                             (let [file-name (gensym)
+                                                   file (io/file (str work-dir "/" file-name ".jsonl"))]
+                                               (write-table-data! id file)
+                                               [table-name (.getAbsolutePath file)])))
+                                   table-name->id)
+            response (http/post (str server-url "/execute")
+                                {:content-type     :json
+                                 :accept           :json
+                                 :body             (json/encode {:code          code
+                                                                 :working_dir   work-dir
+                                                                 :timeout       30
+                                                                 :table_mapping table-name->file})
+                                 :throw-exceptions false
+                                 :as               :json})
+
+            result  (:body response)
+            ;; TODO look into why some tests return json and others strings
+            result (if (string? result) (json/decode result keyword) result)]
+
+        (try
+          (if (and (= 200 (:status response))
+                   (zero? (:exit_code result)))
+            ;; Success - read the output CSV if it exists
+            (let [output-path (:output_file result)]
+              (if (and output-path (.exists (io/file output-path)))
+                {:status 200
+                 :body   {:output (slurp output-path)
+                          :stdout (safe-slurp (:stdout_file result))
+                          :stderr (safe-slurp (:stderr_file result))}}
+                {:status 500
+                 :body   {:error  "Transform did not produce output CSV"
+                          :stdout (safe-slurp (:stdout_file result))
+                          :stderr (safe-slurp (:stderr_file result))}}))
+            ;; Error from execution server
+            {:status 500
+             :body
+             {:error     (or (:error result) "Execution failed")
+              :exit-code (:exit_code result)
+              :timeout   (:timeout result)
+              :stdout    (safe-slurp (:stdout_file result))
+              :stderr    (safe-slurp (:stderr_file result))}})
+          (finally
+            ;; Clean up working directory after use
+            (try
+              (when (.exists work-dir-file)
+                ;; Delete directory and all contents
+                (run! safe-delete (reverse (file-seq work-dir-file))))
+              (catch Exception _)))))
+
+      (catch Exception e
+        {:status 500
+         :body {:error (str "Failed to connect to Python execution server: " (.getMessage e))}}))))

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

@@ -15,3 +15,25 @@
   :export?    false
   :encryption :no
   :audit      :getter)
+
+(setting/defsetting python-execution-server-url
+  (deferred-tru "URL for the Python execution server that runs transform functions.")
+  :type       :string
+  :visibility :internal
+  :default    "http://localhost:5001"
+  :feature    :transforms
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)
+
+(setting/defsetting python-execution-mount-path
+  (deferred-tru "Base path for mounted directory shared between Metabase and Python execution container.")
+  :type       :string
+  :visibility :internal
+  :default    "/tmp/python-exec-work"
+  :feature    :transforms
+  :doc        false
+  :export?    false
+  :encryption :no
+  :audit      :getter)

enterprise/backend/src/metabase_enterprise/transforms/util.clj

@@ -15,12 +15,23 @@
     (keyword schema name)
     (keyword name)))
 
+(defn query-transform?
+  "Check if this is a query transform: native query / mbql query."
+  [transform]
+  (= :query (-> transform :source :type keyword)))
+
+(defn python-transform?
+  "Check if this is a Python transform."
+  [transform]
+  (= :python (-> transform :source :type keyword)))
+
 (defn target-table-exists?
   "Test if the target table of a transform already exists."
-  [{:keys [source target] :as _transform}]
-  (let [db-id (-> source :query :database)
-        {driver :engine :as database} (t2/select-one :model/Database db-id)]
-    (driver/table-exists? driver database target)))
+  [{:keys [source target] :as transform}]
+  (when (query-transform? transform)
+    (let [db-id (-> source :query :database)
+          {driver :engine :as database} (t2/select-one :model/Database db-id)]
+      (driver/table-exists? driver database target))))
 
 (defn target-table
   "Load the `target` table of a transform from the database specified by `database-id`."

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

@@ -78,6 +78,24 @@
                                             :schema schema
                                             :name   table-name}})))))))
 
+(deftest create-python-transform-test
+  (mt/with-premium-features #{:transforms}
+    (mt/dataset transforms-dataset/transforms-test
+      (with-transform-cleanup! [table-name "gadget_products"]
+        (let [schema    (get-test-schema)
+              transform-payload {:name   "My beautiful python runner"
+                                 :source {:type "python"
+                                          :body "print('hello world')"}
+                                 :target {:type   "table"
+                                          :schema schema
+                                          :name   table-name}}
+              transform (mt/user-http-request :crowberto :post 200 "ee/transform"
+                                              transform-payload)]
+          (is (= "print('hello chris')"
+                 (-> (mt/user-http-request :crowberto :put 200 (format "ee/transform/%s" (:id transform))
+                                           (assoc-in transform-payload [:source :body] "print('hello chris')"))
+                     :source :body))))))))
+
 (deftest list-transforms-test
   (mt/test-drivers (mt/normal-drivers-with-feature :transforms/table)
     (testing "Can list without query parameters"

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

@@ -0,0 +1,149 @@
+(ns ^:mb/driver-tests metabase-enterprise.transforms.python-runner-test
+  (:require
+   [clojure.java.jdbc :as jdbc]
+   [clojure.string :as str]
+   [clojure.test :refer :all]
+   [metabase-enterprise.transforms.python-runner :as python-runner]
+   [metabase.driver.sql-jdbc.connection :as sql-jdbc.conn]
+   [metabase.sync.core :as sync]
+   [metabase.test :as mt]))
+
+(set! *warn-on-reflection* true)
+
+(defn template->regex
+  "Convert a template string with $var$ placeholders to a regex pattern.
+   Example: template->regex 'File ___PATH___/script.py, line ___LINE___'
+   will match 'File /tmp/unique-path/script.py, line 123'"
+  [template]
+  (-> template
+      (str/replace "\\" "\\\\")
+      (str/replace "." "\\.")
+      (str/replace "*" "\\*")
+      (str/replace "+" "\\+")
+      (str/replace "?" "\\?")
+      (str/replace "^" "\\^")
+      (str/replace "$" "\\$")
+      (str/replace "{" "\\{")
+      (str/replace "}" "\\}")
+      (str/replace "(" "\\(")
+      (str/replace ")" "\\)")
+      (str/replace "|" "\\|")
+      (str/replace "[" "\\[")
+      (str/replace "]" "\\]")
+      (str/replace #"___PATH___" "(/[\\\\w-]+)+")
+      (str/replace #"___LINE___" "\\\\d+")
+      re-pattern))
+
+(defn- execute [{:keys [code tables]}]
+  (:body (python-runner/execute-python-code code (or tables {}))))
+
+(deftest ^:parallel transform-function-basic-test
+  (testing "executes transform function and returns CSV output"
+    (let [transform-code (str "import pandas as pd\n"
+                              "\n"
+                              "def transform():\n"
+                              "    return pd.DataFrame({'name': ['Alice', 'Bob'], 'age': [25, 30]})")
+          result         (execute {:code transform-code})]
+      (is (=? {:output "name,age\nAlice,25\nBob,30\n"
+               :stdout "Successfully saved 2 rows to CSV\n"
+               :stderr ""}
+              result)))))
+
+(deftest ^:parallel transform-function-missing-test
+  (testing "handles missing transform function"
+    (let [result (execute {:code (str "import pandas as pd\n"
+                                      "\n"
+                                      "# No transform function defined")})]
+      (is (=? {:error     "Execution failed"
+               :exit-code 1
+               :stderr    "ERROR: User script must define a 'transform()' function\n"
+               :stdout    ""}
+              result)))))
+
+(deftest ^:parallel transform-function-wrong-return-type-test
+  (testing "handles transform function returning non-DataFrame"
+    (let [result (execute {:code (str "def transform():\n"
+                                      "    return 'not a dataframe'")})]
+      (is (=? {:error     "Execution failed"
+               :exit-code 1
+               :stderr    "ERROR: Transform function must return a pandas DataFrame, got <class 'str'>\n"
+               :stdout    ""}
+              result)))))
+
+(deftest ^:parallel transform-function-error-test
+  (testing "handles transform function with error"
+    (let [result            (execute {:code (str "def transform():\n"
+                                                 "    raise ValueError('Something went wrong')")})
+          expected-template (str "ERROR: Transform function failed: Something went wrong\n"
+                                 "Traceback (most recent call last):\n"
+                                 "  File \"/app/transform_runner.py\", line ___LINE___, in main\n"
+                                 "    result = script.transform()\n"
+                                 "             ^^^^^^^^^^^^^^^^^^\n"
+                                 "  File \"___PATH___/script.py\", line 2, in transform\n"
+                                 "    raise ValueError('Something went wrong')\n"
+                                 "ValueError: Something went wrong\n")
+          stderr-pattern    (template->regex expected-template)]
+
+      (is (=? {:error     "Execution failed"
+               :exit-code 1
+               :stderr    #(re-matches stderr-pattern %)
+               :stdout    ""
+               :timeout   false}
+              result)))))
+
+(deftest ^:parallel transform-function-complex-dataframe-test
+  (testing "can create complex DataFrames with transform"
+    (let [transform-code (str "import pandas as pd\n"
+                              "\n"
+                              "def transform():\n"
+                              "    data = {'x': [1, 2, 3], 'y': [10, 20, 30], 'z': ['a', 'b', 'c']}\n"
+                              "    return pd.DataFrame(data)")
+          result         (execute {:code transform-code})]
+      (is (=? {:output "x,y,z\n1,10,a\n2,20,b\n3,30,c\n"
+               :stdout "Successfully saved 3 rows to CSV\n"
+               :stderr ""}
+              result)))))
+
+(deftest ^:parallel transform-function-with-db-parameter-test
+  (mt/test-driver [:postgres]
+    (testing "transform function can accept db parameter for forward compatibility"
+      (let [transform-code (str "import pandas as pd\n"
+                                "\n"
+                                "def transform(df):\n"
+                                "    # Test that db object is passed but we don't use it in this test\n"
+                                "    data = {'name': ['Charlie', 'Dana'], 'score': [85, 92]}\n"
+                                "    return pd.DataFrame(data)")
+            result         (execute {:code transform-code})]
+        (is (=? {:output "name,score\nCharlie,85\nDana,92\n"
+                 :stdout "Successfully saved 2 rows to CSV\n"
+                 :stderr ""}
+                result))))))
+
+(deftest transform-function-with-working-database-test
+  (testing "transform function successfully connects to PostgreSQL database and reads data"
+    (mt/test-drivers [:postgres]
+      (mt/with-empty-db
+        (let [db-spec        (sql-jdbc.conn/db->pooled-connection-spec (mt/db))
+              _              (jdbc/execute! db-spec ["DROP TABLE IF EXISTS students"])
+              _              (jdbc/execute! db-spec ["CREATE TABLE students (id INTEGER PRIMARY KEY, name VARCHAR(100), score INTEGER)"])
+              _              (jdbc/execute! db-spec ["INSERT INTO students (id, name, score) VALUES (1, 'Alice', 85), (2, 'Bob', 92), (3, 'Charlie', 88), (4, 'Dana', 90)"])
+
+              _              (sync/sync-database! (mt/db))
+
+              transform-code (str "import pandas as pd\n"
+                                  "\n"
+                                  "def transform(students):\n"
+                                  "    # Calculate average score\n"
+                                  "    avg_score = students['score'].mean()\n"
+                                  "    result = pd.DataFrame({\n"
+                                  "        'student_count': [len(students)],\n"
+                                  "        'average_score': [round(avg_score, 2)]\n"
+                                  "    })\n"
+                                  "    return result")
+              result         (execute {:code   transform-code
+                                       :tables {"students" (mt/id :students)}})]
+
+          (is (=? {:output "student_count,average_score\n4,88.75\n"
+                   :stdout "Successfully saved 1 rows to CSV\n"
+                   :stderr ""}
+                  result)))))))

enterprise/frontend/src/metabase-enterprise/transforms/analytics.ts

@@ -34,7 +34,7 @@ export function trackTransformCreate({
   creationType,
 }: {
   triggeredFrom: "transform-page-create-menu";
-  creationType: "query" | "native" | "saved-question";
+  creationType: "query" | "native" | "python" | "saved-question";
 }) {
   trackSimpleEvent({
     event: "transform_create",

enterprise/frontend/src/metabase-enterprise/transforms/api/python-runner.ts

@@ -0,0 +1,61 @@
+import { Api } from "metabase/api";
+
+interface ExecutePythonRequest {
+  code: string;
+  tables: Record<string, number>;
+}
+
+interface ExecutePythonResponse {
+  output?: string;
+  stdout?: string;
+  stderr?: string;
+  error?: string;
+  exit_code?: number;
+  timeout?: boolean;
+}
+
+export const pythonRunnerApi = Api.injectEndpoints({
+  endpoints: (builder) => ({
+    executePython: builder.mutation<
+      ExecutePythonResponse,
+      ExecutePythonRequest
+    >({
+      query: ({ code, tables }) => ({
+        url: "/api/ee/transform/test",
+        method: "POST",
+        body: {
+          code,
+          tables,
+        },
+      }),
+      transformResponse: (response: any) => {
+        // Extract the nested result from the backend response
+        if (response?.result?.body) {
+          return {
+            output: response.result.body.output,
+            stdout: response.result.body.stdout,
+            stderr: response.result.body.stderr,
+            error: response.result.body.error,
+            exit_code: response.result.body.exit_code,
+            timeout: response.result.body.timeout,
+          };
+        }
+        return response;
+      },
+      transformErrorResponse: (response: any) => {
+        return {
+          status: response?.status || 500,
+          data: {
+            error: response?.data?.error || "Python execution failed",
+            stdout: response?.data?.stdout || "",
+            stderr: response?.data?.stderr || "",
+            exit_code: response?.data?.exit_code,
+          },
+        };
+      },
+    }),
+  }),
+  overrideExisting: true,
+});
+
+export const { useExecutePythonMutation } = pythonRunnerApi;

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

@@ -0,0 +1,300 @@
+import { useState } from "react";
+import { t } from "ttag";
+
+import { useListDatabasesQuery, useListTablesQuery } from "metabase/api";
+import { LoadingAndErrorWrapper } from "metabase/common/components/LoadingAndErrorWrapper";
+import {
+  ActionIcon,
+  Box,
+  Button,
+  Group,
+  Icon,
+  Select,
+  Stack,
+  Text,
+  TextInput,
+} from "metabase/ui";
+import type { Database, Table } from "metabase-types/api";
+
+type TableSelection = {
+  id: string; // unique ID for this selection row
+  tableId: number | undefined;
+  alias: string;
+  aliasManuallySet?: boolean; // Track if user manually edited the alias
+};
+
+type PythonDataPickerProps = {
+  database?: number;
+  tables?: Record<string, number>; // alias -> table-id mapping
+  onChange: (database: number, tables: Record<string, number>) => void;
+};
+
+// Helper function to slugify table names
+function slugify(text: string): string {
+  return text
+    .toLowerCase()
+    .replace(/[^\w\s-]/g, "") // Remove non-word chars
+    .replace(/[\s_-]+/g, "_") // Replace spaces, underscores, hyphens with single underscore
+    .replace(/^_+|_+$/g, ""); // Remove leading/trailing underscores
+}
+
+export function PythonDataPicker({
+  database,
+  tables,
+  onChange,
+}: PythonDataPickerProps) {
+  const [selectedDatabaseId, setSelectedDatabaseId] = useState<
+    number | undefined
+  >(database);
+
+  // Initialize table selections from props
+  const initializeTableSelections = (): TableSelection[] => {
+    if (tables && Object.keys(tables).length > 0) {
+      return Object.entries(tables).map(([alias, tableId], index) => ({
+        id: `table-${index}-${Date.now()}`,
+        tableId,
+        alias,
+        aliasManuallySet: true, // Existing aliases are considered manually set
+      }));
+    }
+    // Start with one empty row by default
+    return [
+      {
+        id: `table-0-${Date.now()}`,
+        tableId: undefined,
+        alias: "",
+        aliasManuallySet: false,
+      },
+    ];
+  };
+
+  const [tableSelections, setTableSelections] = useState<TableSelection[]>(
+    initializeTableSelections,
+  );
+
+  const {
+    data: databasesResponse,
+    isLoading: isLoadingDatabases,
+    error: databaseError,
+  } = useListDatabasesQuery();
+
+  const {
+    data: tablesData,
+    isLoading: isLoadingTables,
+    error: tablesError,
+  } = useListTablesQuery(
+    selectedDatabaseId
+      ? {
+          dbId: selectedDatabaseId,
+          include_hidden: false,
+          include_editable_data_model: true,
+        }
+      : undefined,
+    {
+      skip: !selectedDatabaseId,
+    },
+  );
+
+  // Helper to notify parent of changes
+  const notifyParentOfChange = (
+    dbId: number | undefined,
+    selections: TableSelection[],
+  ) => {
+    if (dbId) {
+      const tablesMap: Record<string, number> = {};
+      selections.forEach((selection) => {
+        if (selection.tableId && selection.alias) {
+          tablesMap[selection.alias] = selection.tableId;
+        }
+      });
+      onChange(dbId, tablesMap);
+    }
+  };
+
+  if (databaseError || tablesError) {
+    return <LoadingAndErrorWrapper error={databaseError || tablesError} />;
+  }
+
+  if (isLoadingDatabases) {
+    return <LoadingAndErrorWrapper loading />;
+  }
+
+  const databases = databasesResponse?.data || [];
+  const databaseOptions = databases.map((db: Database) => ({
+    value: db.id.toString(),
+    label: db.name,
+  }));
+
+  const availableTables = (tablesData || [])
+    .filter((tbl: Table) => tbl.db_id === selectedDatabaseId && tbl.active)
+    .map((tbl: Table) => ({
+      value: tbl.id.toString(),
+      label: tbl.display_name || tbl.name,
+      table: tbl,
+    }));
+
+  const handleDatabaseChange = (value: string | null) => {
+    const dbId = value ? parseInt(value) : undefined;
+    setSelectedDatabaseId(dbId);
+    // Reset table selections when database changes
+    const newSelections = [
+      {
+        id: `table-0-${Date.now()}`,
+        tableId: undefined,
+        alias: "",
+        aliasManuallySet: false,
+      },
+    ];
+    setTableSelections(newSelections);
+    notifyParentOfChange(dbId, newSelections);
+  };
+
+  const handleTableChange = (
+    selectionId: string,
+    value: string | null,
+  ) => {
+    const newSelections = tableSelections.map((selection) => {
+      if (selection.id === selectionId) {
+        const tableId = value ? parseInt(value) : undefined;
+        const table = availableTables.find(
+          (t) => t.value === value,
+        )?.table;
+        
+        // Generate new alias from table name
+        const newAlias = table ? slugify(table.name) : "";
+        
+        return {
+          ...selection,
+          tableId,
+          // Update alias unless user has manually set it
+          alias: selection.aliasManuallySet ? selection.alias : newAlias,
+          aliasManuallySet: selection.aliasManuallySet || false,
+        };
+      }
+      return selection;
+    });
+    setTableSelections(newSelections);
+    notifyParentOfChange(selectedDatabaseId, newSelections);
+  };
+
+  const handleAliasChange = (selectionId: string, alias: string) => {
+    const newSelections = tableSelections.map((selection) =>
+      selection.id === selectionId
+        ? { ...selection, alias, aliasManuallySet: true }
+        : selection,
+    );
+    setTableSelections(newSelections);
+    notifyParentOfChange(selectedDatabaseId, newSelections);
+  };
+
+  const handleAddTable = () => {
+    const newSelections = [
+      ...tableSelections,
+      {
+        id: `table-${tableSelections.length}-${Date.now()}`,
+        tableId: undefined,
+        alias: "",
+        aliasManuallySet: false,
+      },
+    ];
+    setTableSelections(newSelections);
+    // Don't notify parent for adding empty row
+  };
+
+  const handleRemoveTable = (selectionId: string) => {
+    const newSelections = tableSelections.filter(
+      (selection) => selection.id !== selectionId,
+    );
+    setTableSelections(newSelections);
+    notifyParentOfChange(selectedDatabaseId, newSelections);
+  };
+
+  const selectedTableIds = new Set(
+    tableSelections
+      .filter((s) => s.tableId !== undefined)
+      .map((s) => s.tableId?.toString()),
+  );
+
+  return (
+    <Stack gap="md">
+      <Box>
+        <Text size="sm" fw="bold" mb="xs">
+          {t`Source Database`}
+        </Text>
+        <Text size="xs" c="dimmed" mb="sm">
+          {t`Select the database that contains your source data`}
+        </Text>
+        <Select
+          data={databaseOptions}
+          value={selectedDatabaseId?.toString() || null}
+          onChange={handleDatabaseChange}
+          placeholder={t`Select a database`}
+          clearable
+        />
+      </Box>
+      {selectedDatabaseId && (
+        <Box>
+          <Text size="sm" fw="bold" mb="xs">
+            {t`Source Tables`}
+          </Text>
+          <Text size="xs" c="dimmed" mb="sm">
+            {t`Select tables to use as data sources and provide aliases for each`}
+          </Text>
+          <Stack gap="sm">
+            {tableSelections.map((selection, index) => {
+              // Filter available tables to exclude already selected ones (except current selection)
+              const filteredTables = availableTables.filter((table) => 
+                !selectedTableIds.has(table.value) || table.value === selection.tableId?.toString()
+              );
+              
+              return (
+              <Group key={selection.id} gap="sm" align="flex-end">
+                <Select
+                  style={{ flex: 1 }}
+                  data={filteredTables}
+                  value={selection.tableId?.toString() || null}
+                  onChange={(value) => handleTableChange(selection.id, value)}
+                  placeholder={t`Select a table`}
+                  clearable
+                  disabled={isLoadingTables}
+                />
+                <TextInput
+                  style={{ flex: 0.7 }}
+                  value={selection.alias}
+                  onChange={(e) =>
+                    handleAliasChange(selection.id, e.target.value)
+                  }
+                  placeholder={t`Table alias`}
+                  label={index === 0 ? t`Alias` : undefined}
+                />
+                {tableSelections.length > 1 && (
+                  <ActionIcon
+                    onClick={() => handleRemoveTable(selection.id)}
+                    aria-label={t`Remove table`}
+                    color="gray"
+                    variant="subtle"
+                  >
+                    <Icon name="trash" />
+                  </ActionIcon>
+                )}
+              </Group>
+              );
+            })}
+            <Button
+              leftSection={<Icon name="add" />}
+              variant="subtle"
+              onClick={handleAddTable}
+              disabled={
+                !tableSelections[tableSelections.length - 1]?.tableId ||
+                availableTables.length === 0 ||
+                selectedTableIds.size >= availableTables.length
+              }
+            >
+              {t`Add another table`}
+            </Button>
+          </Stack>
+        </Box>
+      )}
+    </Stack>
+  );
+}

enterprise/frontend/src/metabase-enterprise/transforms/components/PythonDataPicker/index.ts

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

enterprise/frontend/src/metabase-enterprise/transforms/components/PythonResultsTable/PythonResultsTable.tsx

@@ -0,0 +1,153 @@
+import { useMemo } from "react";
+import { t } from "ttag";
+
+import { Alert, Box, Text } from "metabase/ui";
+
+interface PythonResultsTableProps {
+  csvData?: string;
+  error?: string;
+  stdout?: string;
+  stderr?: string;
+  isLoading?: boolean;
+}
+
+function parseCSV(csv: string): { headers: string[]; rows: string[][] } {
+  if (!csv || !csv.trim()) {
+    return { headers: [], rows: [] };
+  }
+
+  // Split by newlines to get rows
+  const lines = csv.trim().split("\n");
+  if (lines.length === 0) {
+    return { headers: [], rows: [] };
+  }
+
+  // Parse headers from first line
+  const headers = lines[0].split(",").map(h => h.trim());
+  
+  // Parse data rows
+  const rows = lines.slice(1).map(line => {
+    // Simple CSV parsing - handles basic comma-separated values
+    // For more complex CSV (with quotes, escapes), we'd need a proper parser
+    return line.split(",").map(cell => cell.trim());
+  });
+
+  return { headers, rows };
+}
+
+export function PythonResultsTable({
+  csvData,
+  error,
+  stdout,
+  stderr,
+  isLoading = false,
+}: PythonResultsTableProps) {
+  console.log("PythonResultsTable received props:", {
+    csvData,
+    error,
+    stdout,
+    stderr,
+    isLoading,
+  });
+  
+  const { headers, rows } = useMemo(() => {
+    if (!csvData) {
+      console.log("No CSV data to parse");
+      return { headers: [], rows: [] };
+    }
+    console.log("Parsing CSV data:", csvData);
+    const parsed = parseCSV(csvData);
+    console.log("Parsed result:", parsed);
+    return parsed;
+  }, [csvData]);
+
+  if (isLoading) {
+    return (
+      <Box p="lg">
+        <Text c="text-medium">{t`Running Python script...`}</Text>
+      </Box>
+    );
+  }
+
+  if (error) {
+    return (
+      <Box p="lg">
+        <Alert color="red" title={t`Execution Error`} mb="md">
+          {error}
+        </Alert>
+        {stdout && (
+          <Box mb="md">
+            <Text fw="bold" mb="xs">{t`Standard Output:`}</Text>
+            <Box p="sm" bg="bg-light" style={{ fontFamily: "monospace", whiteSpace: "pre-wrap" }}>
+              {stdout}
+            </Box>
+          </Box>
+        )}
+        {stderr && (
+          <Box>
+            <Text fw="bold" mb="xs">{t`Standard Error:`}</Text>
+            <Box p="sm" bg="bg-light" style={{ fontFamily: "monospace", whiteSpace: "pre-wrap" }}>
+              {stderr}
+            </Box>
+          </Box>
+        )}
+      </Box>
+    );
+  }
+
+  if (!csvData || headers.length === 0) {
+    return (
+      <Box p="lg">
+        <Text c="text-medium">{t`No results to display. Click "Run" to execute the script.`}</Text>
+        {stdout && (
+          <Box mt="md">
+            <Text fw="bold" mb="xs">{t`Standard Output:`}</Text>
+            <Box p="sm" bg="bg-light" style={{ fontFamily: "monospace", whiteSpace: "pre-wrap" }}>
+              {stdout}
+            </Box>
+          </Box>
+        )}
+      </Box>
+    );
+  }
+
+  console.log("Rendering table with:", { headers, rows });
+  
+  return (
+    <Box style={{ overflow: "auto", padding: "1rem", background: "#f5f5f5", border: "1px solid #ddd" }}>
+      {stdout && (
+        <Box p="md" pb={0}>
+          <Text fw="bold" mb="xs">{t`Standard Output:`}</Text>
+          <Box p="sm" bg="bg-light" style={{ fontFamily: "monospace", whiteSpace: "pre-wrap" }} mb="md">
+            {stdout}
+          </Box>
+        </Box>
+      )}
+      <Box style={{ background: "white", padding: "1rem" }}>
+        <Text fw="bold" mb="md">{t`Results Table:`}</Text>
+        <table style={{ width: "100%", borderCollapse: "collapse", border: "1px solid #ddd" }}>
+          <thead>
+            <tr style={{ backgroundColor: "#f0f0f0" }}>
+              {headers.map((header, index) => (
+                <th key={index} style={{ padding: "8px", textAlign: "left", borderBottom: "2px solid #ddd" }}>
+                  {header}
+                </th>
+              ))}
+            </tr>
+          </thead>
+          <tbody>
+            {rows.map((row, rowIndex) => (
+              <tr key={rowIndex} style={{ borderBottom: "1px solid #eee" }}>
+                {row.map((cell, cellIndex) => (
+                  <td key={cellIndex} style={{ padding: "8px" }}>
+                    {cell}
+                  </td>
+                ))}
+              </tr>
+            ))}
+          </tbody>
+        </table>
+      </Box>
+    </Box>
+  );
+}

enterprise/frontend/src/metabase-enterprise/transforms/components/PythonResultsTable/index.ts

@@ -0,0 +1 @@
+export { PythonResultsTable } from "./PythonResultsTable";

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

@@ -0,0 +1,104 @@
+import { useHotkeys } from "@mantine/hooks";
+import { useState, useEffect } from "react";
+
+import { Flex } from "metabase/ui";
+import type { TransformSource } from "metabase-types/api";
+
+import { PythonDataPicker } from "../PythonDataPicker";
+import { EditorHeader } from "../QueryEditor/EditorHeader";
+import { PythonEditor } from "../QueryEditor/PythonEditor";
+
+type PythonTransformEditorProps = {
+  initialSource: TransformSource & { type: "python" };
+  isNew?: boolean;
+  isSaving?: boolean;
+  onSave: (newSource: TransformSource & { type: "python" }) => void;
+  onCancel: () => void;
+  onSourceChange?: (newSource: TransformSource & { type: "python" }) => void;
+};
+
+export function PythonTransformEditor({
+  initialSource,
+  isNew = true,
+  isSaving = false,
+  onSave,
+  onCancel,
+  onSourceChange,
+}: PythonTransformEditorProps) {
+  const [source, setSource] = useState(initialSource);
+  const [isSourceDirty, setIsSourceDirty] = useState(false);
+
+  useEffect(() => {
+    if (onSourceChange) {
+      onSourceChange(source);
+    }
+  }, [source, onSourceChange]);
+
+  const handleScriptChange = (body: string) => {
+    const newSource = {
+      ...source,
+      body,
+    };
+    setSource(newSource);
+    setIsSourceDirty(true);
+  };
+
+  const handleDataChange = (
+    database: number,
+    tables: Record<string, number>,
+  ) => {
+    const newSource = {
+      ...source,
+      "source-database": database,
+      "source-tables": tables,
+    };
+    setSource(newSource);
+    setIsSourceDirty(true);
+  };
+
+  const handleSave = () => {
+    onSave(source);
+  };
+
+
+  const canSave = Boolean(
+    source.body.trim() &&
+      source["source-database"] &&
+      source["source-tables"] &&
+      Object.keys(source["source-tables"]).length > 0,
+  );
+
+  return (
+    <Flex
+      w="100%"
+      h="100%"
+      direction="column"
+      bg="bg-white"
+      data-testid="python-transform-editor"
+    >
+      <EditorHeader
+        isNew={isNew}
+        isSaving={isSaving}
+        canSave={canSave && (isNew || isSourceDirty)}
+        onSave={handleSave}
+        onCancel={onCancel}
+      />
+      <Flex direction="column" style={{ flex: 1 }} p="md">
+        <PythonDataPicker
+          database={source["source-database"]}
+          tables={source["source-tables"]}
+          onChange={handleDataChange}
+        />
+
+        <Flex direction="column" style={{ flex: 1, marginTop: "1rem" }}>
+          <PythonEditor
+            script={source.body}
+            isRunnable={true}
+            onChange={handleScriptChange}
+            tables={source["source-tables"]}
+          />
+        </Flex>
+      </Flex>
+    </Flex>
+  );
+}

enterprise/frontend/src/metabase-enterprise/transforms/components/PythonTransformEditor/index.ts

@@ -0,0 +1 @@
+export { PythonTransformEditor } from "./PythonTransformEditor";

enterprise/frontend/src/metabase-enterprise/transforms/components/QueryEditor/EditorBody/EditorBody.tsx

@@ -11,7 +11,13 @@ import { doesDatabaseSupportTransforms } from "metabase-enterprise/transforms/ut
 import type Question from "metabase-lib/v1/Question";
 import type Database from "metabase-lib/v1/metadata/Database";
 import type NativeQuery from "metabase-lib/v1/queries/NativeQuery";
-import type { Database as ApiDatabase, RecentItem } from "metabase-types/api";
+import type {
+  Database as ApiDatabase,
+  RecentItem,
+  TransformSource,
+} from "metabase-types/api";
+
+import { PythonEditor } from "../PythonEditor";
 
 import S from "./EditorBody.module.css";
 import { ResizableBoxHandle } from "./ResizableBoxHandle";
@@ -21,10 +27,13 @@ const EDITOR_HEIGHT = 400;
 type EditorBodyProps = {
   question: Question;
   isNative: boolean;
+  isPython?: boolean;
+  pythonSource?: TransformSource & { type: "python" };
   isRunnable: boolean;
   isRunning: boolean;
   isResultDirty: boolean;
   onChange: (newQuestion: Question) => void;
+  onPythonChange?: (script: string) => void;
   onRunQuery: () => Promise<void>;
   onCancelQuery: () => void;
   databases: ApiDatabase[];
@@ -33,10 +42,13 @@ type EditorBodyProps = {
 export function EditorBody({
   question,
   isNative,
+  isPython = false,
+  pythonSource,
   isRunnable,
   isRunning,
   isResultDirty,
   onChange,
+  onPythonChange,
   onRunQuery,
   onCancelQuery,
   databases,
@@ -70,6 +82,24 @@ export function EditorBody({
 
   const dataPickerOptions = useDataPickerOptions({ databases });
 
+  const handlePythonChange = (script: string) => {
+    onPythonChange?.(script);
+  };
+
+  if (isPython) {
+    return (
+      <PythonEditor
+        script={pythonSource?.body || ""}
+        isRunnable={isRunnable}
+        isRunning={isRunning}
+        isResultDirty={isResultDirty}
+        onChange={handlePythonChange}
+        onRunScript={onRunQuery}
+        onCancelScript={onCancelQuery}
+      />
+    );
+  }
+
   return isNative ? (
     <NativeQueryEditor
       question={question}

enterprise/frontend/src/metabase-enterprise/transforms/components/QueryEditor/PythonEditor/PythonEditor.tsx

@@ -0,0 +1,233 @@
+import { useState } from "react";
+import { t } from "ttag";
+
+import { CodeEditor } from "metabase/common/components/CodeEditor";
+import { Alert, Box, Button, Group, Text } from "metabase/ui";
+
+import { useExecutePythonMutation } from "../../../api/python-runner";
+
+type PythonEditorProps = {
+  script: string;
+  isRunnable: boolean;
+  onChange: (script: string) => void;
+  tables?: Record<string, number>;
+};
+
+interface ExecutionResult {
+  output?: string;
+  stdout?: string;
+  stderr?: string;
+  error?: string;
+}
+
+function parseCSV(csv: string): { headers: string[]; rows: string[][] } {
+  if (!csv || !csv.trim()) {
+    return { headers: [], rows: [] };
+  }
+
+  const lines = csv.trim().split("\n");
+  if (lines.length === 0) {
+    return { headers: [], rows: [] };
+  }
+
+  const headers = lines[0].split(",").map(h => h.trim());
+  const rows = lines.slice(1).map(line => {
+    return line.split(",").map(cell => cell.trim());
+  });
+
+  return { headers, rows };
+}
+
+export function PythonEditor({
+  script,
+  isRunnable,
+  onChange,
+  tables = {},
+}: PythonEditorProps) {
+  const [isRunning, setIsRunning] = useState(false);
+  const [executionResult, setExecutionResult] = useState<ExecutionResult | null>(null);
+  const [executePython] = useExecutePythonMutation();
+  const handleScriptChange = (newScript: string) => {
+    onChange(newScript);
+    // Don't clear results on every keystroke - keep them visible for reference
+  };
+
+  const handleRunScript = async () => {
+    setIsRunning(true);
+    setExecutionResult(null);
+
+    try {
+      const result = await executePython({
+        code: script,
+        tables: tables,
+      }).unwrap();
+
+
+      if (result?.body && result?.body.error) {
+        setExecutionResult({
+          error: result.body.error || "Execution failed",
+          stdout: result.body.stdout,
+          stderr: result.body.stderr,
+        });
+      } else {
+        setExecutionResult({
+          output: result.output,
+          stdout: result.stdout,
+          stderr: result.stderr,
+        });
+      }
+    } catch (error: any) {
+
+      // The API returns error data directly in error.data
+      const errorData = error?.data || {};
+      const errorMessage = errorData.error ||
+                          error?.message ||
+                          "Failed to execute Python script";
+      const stdout = errorData.stdout || "";
+      const stderr = errorData.stderr || "";
+
+      setExecutionResult({
+        error: errorMessage,
+        stdout: stdout,
+        stderr: stderr,
+      });
+    } finally {
+      setIsRunning(false);
+    }
+  };
+
+  const handleCancelScript = () => {
+    // Cancel functionality can be implemented if needed
+  };
+
+  const { headers, rows } = parseCSV(executionResult?.output || "");
+
+  return (
+    <Box h="100%" style={{ display: "flex", flexDirection: "column" }}>
+      {/* Python Script Editor */}
+      <Box style={{ height: "300px" }}>
+        <CodeEditor
+          value={script}
+          onChange={handleScriptChange}
+          language="python"
+        />
+      </Box>
+
+      {/* Results Section - positioned between editor and buttons */}
+      {(executionResult || isRunning) && (
+        <Box mt="md">
+          {isRunning ? (
+            <Box p="md">
+              <Text c="text-medium">{t`Running Python script...`}</Text>
+            </Box>
+          ) : executionResult?.error ? (
+            <>
+              <Box p="md">
+                <Alert color="red" title={t`Execution Error`}>
+                  {executionResult.error}
+                </Alert>
+              </Box>
+              {executionResult.stdout && (
+                <Box mt="md" p="md" bg="bg-light" style={{ maxHeight: "150px", overflow: "auto" }}>
+                  <Text fw="bold" mb="xs">{t`Standard Output:`}</Text>
+                  <Box style={{ fontFamily: "monospace", whiteSpace: "pre-wrap" }}>
+                    {executionResult.stdout}
+                  </Box>
+                </Box>
+              )}
+              {executionResult.stderr && (
+                <Box mt="md" p="md" bg="bg-light" style={{ maxHeight: "150px", overflow: "auto" }}>
+                  <Text fw="bold" mb="xs">{t`Standard Error:`}</Text>
+                  <Box style={{ fontFamily: "monospace", whiteSpace: "pre-wrap" }}>
+                    {executionResult.stderr}
+                  </Box>
+                </Box>
+              )}
+            </>
+          ) : !executionResult?.output || headers.length === 0 ? (
+            <>
+              <Box p="md">
+                <Text c="text-medium">{t`No results to display.`}</Text>
+              </Box>
+              {executionResult?.stdout && (
+                <Box mt="md" p="md" bg="bg-light" style={{ maxHeight: "150px", overflow: "auto" }}>
+                  <Text fw="bold" mb="xs">{t`Standard Output:`}</Text>
+                  <Box style={{ fontFamily: "monospace", whiteSpace: "pre-wrap" }}>
+                    {executionResult.stdout}
+                  </Box>
+                </Box>
+              )}
+              {executionResult?.stderr && (
+                <Box mt="md" p="md" bg="bg-light" style={{ maxHeight: "150px", overflow: "auto" }}>
+                  <Text fw="bold" mb="xs">{t`Standard Error:`}</Text>
+                  <Box style={{ fontFamily: "monospace", whiteSpace: "pre-wrap" }}>
+                    {executionResult.stderr}
+                  </Box>
+                </Box>
+              )}
+            </>
+          ) : (
+            <>
+              {executionResult.stdout && (
+                <Box mt="md" p="md" bg="bg-light" style={{ maxHeight: "150px", overflow: "auto" }}>
+                  <Text fw="bold" mb="xs">{t`Standard Output:`}</Text>
+                  <Box style={{ fontFamily: "monospace", whiteSpace: "pre-wrap" }}>
+                    {executionResult.stdout}
+                  </Box>
+                </Box>
+              )}
+              {executionResult.stderr && (
+                <Box mt="md" p="md" bg="bg-light" style={{ maxHeight: "150px", overflow: "auto" }}>
+                  <Text fw="bold" mb="xs">{t`Standard Error:`}</Text>
+                  <Box style={{ fontFamily: "monospace", whiteSpace: "pre-wrap" }}>
+                    {executionResult.stderr}
+                  </Box>
+                </Box>
+              )}
+              <Box mt="md" style={{ maxHeight: "300px", overflow: "auto" }}>
+                <Text fw="bold" mb="md">{t`Results Table:`}</Text>
+                <table style={{ width: "100%", borderCollapse: "collapse", border: "1px solid #ddd" }}>
+                  <thead>
+                    <tr style={{ backgroundColor: "#f0f0f0" }}>
+                      {headers.map((header, index) => (
+                        <th key={index} style={{ padding: "8px", textAlign: "left", borderBottom: "2px solid #ddd" }}>
+                          {header}
+                        </th>
+                      ))}
+                    </tr>
+                  </thead>
+                  <tbody>
+                    {rows.map((row, rowIndex) => (
+                      <tr key={rowIndex} style={{ borderBottom: "1px solid #eee" }}>
+                        {row.map((cell, cellIndex) => (
+                          <td key={cellIndex} style={{ padding: "8px" }}>
+                            {cell}
+                          </td>
+                        ))}
+                      </tr>
+                    ))}
+                  </tbody>
+                </table>
+              </Box>
+            </>
+          )}
+        </Box>
+      )}
+
+      {/* Action Buttons */}
+      <Box mt="md">
+        <Group gap="sm">
+          <Button
+            variant="filled"
+            leftSection={isRunning ? undefined : <span>▶</span>}
+            onClick={isRunning ? handleCancelScript : handleRunScript}
+            disabled={!isRunnable && !isRunning}
+            loading={isRunning}
+          >
+            {isRunning ? t`Cancel` : t`Run Python Script`}
+          </Button>
+        </Group>
+      </Box>
+    </Box>
+  );
+}

enterprise/frontend/src/metabase-enterprise/transforms/components/QueryEditor/PythonEditor/index.ts

@@ -0,0 +1 @@
+export { PythonEditor } from "./PythonEditor";

enterprise/frontend/src/metabase-enterprise/transforms/index.tsx

@@ -9,6 +9,7 @@ import { hasPremiumFeature } from "metabase-enterprise/settings";
 import { JobListPage } from "./pages/JobListPage";
 import { JobPage } from "./pages/JobPage";
 import { NewJobPage } from "./pages/NewJobPage";
+import { NewPythonTransformPage } from "./pages/NewPythonTransformPage";
 import { NewTransformQueryPage } from "./pages/NewTransformQueryPage";
 import { RunListPage } from "./pages/RunListPage";
 import { TransformListPage } from "./pages/TransformListPage";
@@ -32,7 +33,16 @@ if (hasPremiumFeature("transforms")) {
           <Route path="runs" component={RunListPage} />
           <Route path=":transformId" component={TransformPage} />
         </Route>
-        <Route path="new/:type" component={NewTransformQueryPage} />
+        <Route
+          path="new/:type"
+          component={(props: any) =>
+            props.params.type === "python" ? (
+              <NewPythonTransformPage />
+            ) : (
+              <NewTransformQueryPage params={props.params} />
+            )
+          }
+        />
         <Route path="new/card/:cardId" component={NewTransformQueryPage} />
         <Route path=":transformId/query" component={TransformQueryPage} />
       </Route>

enterprise/frontend/src/metabase-enterprise/transforms/pages/NewPythonTransformPage/CreatePythonTransformModal/CreatePythonTransformModal.tsx

@@ -0,0 +1,155 @@
+import { Field, Form, Formik } from "formik";
+import { t } from "ttag";
+import * as Yup from "yup";
+
+import { Button, Group, Modal, Stack, TextInput } from "metabase/ui";
+import { useCreateTransformMutation } from "metabase-enterprise/api";
+import type { Transform, TransformSource } from "metabase-types/api";
+
+type PythonTransformApiSource = {
+  type: "python";
+  body: string;
+  "target-database": number;
+  "source-database": number;
+  "source-tables": Record<string, number>;
+};
+
+type CreatePythonTransformModalProps = {
+  source: TransformSource & { type: "python" };
+  onCreate: (transform: Transform) => void;
+  onClose: () => void;
+};
+
+type FormValues = {
+  name: string;
+  description: string | null;
+  targetSchema: string | null;
+  targetName: string;
+};
+
+const getValidationSchema = () =>
+  Yup.object({
+    name: Yup.string().required(t`Name is required`),
+    description: Yup.string().nullable(),
+    targetSchema: Yup.string().nullable(),
+    targetName: Yup.string().required(t`Table name is required`),
+  });
+
+export function CreatePythonTransformModal({
+  source,
+  onCreate,
+  onClose,
+}: CreatePythonTransformModalProps) {
+  const [createTransform] = useCreateTransformMutation();
+
+  const initialValues: FormValues = {
+    name: "",
+    description: null,
+    targetSchema: "public",
+    targetName: "",
+  };
+
+  const handleSubmit = async (values: FormValues) => {
+    try {
+      const transformedSource: PythonTransformApiSource = {
+        type: "python",
+        body: source.body,
+        "target-database": source["source-database"], // for now the same
+        "source-database": source["source-database"],
+        "source-tables": source["source-tables"] || {},
+      };
+
+      const transform = await createTransform({
+        name: values.name,
+        description: values.description,
+        source: transformedSource as any,
+        target: {
+          type: "table",
+          schema: values.targetSchema,
+          name: values.targetName,
+        },
+      }).unwrap();
+
+      onCreate(transform);
+    } catch (error) {
+      console.error("Error creating Python transform:", error);
+      // TODO: Show error toast
+    }
+  };
+
+  return (
+    <Modal
+      opened
+      onClose={onClose}
+      title={t`Create Python Transform`}
+      size="md"
+    >
+      <Formik
+        initialValues={initialValues}
+        validationSchema={getValidationSchema()}
+        onSubmit={handleSubmit}
+      >
+        {({ isSubmitting, isValid }) => (
+          <Form>
+            <Stack gap="md">
+              <Field name="name">
+                {({ field, meta }: any) => (
+                  <TextInput
+                    {...field}
+                    label={t`Transform name`}
+                    placeholder={t`My Python Transform`}
+                    error={meta.touched && meta.error}
+                  />
+                )}
+              </Field>
+
+              <Field name="description">
+                {({ field }: any) => (
+                  <TextInput
+                    {...field}
+                    label={t`Description (optional)`}
+                    placeholder={t`What does this transform do?`}
+                  />
+                )}
+              </Field>
+
+              <Field name="targetName">
+                {({ field, meta }: any) => (
+                  <TextInput
+                    {...field}
+                    label={t`Output table name`}
+                    placeholder={t`my_python_transform`}
+                    error={meta.touched && meta.error}
+                  />
+                )}
+              </Field>
+
+              <Field name="targetSchema">
+                {({ field }: any) => (
+                  <TextInput
+                    {...field}
+                    label={t`Schema (optional)`}
+                    placeholder={t`public`}
+                  />
+                )}
+              </Field>
+
+              <Group justify="flex-end" mt="md">
+                <Button variant="subtle" onClick={onClose}>
+                  {t`Cancel`}
+                </Button>
+                <Button
+                  type="submit"
+                  loading={isSubmitting}
+                  disabled={!isValid}
+                >
+                  {t`Create transform`}
+                </Button>
+              </Group>
+            </Stack>
+          </Form>
+        )}
+      </Formik>
+    </Modal>
+  );
+}

enterprise/frontend/src/metabase-enterprise/transforms/pages/NewPythonTransformPage/CreatePythonTransformModal/index.ts

@@ -0,0 +1 @@
+export { CreatePythonTransformModal } from "./CreatePythonTransformModal";

enterprise/frontend/src/metabase-enterprise/transforms/pages/NewPythonTransformPage/NewPythonTransformPage.tsx

@@ -0,0 +1,73 @@
+import { useDisclosure } from "@mantine/hooks";
+import { useState } from "react";
+import { push } from "react-router-redux";
+
+import { useDispatch } from "metabase/lib/redux";
+import type { Transform, TransformSource } from "metabase-types/api";
+
+import { PythonTransformEditor } from "../../components/PythonTransformEditor";
+import { getTransformListUrl, getTransformUrl } from "../../urls";
+
+import { CreatePythonTransformModal } from "./CreatePythonTransformModal";
+
+const DEFAULT_PYTHON_SOURCE: TransformSource & { type: "python" } = {
+  type: "python",
+  "source-database": 1, // Default to first database, will be updated by user
+  body: `# Write your Python transformation script here
+import pandas as pd
+
+def transform(table_name):
+    """
+    Your transformation function.
+
+    Args:
+        table_name: DataFrame corresponding to the source table with the matching alias
+
+        (rename to match your actual alias, add extra arguments named for each additional table)
+
+    Returns:
+        DataFrame to write to the destination table
+    """
+    # Your transformation logic here
+    return pd.DataFrame([{"row_count": table_name.size}])`,
+};
+
+export function NewPythonTransformPage() {
+  const [source, setSource] = useState<TransformSource & { type: "python" }>(
+    DEFAULT_PYTHON_SOURCE,
+  );
+  const [isModalOpened, { open: openModal, close: closeModal }] =
+    useDisclosure();
+  const dispatch = useDispatch();
+
+  const handleCreate = (transform: Transform) => {
+    dispatch(push(getTransformUrl(transform.id)));
+  };
+
+  const handleSaveClick = (newSource: TransformSource & { type: "python" }) => {
+    setSource(newSource);
+    openModal();
+  };
+
+  const handleCancelClick = () => {
+    dispatch(push(getTransformListUrl()));
+  };
+
+  return (
+    <>
+      <PythonTransformEditor
+        initialSource={DEFAULT_PYTHON_SOURCE}
+        isNew
+        onSave={handleSaveClick}
+        onCancel={handleCancelClick}
+      />
+      {isModalOpened && (
+        <CreatePythonTransformModal
+          source={source}
+          onCreate={handleCreate}
+          onClose={closeModal}
+        />
+      )}
+    </>
+  );
+}

enterprise/frontend/src/metabase-enterprise/transforms/pages/NewPythonTransformPage/index.ts

@@ -0,0 +1 @@
+export { NewPythonTransformPage } from "./NewPythonTransformPage";

enterprise/frontend/src/metabase-enterprise/transforms/pages/NewTransformQueryPage/NewTransformQueryPage.tsx

@@ -90,7 +90,7 @@ function getParsedParams({
   cardId,
 }: NewTransformQueryPageParams): NewTransformQueryPageParsedParams {
   return {
-    type: type === "native" ? "native" : "query",
+    type: type === "native" ? "native" : type === "python" ? "query" : "query",
     cardId: cardId != null ? Urls.extractEntityId(cardId) : undefined,
   };
 }

enterprise/frontend/src/metabase-enterprise/transforms/pages/TransformListPage/CreateTransformMenu/CreateTransformMenu.tsx

@@ -85,6 +85,19 @@ export function CreateTransformMenu() {
               >
                 {t`SQL query`}
               </Menu.Item>
+              <Menu.Item
+                component={ForwardRefLink}
+                to={getNewTransformFromTypeUrl("python")}
+                leftSection={<Icon name="code_block" />}
+                onClick={() => {
+                  trackTransformCreate({
+                    triggeredFrom: "transform-page-create-menu",
+                    creationType: "python",
+                  });
+                }}
+              >
+                {t`Python script`}
+              </Menu.Item>
               <Menu.Item
                 leftSection={<Icon name="folder" />}
                 onClick={handleSavedQuestionClick}

enterprise/frontend/src/metabase-enterprise/transforms/pages/TransformPage/ManageSection/ManageSection.tsx

@@ -19,9 +19,12 @@ type ManageSectionProps = {
 };
 
 export function ManageSection({ transform }: ManageSectionProps) {
+  const sectionLabel =
+    transform.source.type === "python" ? t`Python Script` : t`Query`;
+
   return (
     <TitleSection
-      label={t`Query`}
+      label={sectionLabel}
       rightSection={
         <Group>
           <EditQueryButton transform={transform} />
@@ -30,7 +33,14 @@ export function ManageSection({ transform }: ManageSectionProps) {
       }
     >
       <Card p={0} shadow="none" withBorder>
-        <QueryView query={transform.source.query} />
+        {transform.source.type === "query" && (
+          <QueryView query={transform.source.query} />
+        )}
+        {transform.source.type === "python" && (
+          <div style={{ padding: "1rem" }}>
+            <pre>{transform.source.body}</pre>
+          </div>
+        )}
       </Card>
     </TitleSection>
   );
@@ -41,13 +51,16 @@ type EditQueryButtonProps = {
 };
 
 function EditQueryButton({ transform }: EditQueryButtonProps) {
+  const buttonText =
+    transform.source.type === "python" ? t`Edit script` : t`Edit query`;
+
   return (
     <Button
       component={Link}
       to={getTransformQueryUrl(transform.id)}
       leftSection={<Icon name="pencil_lines" aria-hidden />}
     >
-      {t`Edit query`}
+      {buttonText}
     </Button>
   );
 }

enterprise/frontend/src/metabase-enterprise/transforms/pages/TransformPage/SourceSection/SourceSection.tsx

@@ -0,0 +1,184 @@
+import { t } from "ttag";
+
+import { skipToken, useGetDatabaseQuery, useGetTableQuery } from "metabase/api";
+import Link from "metabase/common/components/Link";
+import CS from "metabase/css/core/index.css";
+import {
+  Divider,
+  Group,
+  Icon,
+  type IconName,
+  Loader,
+  Stack,
+  Text,
+} from "metabase/ui";
+import type { Transform } from "metabase-types/api";
+
+import { SplitSection } from "../../../components/SplitSection";
+import {
+  getBrowseDatabaseUrl,
+  getBrowseSchemaUrl,
+  getQueryBuilderUrl,
+} from "../../../urls";
+
+type SourceSectionProps = {
+  transform: Transform;
+};
+
+export function SourceSection({ transform }: SourceSectionProps) {
+  if (transform.source.type !== "python") {
+    return null;
+  }
+
+  return (
+    <SplitSection
+      label={t`Transform source`}
+      description={t`The data sources for this Python transform.`}
+    >
+      <Group p="lg">
+        <SourceInfo transform={transform} />
+      </Group>
+    </SplitSection>
+  );
+}
+
+type SourceInfoProps = {
+  transform: Transform;
+};
+
+function SourceInfo({ transform }: SourceInfoProps) {
+  if (transform.source.type !== "python") {
+    return null;
+  }
+
+  const sourceDatabaseId = transform.source["source-database"];
+  const sourceTables = transform.source["source-tables"];
+
+  const { data: database, isLoading: isDatabaseLoading } = useGetDatabaseQuery(
+    sourceDatabaseId ? { id: sourceDatabaseId } : skipToken,
+  );
+
+  const hasMultipleTables = sourceTables && Object.keys(sourceTables).length > 0;
+
+  if (isDatabaseLoading) {
+    return <Loader size="sm" />;
+  }
+
+  // Display for multiple tables
+  if (hasMultipleTables) {
+    return (
+      <Stack gap="sm">
+        {database && (
+          <Group gap="sm">
+            <SourceItemLink
+              label={database.name}
+              icon="database"
+              to={getBrowseDatabaseUrl(database.id)}
+              data-testid="source-database-link"
+            />
+          </Group>
+        )}
+        <Divider />
+        <Stack gap="xs">
+          {Object.entries(sourceTables).map(([alias, tableId]) => (
+            <TableWithAlias
+              key={alias}
+              alias={alias}
+              tableId={tableId}
+              databaseId={sourceDatabaseId}
+            />
+          ))}
+        </Stack>
+      </Stack>
+    );
+  }
+
+  // Just show database if no tables selected
+  return (
+    <Group gap="sm">
+      {database != null && (
+        <SourceItemLink
+          label={database.name}
+          icon="database"
+          to={getBrowseDatabaseUrl(database.id)}
+          data-testid="source-database-link"
+        />
+      )}
+    </Group>
+  );
+}
+
+type TableWithAliasProps = {
+  alias: string;
+  tableId: number;
+  databaseId: number;
+};
+
+function TableWithAlias({ alias, tableId, databaseId }: TableWithAliasProps) {
+  const { data: table, isLoading } = useGetTableQuery(
+    tableId ? { id: tableId } : skipToken,
+  );
+
+  if (isLoading) {
+    return <Loader size="xs" />;
+  }
+
+  return (
+    <Group gap="sm" align="flex-start">
+      <Group gap="xs">
+        {table?.schema && (
+          <>
+            <SourceItemLink
+              label={table.schema}
+              icon="folder"
+              to={getBrowseSchemaUrl(databaseId, table.schema)}
+              small
+            />
+            <SourceItemDivider />
+          </>
+        )}
+        <SourceItemLink
+          label={table?.display_name || table?.name || `Table ${tableId}`}
+          icon="table2"
+          to={table ? getQueryBuilderUrl(table.id, table.db_id) : undefined}
+          data-testid={`source-table-${alias}-link`}
+          small
+        />
+      </Group>
+    </Group>
+  );
+}
+
+type SourceItemLinkProps = {
+  label: string;
+  icon: IconName;
+  to?: string;
+  "data-testid"?: string;
+  small?: boolean;
+};
+
+function SourceItemLink({
+  label,
+  icon,
+  to,
+  "data-testid": dataTestId,
+  small = false,
+}: SourceItemLinkProps) {
+  return (
+    <Link
+      className={CS.link}
+      to={to ?? ""}
+      disabled={to == null}
+      data-testid={dataTestId}
+    >
+      <Group gap="xs">
+        <Icon name={icon} size={small ? 14 : 16} />
+        <Text c="inherit" size={small ? "sm" : "md"}>{label}</Text>
+      </Group>
+    </Link>
+  );
+}
+
+function SourceItemDivider() {
+  return <Icon name="chevronright" size={8} />;
+}

enterprise/frontend/src/metabase-enterprise/transforms/pages/TransformPage/SourceSection/index.ts

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

enterprise/frontend/src/metabase-enterprise/transforms/pages/TransformPage/TargetSection/TargetSection.tsx

@@ -54,7 +54,12 @@ type TargetInfoProps = {
 
 function TargetInfo({ transform }: TargetInfoProps) {
   const { source, target, table } = transform;
-  const { database: databaseId } = source.query;
+  const databaseId =
+    source.type === "query"
+      ? source.query.database
+      : source.type === "python"
+        ? source["source-database"]
+        : undefined;
   const { data, isLoading } = useGetDatabaseQuery(
     table == null && databaseId != null ? { id: databaseId } : skipToken,
   );

enterprise/frontend/src/metabase-enterprise/transforms/pages/TransformPage/TargetSection/UpdateTargetModal/UpdateTargetModal.tsx

@@ -78,7 +78,12 @@ function UpdateTargetForm({
   onClose,
 }: UpdateTargetFormProps) {
   const { source, target, table } = transform;
-  const { database: databaseId } = source.query;
+  const databaseId =
+    source.type === "query"
+      ? source.query.database
+      : source.type === "python"
+        ? source["source-database"]
+        : undefined;
   const [updateTransform] = useUpdateTransformMutation();
   const [deleteTransformTarget] = useDeleteTransformTargetMutation();
   const initialValues = useMemo(() => getInitialValues(transform), [transform]);

enterprise/frontend/src/metabase-enterprise/transforms/pages/TransformPage/TransformPage.tsx

@@ -14,6 +14,7 @@ import { HeaderSection } from "./HeaderSection";
 import { ManageSection } from "./ManageSection";
 import { NameSection } from "./NameSection";
 import { RunSection } from "./RunSection";
+import { SourceSection } from "./SourceSection";
 import { TargetSection } from "./TargetSection";
 
 type TransformPageParams = {
@@ -58,6 +59,7 @@ export function TransformPage({ params }: TransformPageProps) {
         <NameSection transform={transform} />
       </Stack>
       <RunSection transform={transform} />
+      <SourceSection transform={transform} />
       <TargetSection transform={transform} />
       <ManageSection transform={transform} />
     </Stack>

enterprise/frontend/src/metabase-enterprise/transforms/pages/TransformQueryPage/TransformQueryPage.tsx

@@ -10,7 +10,12 @@ import {
   useGetTransformQuery,
   useUpdateTransformMutation,
 } from "metabase-enterprise/api";
-import type { DatasetQuery, Transform } from "metabase-types/api";
+import { PythonTransformEditor } from "metabase-enterprise/transforms/components/PythonTransformEditor";
+import type {
+  DatasetQuery,
+  Transform,
+  TransformSource,
+} from "metabase-types/api";
 
 import { QueryEditor } from "../../components/QueryEditor";
 import { getTransformUrl } from "../../urls";
@@ -53,7 +58,7 @@ export function TransformQueryPageBody({
   const dispatch = useDispatch();
   const { sendErrorToast, sendSuccessToast } = useMetadataToasts();
 
-  const handleSave = async (query: DatasetQuery) => {
+  const handleDatasetSave = async (query: DatasetQuery) => {
     const { error } = await updateTransform({
       id: transform.id,
       source: {
@@ -70,16 +75,43 @@ export function TransformQueryPageBody({
     }
   };
 
+  const handlePythonSave = async (
+    source: TransformSource & { type: "python" },
+  ) => {
+    const { error } = await updateTransform({
+      id: transform.id,
+      source: source,
+    });
+
+    if (error) {
+      sendErrorToast(t`Failed to update transform query`);
+    } else {
+      sendSuccessToast(t`Transform query updated`);
+      dispatch(push(getTransformUrl(transform.id)));
+    }
+  };
+
   const handleCancel = () => {
     dispatch(push(getTransformUrl(transform.id)));
   };
 
+  if (transform.source.type === "python") {
+    return (
+      <PythonTransformEditor
+        initialSource={transform.source}
+        isNew={false}
+        isSaving={isLoading}
+        onSave={handlePythonSave}
+        onCancel={handleCancel}
+      />
+    );
+  }
   return (
     <QueryEditor
       initialQuery={transform.source.query}
       isNew={false}
       isSaving={isLoading}
-      onSave={handleSave}
+      onSave={handleDatasetSave}
       onCancel={handleCancel}
     />
   );

enterprise/frontend/src/metabase-enterprise/transforms/urls.ts

@@ -15,7 +15,9 @@ export function getTransformListUrl() {
   return ROOT_URL;
 }
 
-export function getNewTransformFromTypeUrl(type: DatasetQuery["type"]) {
+export function getNewTransformFromTypeUrl(
+  type: DatasetQuery["type"] | "python",
+) {
   return `${ROOT_URL}/new/${type}`;
 }
 

example_db_transform.py

@@ -0,0 +1,33 @@
+import pandas as pd
+
+def transform(db):
+    """
+    Example transform function demonstrating database connectivity.
+    
+    The db parameter provides:
+    - db.read_table(table_name) - reads entire table
+    - db.read_table(table_name, schema) - reads table from specific schema
+    
+    This function would typically:
+    1. Read one or more tables using db.read_table()
+    2. Perform transformations, joins, aggregations, etc.
+    3. Return the final result as a pandas DataFrame
+    """
+    
+    # Example: Read data from multiple tables
+    # customers = db.read_table("customers")
+    # orders = db.read_table("orders") 
+    # products = db.read_table("products", schema="inventory")
+    
+    # Example: Perform transformations and joins
+    # result = customers.merge(orders, on="customer_id")
+    # result = result.merge(products, on="product_id")
+    
+    # For demonstration purposes, return a sample DataFrame
+    # In real usage, this would be the result of database operations
+    data = {
+        'customer_name': ['John Doe', 'Jane Smith', 'Bob Johnson'],
+        'total_orders': [5, 12, 3],
+        'total_spent': [299.99, 1450.50, 89.95]
+    }
+    return pd.DataFrame(data)

frontend/src/metabase-types/analytics/event.ts

@@ -272,7 +272,7 @@ export type TransformJobTriggerManualRunEvent = ValidateEvent<{
 export type TransformCreateEvent = ValidateEvent<{
   event: "transform_create";
   triggered_from: "transform-page-create-menu";
-  event_detail: "query" | "native" | "saved-question";
+  event_detail: "query" | "native" | "python" | "saved-question";
 }>;
 
 export type TransformCreatedEvent = ValidateEvent<{

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

@@ -9,13 +9,18 @@ import type {
 
 import { createMockStructuredDatasetQuery } from "./query";
 
-export function createMockTransformSource(
-  opts?: Partial<TransformSource>,
-): TransformSource {
+export function createMockTransformSource(): TransformSource {
   return {
     type: "query",
     query: createMockStructuredDatasetQuery(),
-    ...opts,
+  };
+}
+
+export function createMockPythonTransformSource(): TransformSource {
+  return {
+    type: "python",
+    body: "# Python script\nprint('Hello, world!')",
+    "source-database": 1,
   };
 }
 

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

@@ -22,10 +22,17 @@ export type Transform = {
   last_run?: TransformRun | null;
 };
 
-export type TransformSource = {
-  type: "query";
-  query: DatasetQuery;
-};
+export type TransformSource =
+  | {
+      type: "query";
+      query: DatasetQuery;
+    }
+  | {
+      type: "python";
+      body: string;
+      "source-database": number;
+      "source-tables": Record<string, number>; // alias -> table-id mapping
+    };
 
 export type TransformTargetType = "table";
 

frontend/src/metabase/admin/settings/components/SettingsNav/SettingsNav.tsx

@@ -24,6 +24,7 @@ export function SettingsNav() {
   const hasSaml = useHasTokenFeature("sso_saml");
   const hasJwt = useHasTokenFeature("sso_jwt");
   const hasScim = useHasTokenFeature("scim");
+  const hasTransforms = useHasTokenFeature("transforms");
 
   return (
     <AdminNavWrapper>

python-runner/Dockerfile

@@ -0,0 +1,25 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install Flask and dependencies
+RUN pip install --no-cache-dir \
+    flask==2.3.3 \
+    pandas==2.1.4 \
+    requests==2.31.0
+
+# Create non-root user for security
+RUN groupadd -r pyexec && \
+    useradd -r -g pyexec pyexec && \
+    mkdir -p /tmp/executions && \
+    chown -R pyexec:pyexec /app /tmp/executions
+
+COPY server.py transform_runner.py ./
+RUN chown -R pyexec:pyexec /app
+#USER pyexec
+
+ENV PYTHONUNBUFFERED=1
+
+EXPOSE 5000
+
+CMD ["python", "server.py"]

python-runner/INTEGRATION.md

@@ -0,0 +1,100 @@
+# Python Execution Server - Transform Runner Integration
+
+## Overview
+
+The Python execution server uses the `transform_runner.py` framework, which provides a standardized way for users to write data transformation functions that integrate with Metabase.
+
+## Server Features (`server.py`)
+- Uses `transform_runner.py` for executing user code
+- Memory limit of 1GB to support pandas/numpy operations
+- Supports table data passed as file paths via table_mapping parameter
+- Returns CSV output file path when transform succeeds
+
+### Transform Runner Framework
+- Users write a `transform()` function that returns a pandas DataFrame
+- The framework handles:
+  - CSV output generation
+  - Error handling and logging
+  - Metabase API integration for reading tables
+  - Parameter injection for table data
+
+### API Specification
+
+The `/execute` endpoint accepts the following parameters:
+```json
+{
+  "code": "Python code with transform() function",
+  "working_dir": "/path/to/work/dir",
+  "timeout": 30,
+  "table_mapping": {                       // Optional
+    "orders": /path/to/orders.jsonl
+    "customers": /path/to/customers.jsonl
+  }
+}
+```
+
+Response includes the output CSV path:
+```json
+{
+  "exit_code": 0,
+  "execution_time": 1.23,
+  "stdout_file": "/path/stdout.log",
+  "stderr_file": "/path/stderr.log",
+  "output_file": "/path/output.csv",  // Added
+  "request_id": "uuid",
+  "timeout": false
+}
+```
+
+## Writing Transform Functions
+
+### Basic Transform
+```python
+import pandas as pd
+
+def transform():
+    df = pd.DataFrame({
+        "column1": [1, 2, 3],
+        "column2": ["a", "b", "c"]
+    })
+    return df
+```
+
+### Transform with Table Parameters
+```python
+import pandas as pd
+
+def transform(orders, customers):
+    """Receives DataFrames from Metabase tables."""
+    merged = pd.merge(orders, customers, on="customer_id")
+    return merged
+```
+
+## Testing
+
+All test fixtures have been updated to use the transform() function pattern:
+- `basic_hello.py` - Basic transform test
+- `error_test.py` - Error handling
+- `timeout_test.py` - Timeout handling
+- `pandas_test.py` - Pandas operations
+- `table_params_test.py` - Table parameter handling
+
+Run tests with:
+```bash
+./run_tests.sh
+```
+
+## Resource Limits
+
+- Memory: 1024MB (increased for pandas/numpy)
+- CPU Time: 60 seconds
+- File Descriptors: 256
+- Processes/Threads: 50
+
+## Benefits
+
+1. **Standardized Interface**: All Python transforms follow the same pattern
+2. **Metabase Integration**: Built-in support for reading Metabase tables
+3. **Automatic CSV Generation**: Transform results automatically saved as CSV
+4. **Error Handling**: Consistent error reporting and logging
+5. **Resource Management**: Proper limits prevent runaway processes

python-runner/Makefile

@@ -0,0 +1,90 @@
+.PHONY: build run stop test clean rebuild logs status help
+
+# Docker image and container names
+IMAGE_NAME = python-exec-server
+CONTAINER_NAME = python-exec-test
+PORT = 5001
+
+# Mount path for shared directory
+MOUNT_PATH = /tmp/python-exec-work
+
+# Default target
+build:
+	docker build -t $(IMAGE_NAME) .
+
+# Run the server
+run: build stop
+	@mkdir -p $(MOUNT_PATH)
+	docker run -d -p $(PORT):5000 --name $(CONTAINER_NAME) \
+		-v $(MOUNT_PATH):$(MOUNT_PATH) \
+		--add-host host.docker.internal:host-gateway \
+		$(IMAGE_NAME)
+	@echo "Server starting on http://localhost:$(PORT)"
+	@echo "Mount path: $(MOUNT_PATH)"
+	@echo "Network: Container can reach host at host.docker.internal"
+	@echo "Waiting for server to be ready..."
+	@sleep 2
+	@curl -s http://localhost:$(PORT)/status > /dev/null && \
+		echo "✅ Server is ready!" && \
+		echo "" && \
+		echo "For Metabase integration, configure:" && \
+		echo "  python-execution-server-url: http://localhost:$(PORT)" && \
+		echo "  python-execution-mount-path: $(MOUNT_PATH)" && \
+		echo "  python-runner-base-url: http://host.docker.internal:3000" || \
+		echo "⚠️ Server may still be starting..."
+
+# Stop and remove the container
+stop:
+	@docker stop $(CONTAINER_NAME) 2>/dev/null || true
+	@docker rm $(CONTAINER_NAME) 2>/dev/null || true
+
+# Run the test suite
+test: build
+	./run_tests.sh
+
+# Quick smoke test without full test suite
+smoke-test: run
+	@echo "Running smoke test..."
+	@mkdir -p /tmp/python-exec-tests/smoke
+	@curl -s -X POST http://localhost:$(PORT)/execute \
+		-H "Content-Type: application/json" \
+		-d '{"code":"import pandas as pd\ndef transform():\n    return pd.DataFrame({\"test\":[\"OK\"]})", "working_dir":"/tmp/python-exec-tests/smoke", "timeout":5}' \
+		| jq -r '.exit_code' | grep -q "^0$$" && echo "✅ Smoke test passed!" || echo "❌ Smoke test failed!"
+
+# View server logs
+logs:
+	docker logs -f $(CONTAINER_NAME)
+
+# Check server status
+status:
+	@curl -s http://localhost:$(PORT)/status | jq . || echo "Server not running"
+
+# Remove the Docker image
+clean: stop
+	docker rmi $(IMAGE_NAME) 2>/dev/null || true
+	rm -rf venv 2>/dev/null || true
+	rm -rf $(MOUNT_PATH) 2>/dev/null || true
+
+# Rebuild from scratch (no cache)
+rebuild: clean
+	docker build --no-cache -t $(IMAGE_NAME) .
+
+# Setup development environment
+dev-setup:
+	python3 -m venv venv
+	./venv/bin/pip install requests
+
+# Show help
+help:
+	@echo "Available targets:"
+	@echo "  build      - Build the Docker image"
+	@echo "  run        - Run the server container"
+	@echo "  stop       - Stop and remove the container"
+	@echo "  test       - Run the full test suite"
+	@echo "  smoke-test - Quick test to verify server works"
+	@echo "  logs       - View server logs"
+	@echo "  status     - Check server status and metrics"
+	@echo "  clean      - Remove container, image, and temp files"
+	@echo "  rebuild    - Clean and rebuild without cache"
+	@echo "  dev-setup  - Setup Python virtual environment"
+	@echo "  help       - Show this help message"

python-runner/README.md

@@ -0,0 +1,141 @@
+# Python Execution Server
+
+A containerized Python execution service for Metabase that safely runs user-defined data transformation functions.
+
+## Architecture
+
+The server uses a transform-based architecture where users write a `transform()` function that:
+- Receives table data as parameters (when connected to Metabase)
+- Processes the data using pandas
+- Returns a DataFrame that's automatically saved as CSV
+
+## Directory Structure
+
+```
+python-runner/
+├── server.py            # Flask web server with execution queue
+├── transform_runner.py  # Transform framework that wraps user code
+├── Dockerfile          # Container definition
+├── Makefile           # Development and deployment commands
+├── run_tests.sh       # Test suite runner
+├── tests/             # Test suite and fixtures
+│   ├── test_server.py # Comprehensive test suite
+│   ├── fixtures/      # Test transform functions
+│   └── README.md      # Test documentation
+├── INTEGRATION.md     # Integration details
+└── venv/             # Python virtual environment (git-ignored)
+```
+
+## Quick Start
+
+```bash
+# Build and run the server
+make run
+
+# Check server status
+make status
+
+# Run tests
+make test
+
+# View logs
+make logs
+
+# Stop server
+make stop
+```
+
+## API Usage
+
+### Execute Transform
+
+```bash
+curl -X POST http://localhost:5001/execute \
+  -H "Content-Type: application/json" \
+  -d '{
+    "code": "import pandas as pd\ndef transform():\n    return pd.DataFrame({\"result\": [1, 2, 3]})",
+    "working_dir": "/tmp/work",
+    "timeout": 30
+  }'
+```
+
+### Check Status
+
+```bash
+curl http://localhost:5001/status
+```
+
+## Writing Transforms
+
+User code must define a `transform()` function:
+
+```python
+import pandas as pd
+
+def transform():
+    # Process data
+    df = pd.DataFrame({
+        "column1": [1, 2, 3],
+        "column2": ["a", "b", "c"]
+    })
+
+    # Return DataFrame (will be saved as CSV)
+    return df
+```
+
+With Metabase table parameters:
+
+```python
+def transform(orders, customers):
+    # Receive DataFrames from Metabase
+    merged = pd.merge(orders, customers, on="customer_id")
+    return merged
+```
+
+## Features
+
+- **Sequential Processing**: Single-worker queue ensures no concurrent execution
+- **Resource Limits**: Memory (1GB), CPU (60s), processes (50)
+- **Error Handling**: Proper error capture and timeout management
+- **File Isolation**: Each execution gets its own working directory
+- **Metrics Tracking**: Request counts, execution times, success rates
+- **Transform Framework**: Standard interface for all Python transformations
+
+## Development
+
+```bash
+# Setup development environment
+make dev-setup
+
+# Run smoke test
+make smoke-test
+
+# Rebuild without cache
+make rebuild
+
+# Clean everything
+make clean
+```
+
+## Testing
+
+The test suite includes 8 comprehensive tests:
+- Server health check
+- Basic transform execution
+- Error handling
+- Timeout enforcement
+- JSON processing
+- Transform with parameters
+- Metrics tracking
+- Concurrent request queuing
+
+Run tests with: `make test`
+
+## Security
+
+- Runs as non-root user (pyexec)
+- Resource limits enforced via setrlimit
+- Process isolation with separate process groups
+- Memory limited to 1GB
+- CPU time limited to 60 seconds
+- File descriptors limited to 256

python-runner/SETUP.md

@@ -0,0 +1,105 @@
+# Python Runner Setup Guide
+
+This guide walks through setting up the Python execution service for use with Metabase transforms.
+
+## Prerequisites
+
+- Docker installed and running
+- Metabase instance running (typically on port 3000)
+- Admin access to Metabase settings
+
+## Step 1: Start the Python Execution Service
+
+From the `python-runner` directory:
+
+```bash
+# Build and start the service
+make run
+```
+
+This will:
+- Build the Docker image
+- Start the container on port 5001
+- Create a mount directory at `/tmp/python-exec-work`
+- Display the configuration values you'll need
+
+## Step 2: Configure Metabase Settings
+
+In Metabase, these settings need to be configured (they're internal settings, not visible in the UI):
+
+### Required Settings
+
+1. **python-execution-server-url**: URL where the Python service is running
+   - Default: `http://localhost:5001`
+   - Update if running on a different host/port
+
+2. **python-execution-mount-path**: Shared directory path between Metabase and Python service
+   - Default: `/tmp/python-exec-work`
+   - Must be accessible by both Metabase and the Python container
+
+## Step 3: Verify Setup
+
+### Check Python Service Status
+
+```bash
+make status
+```
+
+This shows the service metrics and confirms it's running.
+
+### Run a Quick Test
+
+```bash
+make smoke-test
+```
+
+This executes a simple transform to verify the service works.
+
+## Using Python Transforms in Metabase
+
+Once configured, you can:
+
+1. Create a new transform in Metabase
+2. Select "Python" as the transform type
+3. Write your transform function:
+
+```python
+import pandas as pd
+
+def transform(table1, table2):
+    # Your data transformation logic
+    result = pd.merge(table1, table2, on='common_column')
+    return result
+```
+
+4. The transform will receive Metabase tables as pandas DataFrames
+5. Return a DataFrame that will become the output table
+
+## Troubleshooting
+
+### View Service Logs
+
+```bash
+make logs
+```
+
+### Common Issues
+
+1. **Connection refused**: Ensure the Python service is running (`make status`)
+2. **Mount path errors**: Verify `/tmp/python-exec-work` exists and is writable
+3. **Transform timeouts**: Check the `transform-timeout` setting if jobs are timing out
+
+## Development Commands
+
+- `make test` - Run the full test suite
+- `make rebuild` - Rebuild the image from scratch
+- `make clean` - Remove all containers and images
+- `make help` - Show all available commands
+
+## Security Notes
+
+The Python execution service runs with resource limits:
+- Memory: 1GB maximum
+- CPU time: 60 seconds per execution
+- Process isolation in separate containers
+- File system access limited to working directory

python-runner/run_tests.sh

@@ -0,0 +1,61 @@
+#!/bin/bash
+set -e
+
+SERVER_URL="http://localhost:5001"
+CONTAINER_NAME="python-exec-test"
+DOCKER_IMAGE="python-exec-server"
+# Use same mount path as Makefile for consistency
+TEST_DIR="/tmp/python-exec-work"
+
+echo "Python Execution Server Test Runner"
+echo "===================================="
+
+# Create test directory on host
+echo "📁 Creating test directory: $TEST_DIR"
+mkdir -p "$TEST_DIR"
+
+# Check if server is already running
+if curl -s "$SERVER_URL/status" > /dev/null 2>&1; then
+    echo "⚠️  Stopping existing server..."
+    docker stop "$CONTAINER_NAME" 2>/dev/null || true
+    docker rm "$CONTAINER_NAME" 2>/dev/null || true
+fi
+
+echo "🚀 Starting Python execution server with mounted test directory..."
+
+# Stop and remove existing container if it exists
+docker stop "$CONTAINER_NAME" 2>/dev/null || true
+docker rm "$CONTAINER_NAME" 2>/dev/null || true
+
+# Start new container with mounted volume
+docker run -d -p 5001:5000 --name "$CONTAINER_NAME" \
+    -v "$TEST_DIR:$TEST_DIR" \
+    "$DOCKER_IMAGE"
+    
+# Wait for server to be ready
+echo "⏳ Waiting for server to start..."
+for i in {1..30}; do
+    if curl -s "$SERVER_URL/status" > /dev/null 2>&1; then
+        echo "✅ Server is ready!"
+        break
+    fi
+    if [ $i -eq 30 ]; then
+        echo "❌ Server failed to start within 30 seconds"
+        docker logs "$CONTAINER_NAME"
+        exit 1
+    fi
+    sleep 1
+done
+
+# Run the tests
+echo "🧪 Running test suite..."
+SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )"
+source "$SCRIPT_DIR/venv/bin/activate"
+TEST_DIR="$TEST_DIR" python "$SCRIPT_DIR/tests/test_server.py" "$SERVER_URL"
+
+echo ""
+echo "🏁 Test run complete!"
+
+# Cleanup
+echo "🧹 Cleaning up..."
+rm -rf "$TEST_DIR"

python-runner/server.py

@@ -0,0 +1,313 @@
+#!/usr/bin/env python3
+"""
+Python execution server for Metabase.
+Runs Python scripts in isolated processes with resource limits and monitoring.
+"""
+
+import os
+import sys
+import time
+import signal
+import resource
+import threading
+import subprocess
+from pathlib import Path
+from queue import Queue, Empty
+from datetime import datetime, timedelta
+from typing import Optional, Dict, Any
+
+from flask import Flask, request, jsonify
+from werkzeug.exceptions import HTTPException
+
+DEFAULT_TIMEOUT = 30  # seconds
+MAX_MEMORY_MB = 1024  # Increased for pandas/numpy
+MAX_CPU_TIME_SECONDS = 60
+MAX_QUEUE_SIZE = 10
+QUEUE_TIMEOUT = 60  # seconds to wait in queue
+MAX_FILE_DESCRIPTORS = 256
+MAX_PROCESSES = 50
+
+app = Flask(__name__)
+
+metrics = {
+    "total_requests": 0,
+    "successful_executions": 0,
+    "failed_executions": 0,
+    "timeouts": 0,
+    "rejected_requests": 0,
+    "current_queue_size": 0,
+    "average_execution_time": 0,
+    "execution_times": [],
+}
+metrics_lock = threading.Lock()
+
+request_queue = Queue(maxsize=MAX_QUEUE_SIZE)
+current_execution_id = None
+execution_id_lock = threading.Lock()
+
+
+class ExecutionRequest:
+    def __init__(self, code: str, working_dir: str, timeout: int, request_id: str,
+                 table_mapping: Optional[Dict[str, str]] = None):
+        self.code = code
+        self.working_dir = working_dir
+        self.timeout = timeout
+        self.request_id = request_id
+        self.table_mapping = table_mapping
+        self.result_event = threading.Event()
+        self.result = None
+        self.queued_at = time.time()
+
+
+def set_resource_limits():
+    memory_bytes = MAX_MEMORY_MB * 1024 * 1024
+    resource.setrlimit(resource.RLIMIT_AS, (memory_bytes, memory_bytes))
+    resource.setrlimit(resource.RLIMIT_CPU, (MAX_CPU_TIME_SECONDS, MAX_CPU_TIME_SECONDS))
+    resource.setrlimit(resource.RLIMIT_NOFILE, (MAX_FILE_DESCRIPTORS, MAX_FILE_DESCRIPTORS))
+    resource.setrlimit(resource.RLIMIT_NPROC, (MAX_PROCESSES, MAX_PROCESSES))
+
+
+def execute_code(code: str, working_dir: str, timeout: int,
+                 table_mapping: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
+    start_time = time.time()
+
+    work_path = Path(working_dir)
+    if not work_path.exists():
+        return {
+            "exit_code": -1,
+            "execution_time": 0,
+            "error": f"Working directory does not exist: {working_dir}"
+        }
+
+    stdout_file = work_path / "stdout.log"
+    stderr_file = work_path / "stderr.log"
+    script_file = work_path / "script.py"
+    output_file = work_path / "output.csv"
+
+    script_file.write_text(code)
+
+    env = os.environ.copy()
+    env["PYTHONUNBUFFERED"] = "1"
+    env["OUTPUT_FILE"] = str(output_file)
+
+    if table_mapping:
+        import json
+        env["TABLE_FILE_MAPPING"] = json.dumps(table_mapping)
+
+    # Use transform_runner.py if it exists, otherwise run script directly
+    runner_path = Path(__file__).parent / "transform_runner.py"
+    if runner_path.exists():
+        cmd = [sys.executable, str(runner_path)]
+    else:
+        cmd = [sys.executable, str(script_file)]
+
+    try:
+        with open(stdout_file, "w") as stdout, open(stderr_file, "w") as stderr:
+            process = subprocess.Popen(
+                cmd,
+                stdout=stdout,
+                stderr=stderr,
+                cwd=str(work_path),
+                env=env,
+                preexec_fn=set_resource_limits,
+                start_new_session=True
+            )
+
+            try:
+                exit_code = process.wait(timeout=timeout)
+                execution_time = time.time() - start_time
+
+                result = {
+                    "exit_code": exit_code,
+                    "execution_time": execution_time,
+                    "timeout": False,
+                    "stdout_file": str(stdout_file),
+                    "stderr_file": str(stderr_file),
+                }
+
+                # Add output file if it was created
+                if output_file.exists():
+                    result["output_file"] = str(output_file)
+
+                return result
+
+            except subprocess.TimeoutExpired:
+                os.killpg(os.getpgid(process.pid), signal.SIGKILL)
+                process.wait()
+
+                with metrics_lock:
+                    metrics["timeouts"] += 1
+
+                return {
+                    "exit_code": -1,
+                    "execution_time": timeout,
+                    "timeout": True,
+                    "error": f"Script execution timed out after {timeout} seconds",
+                    "stdout_file": str(stdout_file),
+                    "stderr_file": str(stderr_file),
+                }
+
+    except Exception as e:
+        app.logger.error(f"Error executing script: {e}")
+        return {
+            "exit_code": -1,
+            "execution_time": time.time() - start_time,
+            "timeout": False,
+            "error": str(e),
+        }
+
+
+def worker_thread():
+    global current_execution_id
+
+    while True:
+        try:
+            req = request_queue.get(timeout=1)
+
+            wait_time = time.time() - req.queued_at
+            if wait_time > QUEUE_TIMEOUT:
+                req.result = {
+                    "error": f"Request timed out in queue after {wait_time:.1f} seconds",
+                    "exit_code": -1,
+                }
+                req.result_event.set()
+                continue
+
+            with execution_id_lock:
+                current_execution_id = req.request_id
+
+            with metrics_lock:
+                metrics["current_queue_size"] = request_queue.qsize()
+
+            try:
+                result = execute_code(
+                    req.code,
+                    req.working_dir,
+                    req.timeout,
+                    req.table_mapping
+                )
+                result["request_id"] = req.request_id
+
+                with metrics_lock:
+                    if result.get("exit_code") == 0:
+                        metrics["successful_executions"] += 1
+                    else:
+                        metrics["failed_executions"] += 1
+
+                    if not result.get("timeout"):
+                        metrics["execution_times"].append(result["execution_time"])
+                        if len(metrics["execution_times"]) > 100:
+                            metrics["execution_times"] = metrics["execution_times"][-100:]
+                        if metrics["execution_times"]:
+                            metrics["average_execution_time"] = sum(metrics["execution_times"]) / len(metrics["execution_times"])
+
+                req.result = result
+
+            except Exception as e:
+                app.logger.error(f"Worker error: {e}")
+                req.result = {
+                    "error": str(e),
+                    "exit_code": -1,
+                    "request_id": req.request_id,
+                }
+
+            finally:
+                with execution_id_lock:
+                    current_execution_id = None
+
+                req.result_event.set()
+
+        except Empty:
+            continue
+        except Exception as e:
+            app.logger.error(f"Worker thread error: {e}")
+
+
+@app.route("/execute", methods=["POST"])
+def execute():
+    with metrics_lock:
+        metrics["total_requests"] += 1
+
+    data = request.get_json()
+    if not data or "code" not in data or "working_dir" not in data:
+        return jsonify({"error": "code and working_dir are required"}), 400
+
+    code = data["code"]
+    working_dir = data["working_dir"]
+    timeout = data.get("timeout", DEFAULT_TIMEOUT)
+    table_mapping = data.get("table_mapping")
+
+    import uuid
+    request_id = str(uuid.uuid4())
+
+    req = ExecutionRequest(code, working_dir, timeout, request_id,
+                           table_mapping)
+
+    try:
+        request_queue.put_nowait(req)
+        with metrics_lock:
+            metrics["current_queue_size"] = request_queue.qsize()
+    except:
+        with metrics_lock:
+            metrics["rejected_requests"] += 1
+        return jsonify({
+            "error": "Server is at capacity, please retry later",
+            "queue_size": request_queue.qsize(),
+        }), 503
+
+    if not req.result_event.wait(timeout=timeout + QUEUE_TIMEOUT):
+        return jsonify({
+            "error": "Request processing timeout",
+            "request_id": request_id,
+        }), 504
+
+    result = req.result
+    status_code = 200 if result.get("exit_code") == 0 else 500
+
+    return jsonify(result), status_code
+
+
+@app.route("/status", methods=["GET"])
+def status():
+    with execution_id_lock:
+        is_executing = current_execution_id is not None
+        exec_id = current_execution_id
+
+    with metrics_lock:
+        success_count = metrics["successful_executions"]
+        fail_count = metrics["failed_executions"]
+        total_completed = success_count + fail_count
+
+        status_data = {
+            "healthy": True,
+            "queue_size": request_queue.qsize(),
+            "max_queue_size": MAX_QUEUE_SIZE,
+            "currently_executing": is_executing,
+            "current_execution_id": exec_id,
+            "metrics": {
+                "total_requests": metrics["total_requests"],
+                "completed": total_completed,
+                "success_rate": success_count / max(1, total_completed),
+                "timeouts": metrics["timeouts"],
+                "rejected": metrics["rejected_requests"],
+                "avg_execution_time": round(metrics["average_execution_time"], 2),
+            }
+        }
+
+    return jsonify(status_data)
+
+
+@app.errorhandler(Exception)
+def handle_error(e):
+    if isinstance(e, HTTPException):
+        return jsonify({"error": e.description}), e.code
+
+    app.logger.error(f"Unhandled error: {e}")
+    return jsonify({"error": "Internal server error"}), 500
+
+
+if __name__ == "__main__":
+    worker = threading.Thread(target=worker_thread, daemon=True)
+    worker.start()
+
+    app.run(host="0.0.0.0", port=5000, threaded=True)

python-runner/tests/README.md

@@ -0,0 +1,44 @@
+# Python Execution Server Tests
+
+## Structure
+
+- `fixtures/` - Python test scripts used as test inputs
+  - `basic_hello.py` - Basic execution test
+  - `error_test.py` - Error handling test  
+  - `timeout_test.py` - Timeout handling test
+  - `long_task.py` - Long-running task for concurrency testing
+  - `quick_task.py` - Quick task for queue testing
+  - `json_test.py` - JSON processing test
+  - `pandas_test.py` - Pandas integration test
+
+- `test_server.py` - Main test suite with comprehensive tests
+
+## Running Tests
+
+```bash
+# From python-runner directory:
+./run_tests.sh
+```
+
+The test runner will:
+1. Create a shared test directory at `/tmp/python-exec-tests`
+2. Start the Docker container with the test directory mounted
+3. Run all tests using the Python virtual environment
+4. Clean up test files
+
+## Test Coverage
+
+- ✅ Server health check
+- ✅ Basic code execution
+- ✅ Error handling with proper exit codes
+- ✅ Timeout enforcement
+- ✅ Concurrent request queuing
+- ✅ JSON processing
+- ✅ Metrics tracking
+- ✅ File I/O in mounted directories
+
+## Requirements
+
+- Docker
+- Python 3 with venv
+- `requests` library (installed in venv)

python-runner/tests/fixtures/basic_hello.py

@@ -0,0 +1,13 @@
+import pandas as pd
+
+def transform():
+    """Basic transform function that returns a simple DataFrame."""
+    print("Hello from transform function!")
+    
+    # Create a simple DataFrame
+    df = pd.DataFrame({
+        "message": ["Test successful!"],
+        "status": ["OK"]
+    })
+    
+    return df

python-runner/tests/fixtures/error_test.py

@@ -0,0 +1,6 @@
+import pandas as pd
+
+def transform():
+    """Transform function that raises an error."""
+    print("About to cause an error")
+    raise ValueError("This is a test error")

python-runner/tests/fixtures/json_test.py

@@ -0,0 +1,11 @@
+import pandas as pd
+import json
+
+def transform():
+    """Transform function that tests JSON handling."""
+    print("Testing JSON processing in transform")
+    
+    data = {"test": "successful", "count": 42}
+    print(json.dumps(data))
+    
+    return pd.DataFrame([data])

python-runner/tests/fixtures/long_task.py

@@ -0,0 +1,13 @@
+import time
+import pandas as pd
+
+def transform():
+    """Long-running transform function for concurrency testing."""
+    print("Long running task started")
+    time.sleep(3)
+    print("Long running task completed")
+    
+    return pd.DataFrame({
+        "result": ["Task completed"],
+        "duration": [3]
+    })

python-runner/tests/fixtures/metrics_test.py

@@ -0,0 +1,5 @@
+import pandas as pd
+
+def transform():
+    print('metrics test')
+    return pd.DataFrame({"metric": [1]})

python-runner/tests/fixtures/pandas_test.py

@@ -0,0 +1,16 @@
+import pandas as pd
+
+def transform():
+    """Transform function with pandas operations."""
+    print("Testing pandas integration")
+    
+    df = pd.DataFrame({
+        "name": ["Alice", "Bob", "Charlie"],
+        "age": [25, 30, 35],
+        "city": ["NYC", "SF", "LA"]
+    })
+    
+    print(df.to_string())
+    print(f"DataFrame has {len(df)} rows")
+    
+    return df

python-runner/tests/fixtures/quick_task.py

@@ -0,0 +1,9 @@
+import pandas as pd
+
+def transform():
+    """Quick transform function for queue testing."""
+    print("This should be queued")
+    
+    return pd.DataFrame({
+        "message": ["Quick task executed"]
+    })

python-runner/tests/fixtures/table_params_test.py

@@ -0,0 +1,13 @@
+import pandas as pd
+
+def transform(orders, customers):
+    """Transform function that receives table parameters."""
+    print(f"Received orders table with {len(orders)} rows")
+    print(f"Received customers table with {len(customers)} rows")
+    
+    # Merge the tables
+    result = pd.merge(orders, customers, on="customer_id", how="inner")
+    
+    print(f"Merged result has {len(result)} rows")
+    
+    return result

python-runner/tests/fixtures/timeout_test.py

@@ -0,0 +1,9 @@
+import time
+import pandas as pd
+
+def transform():
+    """Transform function that times out."""
+    print("Starting long operation...")
+    time.sleep(5)
+    print("This should not appear")
+    return pd.DataFrame()

python-runner/tests/fixtures/transform_mock.py

@@ -0,0 +1,15 @@
+import pandas as pd
+
+def transform():
+    # Create mock data that would normally come from Metabase
+    print("Transform with simulated table data")
+    
+    df = pd.DataFrame({
+        "id": [1, 2, 3],
+        "value": [10, 20, 30]
+    })
+    
+    # Process the data
+    df["doubled"] = df["value"] * 2
+    
+    return df

python-runner/tests/test_server.py

@@ -0,0 +1,333 @@
+#!/usr/bin/env python3
+"""
+Test suite for Python execution server.
+"""
+
+import os
+import json
+import time
+import tempfile
+import threading
+import subprocess
+from pathlib import Path
+from typing import Dict, Any
+
+import requests
+
+
+class PythonServerTester:
+    def __init__(self, server_url: str = "http://localhost:5001"):
+        self.server_url = server_url.rstrip("/")
+        self.session = requests.Session()
+        self.test_dir = None
+        self.test_subdir = None
+        
+    def setup(self):
+        """Setup test environment."""
+        # Use the mounted test directory from environment variable
+        self.test_dir = os.environ.get("TEST_DIR", "/tmp/python-exec-tests")
+        # Create a unique subdirectory for this test run
+        import uuid
+        self.test_subdir = os.path.join(self.test_dir, str(uuid.uuid4())[:8])
+        os.makedirs(self.test_subdir, exist_ok=True)
+        print(f"Test directory: {self.test_subdir}")
+        
+    def teardown(self):
+        """Cleanup test environment."""
+        # Clean up the test subdirectory
+        if self.test_subdir and Path(self.test_subdir).exists():
+            import shutil
+            shutil.rmtree(self.test_subdir)
+    
+    def load_fixture(self, fixture_name: str) -> str:
+        """Load a test fixture file."""
+        fixture_path = Path(__file__).parent / "fixtures" / f"{fixture_name}.py"
+        return fixture_path.read_text()
+    
+    def execute_code(self, code: str, timeout: int = 10) -> Dict[str, Any]:
+        """Execute code on the server."""
+        payload = {
+            "code": code,
+            "working_dir": self.test_subdir,
+            "timeout": timeout
+        }
+        
+        try:
+            response = self.session.post(
+                f"{self.server_url}/execute",
+                json=payload,
+                timeout=timeout + 5
+            )
+            return response.json()
+        except Exception as e:
+            return {"error": f"Request failed: {e}", "exit_code": -1}
+    
+    def get_status(self) -> Dict[str, Any]:
+        """Get server status."""
+        try:
+            response = self.session.get(f"{self.server_url}/status")
+            return response.json()
+        except Exception as e:
+            return {"error": f"Status request failed: {e}"}
+    
+    def read_file(self, filename: str) -> str:
+        """Read a file from the test directory."""
+        try:
+            return (Path(self.test_subdir) / filename).read_text()
+        except Exception as e:
+            return f"Error reading {filename}: {e}"
+    
+    def test_server_health(self) -> bool:
+        """Test server health check."""
+        print("Testing server health...")
+        status = self.get_status()
+        
+        if "error" in status:
+            print(f"❌ Health check failed: {status['error']}")
+            return False
+            
+        if not status.get("healthy"):
+            print(f"❌ Server not healthy: {status}")
+            return False
+            
+        print("✅ Server health check passed")
+        return True
+    
+    def test_basic_execution(self) -> bool:
+        """Test basic code execution."""
+        print("Testing basic code execution...")
+        
+        code = self.load_fixture("basic_hello")
+        result = self.execute_code(code)
+        
+        if result.get("exit_code") != 0:
+            print(f"❌ Basic execution failed: {result}")
+            return False
+        
+        # Check if CSV output was created
+        if "output_file" not in result:
+            print(f"❌ No output CSV file created: {result}")
+            return False
+        
+        # Read and verify CSV output
+        csv_content = self.read_file("output.csv")
+        if "Test successful!" not in csv_content:
+            print(f"❌ CSV content incorrect: {csv_content}")
+            return False
+            
+        print("✅ Basic execution test passed")
+        return True
+    
+    def test_error_handling(self) -> bool:
+        """Test error handling."""
+        print("Testing error handling...")
+        
+        code = self.load_fixture("error_test")
+        result = self.execute_code(code)
+        
+        if result.get("exit_code") != 1:
+            print(f"❌ Error handling failed - wrong exit code: {result}")
+            return False
+        
+        if result.get("timeout", False):
+            print(f"❌ Error handling failed - marked as timeout: {result}")
+            return False
+            
+        print("✅ Error handling test passed")
+        return True
+    
+    def test_timeout_handling(self) -> bool:
+        """Test timeout handling."""
+        print("Testing timeout handling...")
+        
+        code = self.load_fixture("timeout_test")
+        result = self.execute_code(code, timeout=2)
+        
+        if not result.get("timeout", False):
+            print(f"❌ Timeout handling failed - not marked as timeout: {result}")
+            return False
+        
+        if result.get("exit_code") != -1:
+            print(f"❌ Timeout handling failed - wrong exit code: {result}")
+            return False
+            
+        if abs(result.get("execution_time", 0) - 2) > 0.5:
+            print(f"❌ Timeout handling failed - wrong execution time: {result}")
+            return False
+            
+        print("✅ Timeout handling test passed")
+        return True
+    
+    def test_concurrent_requests(self) -> bool:
+        """Test concurrent request handling (queuing)."""
+        print("Testing concurrent requests...")
+        
+        long_code = self.load_fixture("long_task")
+        quick_code = self.load_fixture("quick_task")
+        
+        results = []
+        threads = []
+        
+        def execute_and_store(code, index):
+            result = self.execute_code(code)
+            result["test_index"] = index
+            results.append(result)
+        
+        # Start long task
+        thread1 = threading.Thread(target=execute_and_store, args=(long_code, 1))
+        thread1.start()
+        threads.append(thread1)
+        
+        # Wait a bit, then start quick task
+        time.sleep(0.5)
+        thread2 = threading.Thread(target=execute_and_store, args=(quick_code, 2))
+        thread2.start()
+        threads.append(thread2)
+        
+        # Wait for both to complete
+        for thread in threads:
+            thread.join(timeout=10)
+        
+        if len(results) != 2:
+            print(f"❌ Concurrent requests failed - got {len(results)} results")
+            return False
+        
+        # Sort by test index to match with original order
+        results.sort(key=lambda x: x.get("test_index", 0))
+        
+        # First should be the long task (3+ seconds)
+        if results[0].get("execution_time", 0) < 2.5:
+            print(f"❌ Long task too fast: {results[0]}")
+            return False
+        
+        # Second should be quick (< 1 second)
+        if results[1].get("execution_time", 0) > 1:
+            print(f"❌ Quick task too slow: {results[1]}")
+            return False
+        
+        # Both should succeed
+        if results[0].get("exit_code") != 0 or results[1].get("exit_code") != 0:
+            print(f"❌ One or both concurrent tasks failed: {results}")
+            return False
+            
+        print("✅ Concurrent requests test passed")
+        return True
+    
+    def test_json_processing(self) -> bool:
+        """Test JSON processing capabilities."""
+        print("Testing JSON processing...")
+        
+        code = self.load_fixture("json_test")
+        result = self.execute_code(code)
+        
+        if result.get("exit_code") != 0:
+            print(f"❌ JSON processing failed: {result}")
+            return False
+            
+        print("✅ JSON processing test passed")
+        return True
+    
+    def test_transform_with_parameters(self) -> bool:
+        """Test transform function with table parameters."""
+        print("Testing transform with table parameters...")
+        
+        # This tests that the transform runner framework works but without actual Metabase API
+        code = self.load_fixture("transform_mock")
+        result = self.execute_code(code)
+        
+        if result.get("exit_code") != 0:
+            print(f"❌ Transform with parameters failed: {result}")
+            return False
+        
+        if "output_file" not in result:
+            print(f"❌ No CSV output from transform: {result}")
+            return False
+            
+        csv_content = self.read_file("output.csv")
+        if "doubled" not in csv_content:
+            print(f"❌ CSV missing expected column: {csv_content}")
+            return False
+            
+        print("✅ Transform with parameters test passed")
+        return True
+    
+    def test_metrics_tracking(self) -> bool:
+        """Test that metrics are being tracked."""
+        print("Testing metrics tracking...")
+        
+        initial_status = self.get_status()
+        initial_requests = initial_status.get("metrics", {}).get("total_requests", 0)
+        
+        # Execute a simple task with transform function
+        code = self.load_fixture("metrics_test")
+        result = self.execute_code(code)
+        
+        if result.get("exit_code") != 0:
+            print(f"❌ Metrics test execution failed: {result}")
+            return False
+        
+        final_status = self.get_status()
+        final_requests = final_status.get("metrics", {}).get("total_requests", 0)
+        
+        if final_requests <= initial_requests:
+            print(f"❌ Metrics not updated: {initial_requests} -> {final_requests}")
+            return False
+            
+        print("✅ Metrics tracking test passed")
+        return True
+    
+    def run_all_tests(self) -> bool:
+        """Run all tests."""
+        print("=" * 60)
+        print("Starting Python Execution Server Test Suite")
+        print("=" * 60)
+        
+        self.setup()
+        
+        tests = [
+            self.test_server_health,
+            self.test_basic_execution,
+            self.test_error_handling,
+            self.test_timeout_handling,
+            self.test_json_processing,
+            self.test_transform_with_parameters,
+            self.test_metrics_tracking,
+            self.test_concurrent_requests,  # Run this last as it takes longest
+        ]
+        
+        passed = 0
+        total = len(tests)
+        
+        try:
+            for test in tests:
+                try:
+                    if test():
+                        passed += 1
+                    print()  # Add spacing between tests
+                except Exception as e:
+                    print(f"❌ Test {test.__name__} crashed: {e}")
+                    print()
+        finally:
+            self.teardown()
+        
+        print("=" * 60)
+        print(f"Test Results: {passed}/{total} tests passed")
+        print("=" * 60)
+        
+        if passed == total:
+            print("🎉 All tests passed!")
+        else:
+            print(f"⚠️  {total - passed} tests failed")
+        
+        return passed == total
+
+
+if __name__ == "__main__":
+    import sys
+    
+    server_url = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:5001"
+    
+    tester = PythonServerTester(server_url)
+    success = tester.run_all_tests()
+    
+    sys.exit(0 if success else 1)

python-runner/transform_runner.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""
+Transform Runner - Executes user-defined transform functions and saves results as CSV.
+
+This script:
+1. Imports the user's code from /sandbox/script.py
+2. Creates a db object with read_table method using Metabase API
+3. Calls the transform() function defined by the user with db parameter
+4. Saves the returned DataFrame as CSV to the output file
+"""
+import os
+import sys
+import traceback
+import pandas as pd
+import json
+from pathlib import Path
+
+def read_jsonl_to_array(filepath):
+    data_array = []
+    with open(filepath, 'r', encoding='utf-8') as f:
+        for line in f:
+            data_array.append(json.loads(line.strip()))
+    return data_array
+
+def read_table(table_file, limit=None):
+    """
+        Read a table from Metabase using the API and return as a pandas DataFrame.
+
+        Args:
+            table_id: A table ID
+            limit: Optional row limit (defaults to API's default limit)
+
+        Returns:
+            pandas.DataFrame: The table data
+    """
+
+    try:
+
+        rows = read_jsonl_to_array(table_file)
+
+        if not rows:
+            # Empty table
+            return pd.DataFrame()
+
+        # Convert to DataFrame
+        df = pd.DataFrame(rows)
+        return df
+
+    except json.JSONDecodeError as e:
+        raise RuntimeError(f"Failed to parse JSON response for table: {e}")
+    except Exception as e:
+        raise RuntimeError(f"Failed to read table: {e}")
+
+
+def main():
+    try:
+        # Get the output file path from environment variable
+        output_file = os.environ.get('OUTPUT_FILE')
+        if not output_file:
+            print("ERROR: OUTPUT_FILE environment variable not set", file=sys.stderr)
+            sys.exit(1)
+
+        table_file_mapping_json = os.environ.get('TABLE_FILE_MAPPING', '{}')
+        table_file_mapping = json.loads(table_file_mapping_json)
+
+        # Import the user's script
+        sys.path.insert(0, '.')
+
+        # Import and execute the user's code
+        try:
+            import script
+        except ImportError as e:
+            print(f"ERROR: Failed to import user script: {e}", file=sys.stderr)
+            sys.exit(1)
+
+        # Check if transform function exists
+        if not hasattr(script, 'transform'):
+            print("ERROR: User script must define a 'transform()' function", file=sys.stderr)
+            sys.exit(1)
+
+        # Call the transform function with named table parameters
+        try:
+            # Check function signature to determine parameters
+            import inspect
+            sig = inspect.signature(script.transform)
+            if len(sig.parameters) > 0:
+                # Build kwargs with DataFrames for each named table
+                kwargs = {}
+                for param_name in sig.parameters:
+                    if param_name in table_file_mapping:
+                        table_file = table_file_mapping[param_name]
+                        kwargs[param_name] = read_table(table_file)
+
+                # Call transform with named arguments
+                result = script.transform(**kwargs)
+            else:
+                # Function takes no parameters, call without arguments for backward compatibility
+                result = script.transform()
+        except Exception as e:
+            print(f"ERROR: Transform function failed: {e}", file=sys.stderr)
+            traceback.print_exc(file=sys.stderr)
+            sys.exit(1)
+
+        # Validate that result is a pandas DataFrame
+        if not isinstance(result, pd.DataFrame):
+            print(f"ERROR: Transform function must return a pandas DataFrame, got {type(result)}", file=sys.stderr)
+            sys.exit(1)
+
+        # Save the DataFrame as CSV
+        try:
+            result.to_csv(output_file, index=False)
+            print(f"Successfully saved {len(result)} rows to CSV", file=sys.stdout)
+        except Exception as e:
+            print(f"ERROR: Failed to save DataFrame as CSV: {e}", file=sys.stderr)
+            traceback.print_exc(file=sys.stderr)
+            sys.exit(1)
+
+    except Exception as e:
+        print(f"FATAL ERROR: {e}", file=sys.stderr)
+        traceback.print_exc(file=sys.stderr)
+        sys.exit(1)
+
+if __name__ == "__main__":
+    main()

src/metabase/server/middleware/session.clj

@@ -198,7 +198,7 @@
     (u.password/verify-password passed-api-key "" api-key)
     (do-useless-hash)))
 
-(mu/defn- current-user-info-for-api-key :- [:maybe ::request.schema/current-user-info]
+(mu/defn current-user-info-for-api-key :- [:maybe ::request.schema/current-user-info]
   "Return User ID and superuser status for an API Key with `api-key-id"
   [api-key :- [:maybe :string]]
   (when (and api-key

src/metabase/upload/core.clj

@@ -7,9 +7,11 @@
 
 (p/import-vars
  [metabase.upload.impl
+  create-from-csv-and-sync!
   can-create-upload?
   current-database
   delete-upload!
   model-hydrate-based-on-upload
+  table-identifier
   update-action-schema
   update-csv!])

src/metabase/upload/impl.clj

@@ -527,7 +527,7 @@
          :num-rows          (count (rest rows))
          :generated-columns 0}))))
 
-(defn- create-from-csv-and-sync!
+(defn create-from-csv-and-sync!
   "This is separated from `create-csv-upload!` for testing"
   [{:keys [db filename file schema table-name display-name]}]
   (let [driver            (driver.u/database->driver db)