A simple implementation of pins for python for an RStudio Connect board

python_pins.py

import requests as req
import pickle
import tempfile
import tarfile
import os
import json
​
def pin_rsconnect(data, pin_name, pretty_pin_name, connect_server, api_key):
  '''
  Make a pin on RStudio Connect.
  
    Parameters:
          data: any object that has a to_json method
          pin_name (str): name of pin, only alphanumeric and underscores
          pretty_pin_name (str): display name of pin
          connect_server (str): RStudio Connect server address e.g. https://connect.example.com/
          api_key (str): API key of a user on RStudio Connect
      
     Return:
         Url of content
  
  '''
  # Save data
  dir = tempfile.TemporaryDirectory()
  data.to_json(dir.name + "/data.txt")
​
  # Create landing page
  i = open(dir.name + "/index.html", "w")
  lines = ["<h1>Python Pin", "\n"]
  for line in lines:
    i.write(line)
  i.close()
​
  # Create Manifest
  manifest = {
    "version": 1, 
    "locale": "en_US", 
    "platform": "3.5.1", 
    "metadata": {
      "appmode": "static", 
      "primary_rmd": None, 
      "primary_html": "index.html", 
      "content_category": "pin", 
      "has_parameters": False
      }, 
      "packages": None, 
      "files": None, 
      "users": None
  }
  with open(dir.name + "/manifest.json", "w") as m:
    json.dump(manifest, m)
​
  # Turn into tarfile
  tf = tempfile.NamedTemporaryFile()
  with tarfile.open(tf.name, "w:gz") as tar:
      tar.add(dir.name, arcname = os.path.basename(dir.name))
​
  content = get_content(pin_name, pretty_pin_name, connect_server, auth)
  content_url =  connect_server + "/__api__/v1/content/" + content['guid']
​
  # Upload Bundle
  with open(tf.name, "rb") as f:
    bundle = req.post(
    content_url + "/bundles",
    headers = auth, 
    data = f
  )
  bundle_id = bundle.json()['id']
​
  # Deploy bundle
  deploy = req.post(
    content_url + "/deploy", 
    headers = auth, 
    json = {'bundle_id': bundle_id}
  )
  return({
  "dash_url": content['dashboard_url'], 
  "content_url" : content['content_url']
  })
​
​
def get_content(pin_name, pretty_pin_name, connect_server, auth):
    content = req.get(
        connect_server + "/__api__/v1/content", 
        headers = auth, 
        params = {"name": pin_name}
    ).json()
    if content: # content item created already
        return content[0]
    else: # New content item, create
        data = {
            'access_type': 'acl',
            'name': pin_name,
            'title': pretty_pin_name
        }
        content = req.post(
            connect_server + "/__api__/v1/content", 
            headers=auth, 
            json=data
        ).json()
        return content
​
​
def pin_get_rsconnect(url):
  '''
  Get data from a python pin on RStudio Connect
  
    Parameters:
      url (str) content solo URL on Connect (NOT dashboard URL)
      
    Returns:
      JSON version of pin
  '''
  res = req.get(url + "/data.txt", headers = auth)
  return(res.json())
​
​
# Try it out
pin = pin_rsconnect(
    data = df, 
    pin_name = "my_pin_7788", 
    pretty_pin_name = "Python Pin", 
    connect_server = connect_server, 
    api_key = api_key
)
pin_get_rsconnect(pin['content_url'])