kenneho.net

Patching Python instance methods

November 10, 2018 Kenneth Holter Comments 0 Comment

I’m working on a project in which one of our services need to publish messages to a RabbitMQ exchange. Consider this code excerpt from the message queue client:

# file: mq_client.py
import pika

class MQ_Client():
    connection = None
    channel = None
    exchange_name="my_exchange"

    def connect(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host="mq_server")
        self.channel = connection.channel()

    def publish(self, message):

        properties = pika.BasicProperties(content_type='text/json', delivery_mode=1)

        if self.channel.basic_publish(exchange=self.exchange_name, routing_key='', properties=properties, body=message):
            log.info("Successfully published this message to exchange \"" + self.exchange_name + "\": " + message)
        else:
            log.error("Failed to publish message \"" + message + "\" to the exchange \"" + self.exchange_name + "\"!")
            raise Exception("Failed to publish message to the queue.")

# file: mq_client.py

import pika

class MQ_Client():

connection = None

channel = None

exchange_name="my_exchange"

def connect(self):

self.connection = pika.BlockingConnection(pika.ConnectionParameters(host="mq_server")

self.channel = connection.channel()

def publish(self, message):

properties = pika.BasicProperties(content_type='text/json', delivery_mode=1)

if self.channel.basic_publish(exchange=self.exchange_name, routing_key='', properties=properties, body=message):

log.info("Successfully published this message to exchange \"" + self.exchange_name + "\": " + message)

else:

log.error("Failed to publish message \"" + message + "\" to the exchange \"" + self.exchange_name + "\"!")

raise Exception("Failed to publish message to the queue.")

In my unit tests I wanted to verify that an unsuccessful attempt to publish the message, would cause trigger an exception to be thrown. To test this, I needed basic_publish to return False. At first I found it tricky to patch instance methods such as basic_publish , but after coming across this excellent…

Read More Read More

Testing a Flask REST API using Factory Boy and an in-memory sqlite database

September 28, 2018 Kenneth Holter Comments 0 Comment

I’m developing a REST API using Python’s Flask microframework. As I wanted to do test driven development to ensure good test coverage, I needed a working testing framework. After checking out a few different options, I found one that works well for my use case. I thought I’d share my setup in case it may be of help to others. First, this is how I create the server:

# File: server/server.py
import connexion
from database import db
def create_app(config):
    connexion_app = connexion.FlaskApp(__name__)  # For stock Flask users, use something like "app <span class="pl-k">=</span> Flask(<span class="pl-c1">__name__</span>)" instead
    connexion_app.add_api('api.yaml', strict_validation=True)
    app = connexion_app.app
    app.config['SQLALCHEMY_DATABASE_URI'] = config.get_database_uri()
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    db.init_app(app)
    return app

# File: server/server.py

import connexion

from database import db

def create_app(config):

connexion_app = connexion.FlaskApp(__name__) # For stock Flask users, use something like "app <span class="pl-k">=</span> Flask(<span class="pl-c1">__name__</span>)" instead

connexion_app.add_api('api.yaml', strict_validation=True)

app = connexion_app.app

app.config['SQLALCHEMY_DATABASE_URI'] = config.get_database_uri()

app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db.init_app(app)

return app

Passing custom HTTP headers using Flask’s test_client

September 22, 2018 Kenneth Holter Comments 0 Comment

Passing custon HTTP headers using Flask’s test_client object should have been as easy as doing this:

headers = { 'API-KEY': 'myKey' }            
response = self.client.get('/users', follow_redirects=True, headers=headers)

1 2	headers = { 'API-KEY': 'myKey' } response = self.client.get('/users', follow_redirects=True, headers=headers)

For some reason this didn’t work in my Flask project. In case others have the same issue as I did, maybe the solution I found to work may work for others too:

import unittest
from server import create_app

class UsersTests(unittest.TestCase):

    def setUp(self):
        app = create_app()
        self.client = app.test_client()

    def test_custom_headers(self):
        self.client.environ_base['HTTP_API_KEY'] = "mySecretKey"
        self.client.environ_base['HTTP_ACCEPT'] = 'application/xml'
        response = self.client.get("/users", follow_redirects=True)

import unittest

from server import create_app

class UsersTests(unittest.TestCase):

def setUp(self):

app = create_app()

self.client = app.test_client()

def test_custom_headers(self):

self.client.environ_base['HTTP_API_KEY'] = "mySecretKey"

self.client.environ_base['HTTP_ACCEPT'] = 'application/xml'

response = self.client.get("/users", follow_redirects=True)

Mocking SQLAlchemy models in Python

September 22, 2018 Kenneth Holter Comments 0 Comment

I’m developing an API using Python’s Flask framework. The API uses SQLAlchemy, which means that database rows are made available as regular Python classes (referred to as database models) in my Python code. One of my models looks something like this:

class View_users(db.Model):
   __tablename__ = 'view_active_users'
   name = db.Column(db.String(80), unique=True, nullable=False)

class View_users(db.Model):

__tablename__ = 'view_active_users'

name = db.Column(db.String(80), unique=True, nullable=False)

My Data Access Object (DAO) for communicating with the database models looks like this:

# File: server/daos/users.py
from server.models.users import View_users
class UsersDAO():
    def get_data(self):
    return View_users.query.all()

# File: server/daos/users.py

from server.models.users import View_users

class UsersDAO():

def get_data(self):

return View_users.query.all()

While writing tests for the API, I wanted to mock out the database models, so that my tests wouldn’t depend on an actual…

Read More Read More

Alarm clock for kids

August 16, 2016 Kenneth Holter Comments 0 Comment

I’ve created a very simple Raspberry Pi project that I use to let my kid know if its daytime or nighttime. When I say goodnight and leave my kids bedroom, I trigger a simple script on my Raspberry Pi with a 7″ touchscreen attached. The screen will display a daytime image for a few minutes, before switching to a nighttime image. The nighttime image is an indication for my kid to put down her toys and go to bed. Over…

Read More Read More

Using the “Oboe” Javascript library to speed up data processing

February 13, 2016 Kenneth Holter Comments 0 Comment

One of our node.js applications at work was designed to call an internal rest service, cache the whole response, and iterate over the data to separate relevant data from non-relevant data, and write the relevant data to file. To speed things up a bit, we wanted to process, filter and write the data to file as it arrived. For this task, the excellent Javascript library “Oboe” came in handy:

var oboe = require('oboe');

var fs = require('fs');

var jsonFileStream = fs.createWriteStream('myfile.json')[...]
var acceptHeader = {
    Accept: 'application/json'
}
var requestHeader = {
    url: "http://rest.example.local/myapp",
    headers: acceptHeader
}
oboe(requestHeader)
    .node('{ref hostname}', function(item) {
        handleItem(node, write_item_to_file);
    })
    .done(function() {
        jsonFileStream.end();
    });

function handleItem(item, callback) {
    if (isInterestingData(item)) {
        callback(item);
    }
}

function isInterestingData(item) {...}

function write_item_to_file(item) {...}

var oboe = require('oboe');

var fs = require('fs');

var jsonFileStream = fs.createWriteStream('myfile.json')[...]

var acceptHeader = {

Accept: 'application/json'

}

var requestHeader = {

url: "http://rest.example.local/myapp",

headers: acceptHeader

}

oboe(requestHeader)

.node('{ref hostname}', function(item) {

handleItem(node, write_item_to_file);

})

.done(function() {

jsonFileStream.end();

});

function handleItem(item, callback) {

if (isInterestingData(item)) {

callback(item);

}

function isInterestingData(item) {...}

function write_item_to_file(item) {...}

Keywords: HTTP response streaming

Using Maven to build and run Java applications

March 10, 2015 Kenneth Holter Comments 0 Comment

Say you want to have Maven handle the dependencies for your java application containing the well known “public static void main” method. Create a Maven project, put your MyApplication.java file under the “src”-folder, and add something like this to you pom.xml:

<?xml version="1.0" encoding="UTF-8"?>
<build>
   <sourceDirectory>src</sourceDirectory>
   <resources>
      <resource>
         <directory>src</directory>
         <excludes>
            <exclude>**/*.java</exclude>
         </excludes>
      </resource>
   </resources>
   <plugins>
      <plugin>
         <artifactId>maven-compiler-plugin</artifactId>
         <version>3.1</version>
         <configuration>
            <source>1.7</source>
            <target>1.7</target>
         </configuration>
      </plugin>
      <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>exec-maven-plugin</artifactId>
         <version>1.3.2</version>
         <executions>
            <execution>
               <goals>
                  <goal>java</goal>
               </goals>
            </execution>
         </executions>
         <configuration>
            <includePluginDependencies>true</includePluginDependencies>
            <executable>java</executable>
            <mainClass>MyApplication</mainClass>
         </configuration>
      </plugin>
   </plugins>
</build>

<?xml version="1.0" encoding="UTF-8"?>

<build>

</excludes>

</resource>

</resources>

<artifactId>maven-compiler-plugin</artifactId>

</configuration>

</plugin>

<groupId>org.codehaus.mojo</groupId>

<artifactId>exec-maven-plugin</artifactId>

<goals>

</goals>

</execution>

</executions>

<mainClass>MyApplication</mainClass>

</configuration>

</plugin>

</plugins>

</build>

Now you can run your application using the exec-maven-plugin, like this:

mvn clean compile exec:java -Dexec.args="arg1 arg2 arg3"

1	mvn clean compile exec:java -Dexec.args="arg1 arg2 arg3"

“Løp til bussen” – Android-app’en som forteller deg om bussen din er forsinket

January 25, 2015 Kenneth Holter Comments 0 Comment

For å lære meg litt om Android-programmering jeg har laget en enkel app. App’en er svært enkel, og har som eneste formål å vise sanntidsinfo for din neste reise med buss, bane, tog, trikk eller båt. Utgangspunktet mitt var at jeg enkelt ønsket å finne ut om jeg måtte løpe til bussen eller ikke, for jeg har altfor mange ganger løpt til bussen tidlig om morgenn, for så å bli stående å vente på den fordi den er forsinket.

Simple script for dumping a MySQL database

October 20, 2014 Kenneth Holter Comments 0 Comment

As file system backups might fail to backup open files, it’s good practice to close any open files before running the backup. Closing a MySQL database file means you will have to stop the MySQL server itself, causing causing a database outage while the backup is running. But fortunately there’s a smarter way to backup you database – simple dump the entire database to a file (which is automatically closed when the dump is completed), and have your file system…

Read More Read More

Setting up Sublime Text as a node.js development environment

August 21, 2014 Kenneth Holter Comments 0 Comment

Sublime Text makes a great IDE for node.js development. First, install the node.js software and get your initial project up and running. Various tutorials on the web should get you going. When it’s time to run your app, configure the Sublime Text build system to build it and run it. To avoid having to start and stop your Node server manually, make use of the tool Nodemon – this tool will build and run your app automatically when you edit…

Read More Read More