Skip to content

Instantly share code, notes, and snippets.

@snowch
Created April 23, 2015 12:56
Show Gist options
  • Save snowch/9f3b9f35ed62436f8d12 to your computer and use it in GitHub Desktop.
Save snowch/9f3b9f35ed62436f8d12 to your computer and use it in GitHub Desktop.

Objective

Test that changing the database linked to the account is thread safe.

How to test by making parallel invocations of?

  • /vechicles
  • /customers

How can you be sure that a parallel invocation of /vehicles and /customers will actually run parallel? E.g. on different cores at exactly the same time?

application.py

# -*- encoding: utf8 -*-
"""
Main Flask class.
"""

from flask import Flask, jsonify

app = Flask(__name__)
app.debug = False

@app.route("/vehicles", methods=['GET', 'OPTIONS'])
def vehicles():
    vechicles = Vehicles()
    return vehicles.do_something()  # actual method not important

@app.route("/customers", methods=['GET', 'OPTIONS'])
def customers():
    customers = Customers()
    return customers.do_something()  # actual method not important

if __name__ == "__main__":
    app.run()

base.py

import os
import cloudant

from   requests.adapters import HTTPAdapter

# the httpAdapter is a global python object that is shared by
# all Cloudan Python Account instances
httpAdapter = HTTPAdapter(pool_connections=10, pool_maxsize=100)

class Base(object):
    def __init__(self, db='vehicles', async=False):
        self.dbname   = db
        self.username = os.environ.get('CLOUDANT_DB')
        self.apikey = os.environ.get('CLOUDANT_DB_APIKEY')
        self.apipassword = os.environ.get('CLOUDANT_DB_APIPASSWORD')
        self.account  = cloudant.Account(self.username, async=async)
        self.account._session.auth = (self.apikey, self.apipassword)
        self.account._session.adapters['https://'] = httpAdapter
        self.db       = self.account.database(db)

Vehicles.py

# -*- encoding: utf8 -*-
from . import Base

class Vehicles(Base):ch

    def __init__(self):

        super(self.__class__, self).__init__(async=False)

customers.py

# -*- encoding: utf8 -*-
from . import Base

class Customers(Base):

    def __init__(self):

        super(self.__class__, self).__init__(db='customers', async=False)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment