The following are provided as examples on how to interact with the Google Cloud Print API using Python, and will be of relevance both whether you are looking to receive print jobs and manage printers, or whether you wish to submit print jobs to GCP. You can easily interact with the Cloud Print API using HTTP requests. Python provides some modules to encode multipart data and to interact with the HTTP protocol using urllib and urllib2. All of the following examples will assume you've imported the needed Python modules.

Contents

1. General
   1. General Python Modules
   2. MultiPart Form Data
   3. Cookie Extraction
   4. Utility Functions
2. Authentication
   1. Google Credentials
   2. Auth Token
3. Google Cloud Print Service Queries
   1. Printer Queries
4. Printer Registration
   1. Anonymous Registration
   2. Authenticated Registration
5. Printer Jobs

General

Relevant Python Modules

Code samples assume the following definitions:

CRLF = '\r\n'
BOUNDARY = mimetools.choose_boundary()

# The following are used for authentication functions.
FOLLOWUP_HOST = 'www.google.com/cloudprint'
FOLLOWUP_URI = 'select%2Fgaiaauth'
GAIA_HOST = 'www.google.com'
LOGIN_URI = '/accounts/ServiceLoginAuth'
LOGIN_URL = 'https://www.google.com/accounts/ClientLogin'
SERVICE = 'cloudprint'
OAUTH = 'This_should_contain_oauth_string_for_your_username'

# The following are used for general backend access.
CLOUDPRINT_URL = 'https://www.google.com/cloudprint'
# CLIENT_NAME should be some string identifier for the client you are writing.
CLIENT_NAME = 'Cloud Print API Client'

# The following object is used in the sample code, but is not necessary.
logger = (code not included here, but it should be an initialized logger object with the handler and formatting or your choice)

# All of the calls to GetUrl assume you've run something like this:
tokens = GetAuthTokens(email, password)

Note: To obtain the OAUTH constant above, get an oauth client id by following this link.

MultiPart Form Data

When sending many of the requests you'll need to encode the request into multiple parts. Here's a sample method of encoding your data.

def EncodeMultiPart(fields, files, file_type='application/xml'):
    """Encodes list of parameters and files for HTTP multipart format.

    Args:
      fields: list of tuples containing name and value of parameters.
      files: list of tuples containing param name, filename, and file contents.
      file_type: string if file type different than application/xml.
    Returns:
      A string to be sent as data for the HTTP post request.
    """
    lines = []
    for (key, value) in fields:
      lines.append('--' + BOUNDARY)
      lines.append('Content-Disposition: form-data; name="%s"' % key)
      lines.append('')  # blank line
      lines.append(value)
    for (key, filename, value) in files:
      lines.append('--' + BOUNDARY)
      lines.append(
          'Content-Disposition: form-data; name="%s"; filename="%s"'
          % (key, filename))
      lines.append('Content-Type: %s' % file_type)
      lines.append('')  # blank line
      lines.append(value)
    lines.append('--' + BOUNDARY + '--')
    lines.append('')  # blank line
    return CRLF.join(lines)

When you are passing in file_type, you could use the mimetypes module to get the file type by calling mimetypes.guess_type(filename). Fields will typically be some key-value pairs that the server will need. In our case, we'll typically pass in things like printerid, content identifiers, print job title, etc. Files may consist of things like a printer capabilities file, or a raw graphic file.

HTTP Get/Post Method

We will need a method for sending HTTP requests to the Cloud Print Service. We also need to retrieve authentication tokens from Google's account authentication service; consequently the HTTP requests need mechanisms to include this authentication data in the cookie headers. The following method depends on the Python module urllib2. Here is a sample method for these HTTP requests.

