How to upload a Pulsar Functions jar package in Python with the REST API

[engineer537]

Hi Pulsar Community!

I hope you're doing well. 😊
I'm trying to upload a JAR package to a Pulsar cluster using Python. Could you please guide me on how to achieve this using either pulsar-admin, pulsar-client, or the RESTful API, but without relying on subprocess?

Any help or pointers would be greatly appreciated. Thank you!

[lhotari]

Here's an example of using the REST API:
#22266 (comment)

this is the way curl can be used to create functions from a jar/nar file:

cat >/tmp/functionconfig.json <<EOF
{
  "tenant": "public",
  "namespace": "default",
  "name": "myfunction",
  "className": "MyFunction",
  "runtime": "JAVA",
  "inputs": [
    "public/default/input-topic"
  ],
  "output": "public/default/output-topic",
  "autoAck": true,
  "parallelism": 1
}
EOF

curl -H "Authorization: Bearer $(cat token)" -v -X POST \
  -F "functionConfig=@/tmp/functionconfig.json;type=application/json" \
  -F "data=@function.jar;type=application/octet-stream" \
  http://pulsar-function-worker:6750/admin/v3/functions/public/default/myfunction

You can adapt that to Python with ChatGPT or Claude.
Claude generated Python code for the above example (I haven't validated the solution):

import json
import requests
import os

def deploy_pulsar_function(token_path, function_jar_path):
    # Create function config
    function_config = {
        "tenant": "public",
        "namespace": "default", 
        "name": "myfunction",
        "className": "MyFunction",
        "runtime": "JAVA",
        "inputs": ["public/default/input-topic"],
        "output": "public/default/output-topic",
        "autoAck": True,
        "parallelism": 1
    }
    
    # Save config to temp file
    config_path = "/tmp/functionconfig.json"
    with open(config_path, "w") as f:
        json.dump(function_config, f)
    
    # Read token
    with open(token_path) as f:
        token = f.read().strip()
    
    # Prepare request
    headers = {"Authorization": f"Bearer {token}"}
    files = {
        "functionConfig": ("functionconfig.json", open(config_path, "rb"), "application/json"),
        "data": ("function.jar", open(function_jar_path, "rb"), "application/octet-stream")
    }
    
    # Make request
    url = "http://pulsar-function-worker:6750/admin/v3/functions/public/default/myfunction"
    response = requests.post(url, headers=headers, files=files)
    
    # Cleanup
    os.remove(config_path)
    
    return response.status_code, response.text

if __name__ == "__main__":
    status, text = deploy_pulsar_function("token", "function.jar")
    print(f"Status: {status}")
    print(f"Response: {text}")