spec.json

{
  "swagger": "2.0",
  "info": {
    "version": "1.0.0",
    "title": "test"
  },
  "host": "localhost",
  "basePath": "/",
  "schemes": [
    "http"
  ],
  "consumes": [
    "application/json"
  ],
  "produces": [
    "application/json"
  ],"security":[
    {
      "basic_auth":[]
    }
  ],  
  "x-api-definition": {
    "api_name": "thing",
    "personnel": {
      "owner": "person",
      "development_team": "person",
      "developers": [
        "Tyghe.Vallard"
      ]
    },
    "lifecycle_status": "BUILD",
    "production_date": null,
    "retirement_date": null,
    "api_framework": "PASSTHROUGH-NODE-PROXY",
    "overall_data_classification": "INTERNAL",
    "endpoints": {
      "external": {
        "production" : null,
        "stage": null,
        "ci": null,
        "qa": null
      },
      "internal": {
        "production": "",
        "stage": "",
        "ci": null,
        "qa": null
      }
    },
    "links": {
      "issue_tracker": "",
      "api_wiki_page": "",
      "code_repository": "",
      "ci_pipeline": ""
    }
  },
  "paths": {
    "/create": {
      "x-data_classification": "INTERNAL",
      "post": {
        "operationId": "create",
        "x-incident_priority": "P4",
        "x-response_time_sla": "60000ms",
        "x-success_http_code": [
          "202"
        ],
        "x-expected_tps": "100",
        "description": "Creates",
        "produces": [
          "application/json"
        ],
        "parameters": [
          {
            "name": "accept",
            "in": "header",
            "description": "application/json",
            "required": true,
            "type": "string",
            "default": "application/json"
          },
          {
            "name": "key",
            "in": "query",
            "description": "The consumer's valid API key.",
            "required": true,
            "type": "string"
          },
          {
            "name": "Authorization",
            "in": "header",
            "description": "Basic Auth access token.",
            "required": true,
            "type": "string"
          },
          {
            "name": "body",
            "in": "body",
            "description": "Create body",
            "required": true,
            "schema": {
              "$ref": "#/definitions/Create"
            }
          }
        ],
        "responses": {
          "202": {
            "description": "Test",
            "schema": {
              "$ref": "#/definitions/Create"
            }
          }
        }
      }
    }
  },
  "definitions": {
    "id": {
      "type": "string",
      "description": "Unique identifier for instance"
    },
    "Create": {
      "type": "object",
      "required": [
        "id"
      ],
      "properties": {
        "id": {
          "$ref": "#/definitions/id"
        }
      }
    }
  }
}

test.py

from path import Path
import json
from path import Path
from bravado.client import SwaggerClient
from bravado.exception import HTTPError
from bravado.requests_client import RequestsClient
import requests
from requests import exceptions

LOGFILE = Path('/dev/null')

#from test_v1 import *
def clean_x_defs(spec):
    '''
    Clean all x-* keys from the spec as the Bravado client doesn't seem to handle
    them correctly

    :param dict spec: dict representation of swagger spec
    :returns: swagger spec dict with all x-* and X-* keys removed
    '''
    for k in spec.keys():
        if k.startswith('x-'):
            #print "Removed {0} from spec".format(k)
            del spec[k]
    for k in spec:
        if isinstance(spec[k], dict):
            spec[k] = clean_x_defs(spec[k])
    return spec

def load_swagger(spec_pth):
    '''
    Load the swagger json
    
    :param str spec_pth: Path to swagger json file
    :returns: dict representation of swagger spec
    '''
    spec_pth = Path(spec_pth)
    spec = None
    with open(spec_pth.expand()) as fh:
        spec = json.load(fh)
    spec = clean_x_defs(spec)
    return spec

class ResponseException(Exception):
    '''
    Excpetion to wrap some message with an original exception
    to preserve the message
    '''
    def __init__(self, msg, original_exception):
        self.msg = msg
        self.original_exception = original_exception

def response_to_string(response, op):
    '''
    Convert a Bravado swagger response into a easily viewable string
    '''
    sio = StringIO()
    sio.write('Status Code:\n')
    pprint(response.status_code, stream=sio)
    sio.write("Text:\n")
    pprint(response.text, stream=sio)
    sio.write("Reason:\n")
    pprint(response.reason, stream=sio)
    sio.write("Headers\n")
    pprint(response.headers, stream=sio)
    if response.status_code != 204:
        sio.write("Json\n")
        try:
            pprint(response.json(), stream=sio)
        except Exception as e:
            raise ResponseException(sio.getvalue(), e)
    return sio.getvalue()

def print_response(response, op):
    '''
    Prints a response from bravado to the terminal
    '''
    try:
        print response_to_string(response, op)
    except ResponseException as e:
        print e.msg
        print str(e.original_exception)

def get_client(spec, ssl_verify=False):
    '''
    Get bravado client from spec

    :param SwaggerClient spec: swagger parsed spec
    :param bool ssl_verify: Verify ssl or not
    :returns: bravado client
    '''
    # Get client for our swagger spec     
    http_client = RequestsClient()
    http_client.session.verify = ssl_verify
    client = SwaggerClient.from_spec(
        spec, config={'also_return_response': True},
        http_client=http_client
    )
    return client

def log(msg):
    if not msg.endswith('\n'):
        msg += '\n'
    LOGFILE.write_text(msg, append=True)

def log_response(response, op):
    try:
        response_str = response_to_string(response, op)
        log(response_str)
    except ResponseException as e:
        response_str = e.msg
        log(response_str)
        log(str(e.original_exception))
        raise e.original_exception

def create_container(client, headers, **options):
    '''
    Issue a create operation on the WASAPI api using the supplied
    options

    :returns: swagger_model, http_response object
    '''
    Create = client.get_model('Create')
    create = Create(**options)
    log('--------- Create Request: {0}'.format(create))
    create_response = client.create.create(
        body=create,
        key='apikey',
        _request_options={
            "headers": headers,
            "response_callbacks": [print_response, log_response]
        }
    )
    print create_response
    print dir(create_response)
    # swagger_response is a swagger model based on result spec(aka 201 response)
    # response is some bravado response
    swagger_response, response = create_response.result()
    return (swagger_response, response)

if __name__ == '__main__':
  body = dict(
    id='test'
  )

  headers = {
      'accept': 'application/json',
      'Authorization': 'token'
  }

  spec = load_swagger('spec.json')
  spec['host'] = 'localhost:4567'
  spec['schemes'] = ['http']

  client = get_client(spec)

  create_container(client, headers, **body)