Jupyter -> Vega -> Kibana visualization (without disabling security). How?

Elastic Stack Kibana
,
2

I have to say that I wasn't aware of this ML project; it is genius!; the authorization problem could be solved easily; just create a role with privileges to run as "kiabana_system" and add to your Elastic client a header; something like this:

headers = {
    'es-security-runas-user': 'kibana_system'
}
cloud_id = "secret"
http_auth = ("username", "password")

es = Elasticsearch(cloud_id=cloud_id, http_auth=http_auth,
                    headers=headers)

However, even if the visualization gets saved directly to the kibana index; it doesn't get listed in the visualizations list, so this is not really a solution in the long term; the way to go is with the Kibana API and create a Saved Object; that should work on the long run. Having that in mind a quick modification of the example and buala... it works!:

def saveVegaVis(client, index, visName, altairChart):
    chart_json = json.loads(altairChart.to_json())
    chart_json['data']['url'] = {
        "%context%": True,
        "index": index,
        "body": { 
            "size": 10000
        }
    }

    visState = {
      "type": "vega",
      "aggs": [],
      "params": {
        "spec": json.dumps(chart_json, sort_keys=True, indent=4, separators=(',', ': ')),
      },
      "title": visName
    }

    visSavedObject={
        #"visualization" : {
        "attributes" : {
          "title" : visName,
          "visState" : json.dumps(visState, sort_keys=True, indent=4, separators=(',', ': ')),
          "uiStateJSON" : "{}",
          "description" : "",
          "version" : 1,
          "kibanaSavedObjectMeta" : {
            "searchSourceJSON" : json.dumps({
              "query": {
                "language": "kuery",
                "query": ""
              },
              "filter": []
            }),
          }
        }#,
        #"type" : "visualization",
        #"references" : [ ],
        #"migrationVersion" : {
        #  "visualization" : "7.7.0"
        #},
        #"updated_at" : datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S.000Z")
    }

    import requests
    
    myurl = 'http://kib01:5601/api/saved_objects/visualization'

    headers = {
        'kbn-xsrf' : 'true',
        'Content-Type' : 'application/json'
    }
    
    x = requests.post(myurl, data=json.dumps(visSavedObject), auth=("myuser", "mysupersecretpass"), headers=headers)
    
    print(x.text)
    
    #return client.index(index='.kibana',id='visualization:'+visName,body=visSavedObject)