def GetUrl(url, tokens, data=None, cookies=False, anonymous=False):
  """Get URL, with GET or POST depending data, adds Authorization header.

  Args:
    url: Url to access.
    tokens: dictionary of authentication tokens for specific user.
    data: If a POST request, data to be sent with the request.
    cookies: boolean, True = send authentication tokens in cookie headers.
    anonymous: boolean, True = do not send login credentials.
  Returns:
    String: response to the HTTP request.
  """
  request = urllib2.Request(url)
  if not anonymous:
    if cookies:
      logger.debug('Adding authentication credentials to cookie header')
      request.add_header('Cookie', 'SID=%s; HSID=%s; SSID=%s' % (
          tokens['SID'], tokens['HSID'], tokens['SSID']))
    else:  # Don't add Auth headers when using Cookie header with auth tokens.
      request.add_header('Authorization', 'GoogleLogin auth=%s' % tokens['Auth'])
  request.add_header('X-CloudPrint-Proxy', 'api-prober')
  if data:
    request.add_data(data)
    request.add_header('Content-Length', str(len(data)))
    request.add_header('Content-Type', 'multipart/form-data;boundary=%s' % BOUNDARY)

  # In case the gateway is not responding, we'll retry.
  retry_count = 0
  while retry_count < 5:
    try:
      result = urllib2.urlopen(request).read()
      return result
    except urllib2.HTTPError, e:
      # We see this error if the site goes down. We need to pause and retry.
      err_msg = 'Error accessing %s\n%s' % (url, e)
      logger.error(err_msg)
      logger.info('Pausing %d seconds', 60)
      time.sleep(60)
      retry_count += 1
      if retry_count == 5:
        return err_msg

Cookie Extraction

The following method is used to extract a cookie from a set-cookie string.

def GetCookie(cookie_key, cookie_string):
    """Extract the cookie value from a set-cookie string.

    Args:
      cookie_key: string, cookie identifier.
      cookie_string: string, from a set-cookie command.
    Returns:
      string, value of cookie.
    """
    logger.debug('Getting cookie from %s', cookie_string)
    id_string = cookie_key + '='
    cookie_crumbs = cookie_string.split(';')
    for c in cookie_crumbs:
      if id_string in c:
        cookie = c.split(id_string)
        return cookie[1]
    return None

Utility Functions

The following functions perform some general utility operations that we'll use in our other code.

def ConvertJson(json_str):
  """Convert json string to a python object.

  Args:
    json_str: string, json response.
  Returns:
    dictionary of deserialized json string.
  """
  j = {}
  try:
    j = json.loads(json_str)
    j['json'] = True
  except ValueError, e:
    # This means the format from json_str is probably bad.
    logger.error('Error parsing json string %s\n%s', json_str, e)
    j['json'] = False
    j['error'] = e

  return j

def GetKeyValue(line, sep=':'):
    """Return value from a key value pair string.

    Args:
      line: string containing key value pair.
      sep: separator of key and value.
    Returns:
      string: value from key value string.
    """
    s = line.split(sep)
    return StripPunc(s[1])

def StripPunc(s):
  """Strip puncuation from string, except for - sign.

  Args:
    s: string.
  Returns:
    string with puncuation removed.
  """
  for c in string.punctuation:
    if c == '-':  # Could be negative number, so don't remove '-'.
      continue
    else:
      s = s.replace(c, '')
  return s.strip()

def Validate(response):
  """Determine if JSON response indicated success."""
  if response.find('"success": true') > 0:
    return True
  else:
    return False

def GetMessage(response):
  """Extract the API message from a Cloud Print API json response.

  Args:
    response: json response from API request.
  Returns:
    string: message content in json response.
  """
  lines = response.split('\n')
  for line in lines:
    if '"message":' in line:
      msg = line.split(':')
      return msg[1]

  return None

def ReadFile(pathname):
  """Read contents of a file and return content.

  Args:
    pathname: string, (path)name of file.
  Returns:
    string: contents of file.
  """
  try:
    f = open(pathname, 'rb')
    try:
      s = f.read()
    except IOError, e:
      logger('Error reading %s\n%s', pathname, e)
    finally:
      f.close()
      return s
  except IOError, e:
    logger.error('Error opening %s\n%s', pathname, e)
    return None

def WriteFile(file_name, data):
  """Write contents of data to a file_name.

  Args:
    file_name: string, (path)name of file.
    data: string, contents to write to file.
  Returns:
    boolean: True = success, False = errors.
  """
  status = True

  try:
    f = open(file_name, 'wb')
    try:
      f.write(data)
    except IOError, e:
      logger.error('Error writing %s\n%s', file_name, e)
      status = False
    finally:
      f.close()
  except IOError, e:
    logger.error('Error opening %s\n%s', file_name, e)
    status = False

  return status

def Base64Encode(pathname):
  """Convert a file to a base64 encoded file.

  Args:
    pathname: path name of file to base64 encode..
  Returns:
    string, name of base64 encoded file.
  For more info on data urls, see:
    http://en.wikipedia.org/wiki/Data_URI_scheme
  """
  b64_pathname = pathname + '.b64'
  file_type = mimetypes.guess_type(pathname)[0] or 'application/octet-stream'
  data = ReadFile(pathname)

  # Convert binary data to base64 encoded data.
  header = 'data:%s;base64,' % file_type
  b64data = header + base64.b64encode(data)

  if WriteFile(b64_pathname, b64data):
    return b64_pathname
  else:
    return None

Authentication

Google Cloud Print uses Google login credentials for authorization and authentication. The token dictionary from GaiaLogin() will be combined with the token dictionary for GetLoginCredentials() to form a complete token dictionary. The following methods can be used to retrieve the needed authentication credentials for a specific user. These methods make use of the ClientLogin API. Note however that we recommend that developers use OAuth 2.0. See here for more information on Google's various auth mechanisms, and our Developer Guide for details on using GCP with OAuth.

Google Credentials

def GaiaLogin(email, password):
    """Login to gaia using HTTP post to the gaia login page.

    Args:
      email: string,
      password: string
    Returns:
      dictionary of authentication tokens.
    """
    tokens = {}
    cookie_keys = ['SID', 'LSID', 'HSID', 'SSID']
    email = email.replace('+', '%2B')
    # Needs to be some random string.
    galx_cookie = base64.b64encode('%s%s' % (email, time.time()))

    # Simulate submitting a gaia login form.
    form = ('ltmpl=login&fpui=1&rm=hide&hl=en-US&alwf=true'
            '&continue=https%%3A%%2F%%2F%s%%2F%s'
            '&followup=https%%3A%%2F%%2F%s%%2F%s'
            '&service=%s&Email=%s&Passwd=%s&GALX=%s' % (FOLLOWUP_HOST,
            FOLLOWUP_URI, FOLLOWUP_HOST, FOLLOWUP_URI, SERVICE, email,
            password, galx_cookie))
    login = httplib.HTTPS(GAIA_HOST, 443)
    login.putrequest('POST', LOGIN_URI)
    login.putheader('Host', GAIA_HOST)
    login.putheader('content-type', 'application/x-www-form-urlencoded')
    login.putheader('content-length', str(len(form)))
    login.putheader('Cookie', 'GALX=%s' % galx_cookie)
    logger.debug('Sent POST content: %s', form)
    login.endheaders()
    logger.info('HTTP POST to https://%s%s', GAIA_HOST, LOGIN_URI)
    login.send(form)

    (errcode, errmsg, headers) = login.getreply()
    login_output = login.getfile()
    login_output.close()
    login.close()
    logger.info('Login complete.')

    if errcode != 302:
      logger.error('Gaia HTTP post returned %d, expected 302', errcode)
      logger.error('Message: %s', errmsg)

    for line in str(headers).split('\r\n'):
      if not line: continue
      (name, content) = line.split(':', 1)
      if name.lower() == 'set-cookie':
        for k in cookie_keys:
          if content.strip().startswith(k):
            tokens[k] = GetCookie(k, content)

    if not tokens:
      logger.error('No cookies received, check post parameters.')
      return None
    else:
      logger.debug('Received the following authorization tokens.')
      for t in tokens:
        logger.debug(t)
      return tokens

Auth Token

def GetAuthTokens(email, password):
    """Assign login credentials from GAIA accounts service.

    Args:
      email: Email address of the Google account to use.
      password: Cleartext password of the email account.
    Returns:
      dictionary containing Auth token.
    """
    # First get GAIA login credentials using our GaiaLogin method.
    tokens = GaiaLogin(email, password)

    # We still need to get the Auth token.
    params = {'accountType': 'GOOGLE',
              'Email': email,
              'Passwd': password,
              'service': SERVICE,
              'source': CLIENT_NAME}
    stream = urllib.urlopen(LOGIN_URL, urllib.urlencode(params))

    for line in stream:
      if line.strip().startswith('Auth='):
        tokens['Auth'] = line.strip().replace('Auth=', '')

Note: While API calls can be made to work with several different Google authentication schemes, we recommend that you use OAuth2. Reference this page to see how to change the authentication request above to use OAuth2.

Google Cloud Print Service Queries

The following functions will illustrate how to make some general queries against the Cloud Print Service backend. In order to get data, these functions require authentication tokens acquired from GetAuthTokens(). The following examples assume tokens = GetAuthTokens(valid_email@address>, valid_password).

Printer Queries

def GetPrinters(proxy=None):
  """Get a list of all printers, including name, id, and proxy.

  Args:
    proxy: name of proxy to filter by.
  Returns:
    dictionary, keys = printer id, values = printer name, and proxy.
  """
  printers = {}
  values = {}
  tokens = ['"id"', '"name"', '"proxy"']
  for t in tokens:
    values[t] = ''

  if proxy:
    response = GetUrl('%s/list?proxy=%s' % (CLOUDPRINT_URL, proxy), tokens)
  else:
    response = GetUrl('%s/search' % CLOUDPRINT_URL, tokens)

  sections = response.split('{')
  for printer in sections:
    lines = printer.split(',')
    for line in lines:
      for t in tokens:
        if t in line:
          values[t] = GetKeyValue(line)
    if values['"id"']:
      printers[values['"id"']] = {}
      printers[values['"id"']]['name'] = values['"name"']
      printers[values['"id"']]['proxy'] = values['"proxy"']

  return printers

def Query(printer, proxy=None):
  """Query for a printer by name.

  Args:
    printer: string, name of printer.
    proxy: string, id of proxy to filter by.
  Returns:
    string: the printer_id or None if not found.
  """
  if proxy:
    printers = GetPrinters(proxy=proxy)
    for p in printers:
      if printers[p]['name'] == printer:
        return p
  else:
    printer_id = None
    response = GetUrl('%s/search?q=%s' % (CLOUDPRINT_URL, printer), tokens)
    sections = response.split('"printers": [')
    lines = sections[1].split(',')
    for line in lines:
      if '"id"' in line:
        printer_id = GetKeyValue(line)
      elif '"name"' in line:
        printer_name = GetKeyValue(line)
        if printer_name == printer:
          logger.debug('Printer %s is registered', printer)
          if printer_id:
            return printer_id
          else:
            logger.error('Malformed api response.')
            return None

  return None

Printer Registration

The following functions will illustrate how to register and delete printers with the Cloud Print Service. Anonymous registration is supported as well.

Anonymous Registration

Anonymous registration requires registering without any login credentials, and then taking some of the returning tokens to complete the registration. Here are the steps required:

• Access registration URL using HTTPS without authentication tokens
• Get token back from Cloud Print Service
• Use the token to claim the printer (with authentication tokens)
• Get the auth token back from the claim printer step
• Send auth token to polling URL

You can learn more about anonymous registration from our developer guide.

Here is function that will send a register request without credentials.

def RegisterAnonPrinter(printer):
    """Make a Printer Registration call as anonymous user.

    Args:
      printer: string, name of printer.
    Returns:
      dictionary: this will contain a boolean status and a json string.
    """
    result = {}
    data = EncodeMultiPart([('printer', printer),
                            ('proxy', 'SelfRegProxy')],
                            [('capabilities', 'capabilities',
                             Constants.CAPS)])
    response = GetUrl('%s/register' % CLOUDPRINT_URL, data, anonymous=True)

    return response

The following method will use RegisterAnonPrinter() above to complete the entire self registration process:

def SelfRegister():
  """A function to initiate and complete the entire self registration process.

  Returns:
    boolean: True = Success, False = errors.
  """
  pname = 'SelfRegPrinterName'
  response = RegisterAnonPrinter(pname)
  if Validate(response):
    reg = ConvertJson(response)
    printer_id = reg['printers'][0]['id']
    if not reg['json']:
      logger.error(reg['error'])
      return False
    else:
      poll_url = '%s%s' % (reg['polling_url'], OAUTH)
      # This is manual step to go and claim your printer.
      print 'Go claim your printer at this url:'
      print 'https://www.google.com/cloudprint/claimprinter.html'
      print 'Use token: %s' % reg['registration_token']
      raw_input('Press Enter to continue once you've claimed your printer...')
      res = GetUrl(poll_url, anonymous=True)
      if Validate(res):
        poll = ConvertJson(res)
        if not poll['json']:
          logger.error(poll['error'])
          return False
        else:
          logger.info('Self Registration Authorization Code: %s',
                      poll['authorization_code'])
          reg_id = Query(pname)
          if reg_id == printer_id:
            return True
          else:
            logger.error('Registered id doesn't match printer id.')
            return False
      else:  # poll_url wasn't successful.
        logger.error(GetMessage(res))
        return False
  else:  # RegisterAnonPrinter() was not successful.
    logger.error(GetMessage(response))
    return False

Authenticated Registration

These functions require authentication tokens acquired from GetAuthTokens().

def RegisterPrinter(printer, print_cap):
  """Make a Printer Registration Call.

  Args:
    printer: string, name of printer, also used as id and proxy name.
    print_cap: string, XML formatted string of printer capabilities.
  Returns:
    boolean: True = success, False = errors.
  """
  edata = EncodeMultiPart([('printer', printer),
                           ('printerid', printer),
                           ('proxy', printer)],
                           [('capabilities', 'capabilities',
                            print_cap)])
  response = GetUrl('%s/register' % CLOUDPRINT_URL, tokens, data=edata)
  return Validate(response)


def DeletePrinter(printerid):
  """Delete the specified printerid from the cloud printer service.

  Args:
    printerid: string, unique id registered on back end.
  Returns:
    boolean: True = deleted, False = errors.
  """
  results = {}
  msg_tokens = ['"success"', '"message"']
  response = GetUrl('%s/delete?printerid=%s' % (CLOUDPRINT_URL, printerid), tokens)
  lines = response.split(',')
  for line in lines:
    for t in msg_tokens:
      if t in line:
        results[t] = GetKeyValue(line)
  if results['"success"'] == 'true':
    logger.debug('Printer id %s removed.', printerid)
    return True
  else:
    logger.info('Error removing printer id %s', printerid)
    logger.info('%s', results['"message"'])
    return False

def DeletePrinters(proxy_id=None, wildcard=None):
  """Delete all printers, filtering by proxy_id or printer name.

  Args:
    proxy_id: string, proxy id to filter by.
    wildcard: string, only delete printers containing this string.
  Returns:
    integer: number of errors encountered during entire delete cycle.
  """
  errors = 0
  if proxy_id:
    printers = GetPrinters(proxy=proxy_id)
  else:
    printers = GetPrinters()

  for p in printers:  # This should be the printer id.
    if wildcard:
      if wildcard in printers[p]['name']:
        status = DeletePrinter(p)
      else:
        status = None
    else:
      status = DeletePrinter(p)
    if not status:
      errors += 1

  return errors

Printer Jobs

The following functions demonstrate how to interact with printer jobs.

def GetJobs(printerid=None, jobid=None):
  """Get a list of printer jobs.

  Args:
    printerid: if specified, filter by printer id.
    jobid: if specified, filter by job id.
  Returns:
    dictionary of job id's (key) and printer id's (value).
  """
  job_tokens = {'k': '"id"',
                'v': '"printerid"',
               }
  values = {}
  jobs = {}

  if printerid:
    response = GetUrl('%s/fetch?printerid=%s' % (CLOUDPRINT_URL, printerid), tokens)
  elif jobid:
    response = GetUrl('%s/fetch?jobid=%s' % (CLOUDPRINT_URL, jobid), tokens)
  else:
    response = GetUrl('%s/jobs' % CLOUDPRINT_URL, tokens)
  sections = response.split('{')
  for job in sections:
    lines = job.split(',')
    for line in lines:
      for t in job_tokens:
        if job_tokens[t] in line:
          values[t] = GetKeyValue(line)
    if values:
      jobs[values['k']] = values['v']
      values = {}

  return jobs

There are various types of print jobs you could submit, ranging from raw graphic files, PDF files, to a Google document or email thread. Here is a method that will handle submitting these types of printer jobs.

def SubmitJob(printerid, jobtype, jobsrc):
  """Submit a job to printerid with content of dataUrl.

  Args:
    printerid: string, the printer id to submit the job to.
    jobtype: string, must match the dictionary keys in content and content_type.
    jobsrc: string, points to source for job. Could be a pathname or id string.
  Returns:
    boolean: True = submitted, False = errors.
  """
  if jobtype == 'pdf':
    b64file = Base64Encode(jobsrc])
    fdata = ReadFile(b64file)
    hsid = True
  elif jobtype in ['png', 'jpeg']
    fdata = ReadFile(jobsrc)
  else:
    fdata = None

  # Make the title unique for each job, since the printer by default will name
  # the print job file the same as the title.

  datehour = time.strftime('%b%d%H%M', time.localtime())
  title = '%s%s' % (datehour, jobsrc)
    """The following dictionaries expect a certain kind of data in jobsrc, depending on jobtype:
    jobtype                jobsrc
    ================================================
    pdf                    pathname to the pdf file
    png                    pathname to the png file
    jpeg                   pathname to the jpeg file
    ================================================
    """
  content = {'pdf': fdata,
             'jpeg': jobsrc,
             'png': jobsrc,
            }
  content_type = {'pdf': 'dataUrl',
                  'jpeg': 'image/jpeg',
                  'png': 'image/png',
                 }
  headers = [('printerid', printerid),
             ('title', title),
             ('content', content[jobtype]),
             ('contentType', content_type[jobtype])]
  files = [('capabilities', 'capabilities', '{"capabilities":[]}')]
  if jobtype in ['pdf', 'jpeg', 'png']:
    files.append(('content', jobsrc, fdata))
    edata = EncodeMultiPart(headers, files, file_type=content_type[jobtype])
  else:
    edata = EncodeMultiPart(headers, files)

  response = GetUrl('%s/submit' % CLOUDPRINT_URL, tokens, data=edata,
                    cookies=False)
  status = Validate(response)
  if not status:
    error_msg = GetMessage(response)
    logger.error('Print job %s failed with %s', jobtype, error_msg)

  return status

Except as otherwise noted, the content of this page is licensed under the Creative Commons Attribution 3.0 License, and code samples are licensed under the Apache 2.0 License. For details, see our Site Policies.