SaltyCrane Blog — Notes on JavaScript and web development

How to use pip with crate.io

Date:  |  Tags: install_setup, python  |  5 Comments
Here's how to use pip with crate.io (in case pypi.python.org goes down):
$ pip install --index-url=https://simple.crate.io yolk
Or with logging to see what's happening:
$ pip install --log=my-pip-debug.log --index-url=https://simple.crate.io yolk

See also

How to run a Django local development server on a remote machine and access it in your browser on your local machine using SSH port forwarding

Date:  |  Modified:  |  Tags: django, linux, python, ssh, webdev  |  2 Comments

Here is how to run a Django local development server on a remote machine and access it in your browser on your local machine using SSH port forwarding. (This is useful if there is a firewall blocking access to the port of your Django local dev server (port 8000).

  1. On the local host, SSH to the remote host:
    $ ssh -v -L 9000:localhost:8000 [email protected]
  2. On the remote host, run the Django dev server:
    [email protected]:/path/to/my/django/project$ python manage.py runserver 0.0.0.0:8000
  3. On the local host, go to http://localhost:9000 in the browser

Note: The local port and the remote port can be the same (i.e. you can use 8000 instead of 9000). I just made them different to show which port is which.

Using LocalForward in your ~/.ssh/config

You can also achieve the same results by using the LocalForward in your ~/.ssh/config file:

Host myremote
  User eliot
  HostName my.remotehost.com
  LocalForward 9000 localhost:8000

Reference

http://magazine.redhat.com/2007/11/06/ssh-port-forwarding/

Testing HTTPS w/ Flask's development server using stunnel on Ubuntu

Date:  |  Tags: python, ubuntu, webdev  |  2 Comments

Our website is served over HTTPS. To more easily test certain issues (e.g. mixed mode content warnings, or Mapquest SSL tile servers), I wanted to access my Flask local development server over HTTPS. These two articles describe how to do this using stunnel: Testing HTTPS with Django's Development Server, Django Development Server with HTTPS. Using stunnel, you can hit pages on your Django/Flask local dev server over HTTPS instead of HTTP. Here is how I installed it on Ubuntu Precise 12.04:

  • Install SSL development files
    $ sudo apt-get install libssl-dev
  • Go to https://www.stunnel.org/downloads.html and download stunnel-4.54.tar.gz
  • Unpack, compile, install.
    $ tar xvf stunnel-4.54.tar.gz 
$ cd stunnel-4.54 
$ ./configure --prefix=/home/saltycrane/lib/stunnel-4.54 
$ make 
$ make install
    NOTE: the make install step asked me a number of questions and created a certificate file at /home/saltycrane/lib/stunnel-4.54/etc/stunnel/stunnel.pem. Accept all the defaults for the certificate information (accurate certificate information isn't needed for this application).
  • Create a stunnel configuration file, /home/saltycrane/lib/stunnel-4.54/etc/stunnel/dev_https:
    pid =
cert = /home/saltycrane/lib/stunnel-4.54/etc/stunnel/stunnel.pem
debug = 7
foreground = yes

[https]
accept = 7000
connect = 5000
  • Start stunnel:
    $ /home/saltycrane/lib/stunnel-4.54/bin/stunnel /home/saltycrane/lib/stunnel-4.54/etc/stunnel/dev_https
2012.10.17 17:40:52 LOG7[12468:140357811214080]: Clients allowed=500
2012.10.17 17:40:52 LOG5[12468:140357811214080]: stunnel 4.54 on x86_64-unknown-linux-gnu platform
2012.10.17 17:40:52 LOG5[12468:140357811214080]: Compiled/running with OpenSSL 1.0.1 14 Mar 2012
2012.10.17 17:40:52 LOG5[12468:140357811214080]: Threading:PTHREAD SSL:+ENGINE+OCSP Auth:none Sockets:POLL+IPv6
2012.10.17 17:40:52 LOG5[12468:140357811214080]: Reading configuration from file /home/saltycrane/lib/stunnel-4.54/etc/stunnel/dev_https
2012.10.17 17:40:52 LOG7[12468:140357811214080]: Compression not enabled
2012.10.17 17:40:52 LOG7[12468:140357811214080]: Snagged 64 random bytes from /home/saltycrane/.rnd
2012.10.17 17:40:52 LOG7[12468:140357811214080]: Wrote 1024 new random bytes to /home/saltycrane/.rnd
2012.10.17 17:40:52 LOG7[12468:140357811214080]: PRNG seeded successfully
2012.10.17 17:40:52 LOG6[12468:140357811214080]: Initializing service [https]
2012.10.17 17:40:52 LOG7[12468:140357811214080]: Certificate: /home/saltycrane/lib/stunnel-4.54/etc/stunnel/stunnel.pem
2012.10.17 17:40:52 LOG7[12468:140357811214080]: Certificate loaded
2012.10.17 17:40:52 LOG7[12468:140357811214080]: Key file: /home/saltycrane/lib/stunnel-4.54/etc/stunnel/stunnel.pem
2012.10.17 17:40:52 LOG7[12468:140357811214080]: Private key loaded
2012.10.17 17:40:52 LOG7[12468:140357811214080]: Using DH parameters from /home/saltycrane/lib/stunnel-4.54/etc/stunnel/stunnel.pem
2012.10.17 17:40:52 LOG7[12468:140357811214080]: DH initialized with 1024-bit key
2012.10.17 17:40:52 LOG7[12468:140357811214080]: ECDH initialized with curve prime256v1
2012.10.17 17:40:52 LOG7[12468:140357811214080]: SSL options set: 0x00000004
2012.10.17 17:40:52 LOG5[12468:140357811214080]: Configuration successful
2012.10.17 17:40:52 LOG7[12468:140357811214080]: Service [https] (FD=7) bound to 0.0.0.0:7000
2012.10.17 17:40:52 LOG7[12468:140357811214080]: No pid file being created
  • Start the python dev server:
    $ HTTPS=1 python bin/runserver.py 0.0.0.0 5000
  • Go to https://localhost:7000 in your browser

See also

Python data object motivated by a desire for a mutable namedtuple with default values

Date:  |  Modified:  |  Tags: datastructures, python  |  2 Comments

UPDATE 2016-08-12: Read Glyph's post and use the attrs library instead.

Reasons to use this instead of a namedtuple:

  • I want to change fields at a later time (mutability)
  • I want to specify a subset of the fields at instantiation and have the rest be set to a default value

Reasons to use this instead of a dict:

  • I want to explicitly name the fields in the object
  • I want to disallow setting fields that are not explicitly named*
  • I want to specify a subset of the fields at instantiation and have the rest be set to a default value
  • I want to use attribute style access (dot notation to access fields)

Reasons to use this instead of a regular Python class:

  • I don't want to duplicate field names in the __init__() method signature and when setting instance attributes of the same name.
  • I want to disallow setting fields that are not explicitly named*
  • I want to be able to easily convert the object to a dict or a tuple
  • I want to save memory

*Note: This Stack Overflow answer warns against using __slots__ for my goal of disallowing setting fields that are not explicitly named. It says metaclasses or decorators should be abused by us control freaks and static typing weenies instead. To comply with that advice, if you don't care about saving memory, __slots__ could be replaced with a non-special attribute, such as _fields. If that is done, attribute creation would no longer be limited.

See also:

class DataObject(object):
    """
    An object to hold data. Motivated by a desire for a mutable namedtuple with
    default values. To use, subclass, and define __slots__.

    The default default value is None. To set a default value other than None,
    set the `default_value` class variable.

    Example:
        class Jello(DataObject):
            default_value = 'no data'
            __slots__ = (
                'request_date',
                'source_id',
                'year',
                'group_id',
                'color',
                # ...
            )
    """
    __slots__ = ()
    default_value = None

    def __init__(self, *args, **kwargs):
        # Set default values
        for att in self.__slots__:
            setattr(self, att, self.default_value)

        # Set attributes passed in as arguments
        for k, v in zip(self.__slots__, args):
            setattr(self, k, v)
        for k, v in kwargs.items():
            setattr(self, k, v)

    def asdict(self):
        return dict(
            (att, getattr(self, att)) for att in self.__slots__)

    def astuple(self):
        return tuple(getattr(self, att) for att in self.__slots__)

    def __repr__(self):
        return '{}({})'.format(
            self.__class__.__name__,
            ', '.join('{}={}'.format(
                    att, repr(getattr(self, att))) for att in self.__slots__))

Tests:

import unittest

class DataObjectTestCase(unittest.TestCase):
    def test_instantiation_using_args(self):
        class MyData(DataObject):
            __slots__ = ('att1', 'att2')

        md = MyData('my attr 1', 'my attr 2')
        self.assertEqual(md.att1, 'my attr 1')
        self.assertEqual(md.att2, 'my attr 2')

    def test_instantiation_using_kwargs(self):
        class MyData(DataObject):
            __slots__ = ('att1', 'att2')

        md = MyData(att1='my attr 1', att2='my attr 2')
        self.assertEqual(md.att1, 'my attr 1')
        self.assertEqual(md.att2, 'my attr 2')

    def test_default_default_value(self):
        class MyData(DataObject):
            __slots__ = ('att1', 'att2')

        md = MyData(att1='my attr 1')
        self.assertEqual(md.att1, 'my attr 1')
        self.assertEqual(md.att2, None)

    def test_custom_default_value(self):
        class MyData(DataObject):
            default_value = 'custom default value'
            __slots__ = ('att1', 'att2')

        md = MyData(att1='my attr 1')
        self.assertEqual(md.att1, 'my attr 1')
        self.assertEqual(md.att2, 'custom default value')

    def test_set_value_after_instantiation(self):
        class MyData(DataObject):
            __slots__ = ('att1', 'att2')

        md = MyData(att1='my attr 1')
        self.assertEqual(md.att1, 'my attr 1')
        self.assertEqual(md.att2, None)
        md.att1 = 5
        md.att2 = 9
        self.assertEqual(md.att1, 5)
        self.assertEqual(md.att2, 9)

    def test_attribute_not_defined_in__slots__(self):
        class MyData(DataObject):
            __slots__ = ('att1', 'att2')

        with self.assertRaises(AttributeError):
            MyData(att3='my attr 3')

        with self.assertRaises(AttributeError):
            md = MyData()
            md.att3 = 45

    def test_asdict(self):
        class MyData(DataObject):
            __slots__ = ('att1', 'att2')

        md = MyData(att1='my attr 1', att2='my attr 2')
        self.assertEqual(
            md.asdict(), {'att1': 'my attr 1', 'att2': 'my attr 2'})

    def test_tuple(self):
        class MyData(DataObject):
            __slots__ = ('att1', 'att2')

        md = MyData(att1='my attr 1', att2='my attr 2')
        self.assertEqual(md.astuple(), ('my attr 1', 'my attr 2'))

    def test___repr__(self):
        class MyData(DataObject):
            __slots__ = ('att1', 'att2')

        md = MyData(att1='my attr 1', att2='my attr 2')
        self.assertEqual(repr(md), "MyData(att1='my attr 1', att2='my attr 2')")

Note: previously, I included the following method in the class. However, this is not necessary. If __slots__ is defined in DataObject and the subclass, any attribute not in __slots__ will automatically raise an AttributeError.

#     def __setattr__(self, name, value):
#         if name not in self.__slots__:
#             raise AttributeError("%s is not a valid attribute in %s" % (
#                     name, self.__class__.__name__))
#         super(DataObject, self).__setattr__(name, value)

Setting up a Linux DVR w/ MythTV, Ubuntu 12.04, and a Hauppauge WinTV-HVR 1250 TV tuner card

Date:  |  Modified:  |  Tags: hardware, linux  |  6 Comments

Setting up MythTV involves a little pain, but once it's set up, it's pretty great. And you don't have to spend lots of money on a DVR from the cable company. With my modest hardware specs, playback is smooth and clear, however Picture in Picture is too jittery to be useful. Here's what I did to get my MythTV DVR running on my Ubuntu machine.

Parameters

Install the Hauppauge WinTV-HVR 1250 TV tuner card

Put the card in the computer. Connect the TV antenna to the card.

Check the TV tuner card is recognized

Ubuntu 12.04 includes drivers for the Hauppauge 1250 TV tuner card, so I did not need to install any drivers.

Update 2016-10-16:On Ubuntu 16.04, look in /var/log/syslog instead of /var/log/dmesg.

$ cat /var/log/dmesg
[   15.211985] cx23885 driver version 0.0.3 loaded
[   15.214279] cx23885 0000:03:00.0: PCI INT A -> GSI 17 (level, low) -> IRQ 17
[   15.214492] CORE cx23885[0]: subsystem: 0070:2259, board: Hauppauge WinTV-HVR1255 [card=20,autodetected]
[   15.214600] IR NEC protocol handler initialized
[   15.230936] IR RC5(x) protocol handler initialized
[   15.235576] MCE: In-kernel MCE decoding enabled.
[   15.237132] IR RC6 protocol handler initialized
[   15.237703] EDAC MC: Ver: 2.1.0
[   15.238256] AMD64 EDAC driver v3.4.0
[   15.242493] IR JVC protocol handler initialized
[   15.246743] IR Sony protocol handler initialized
[   15.250908] IR MCE Keyboard/mouse protocol handler initialized
[   15.256862] lirc_dev: IR Remote Control driver registered, major 250 
[   15.257125] IR LIRC bridge handler initialized
[   15.284735] lp0: using parport0 (interrupt-driven).
[   15.361892] tveeprom 0-0050: Hauppauge model 22111, rev E2F5, serial# 8323201
[   15.361895] tveeprom 0-0050: MAC address is 00:0d:fe:7f:00:81
[   15.361897] tveeprom 0-0050: tuner model is NXP 18271C2 (idx 155, type 54)
[   15.361899] tveeprom 0-0050: TV standards NTSC(M) ATSC/DVB Digital (eeprom 0x88)
[   15.361901] tveeprom 0-0050: audio processor is CX23888 (idx 40)
[   15.361903] tveeprom 0-0050: decoder processor is CX23888 (idx 34)
[   15.361904] tveeprom 0-0050: has no radio, has IR receiver, has no IR transmitter
[   15.361906] cx23885[0]: hauppauge eeprom: model=22111
[   15.361909] cx23885_dvb_register() allocating 1 frontend(s)

Install MythTV

$ sudo apt-get install mythtv

Set up the MythTV backend

Run mythtv-setup to select your TV tuner card and scan for channels.

$ mythtv-setup

Click "Yes" to add your user to the "mythtv" group.

Click "Yes" to restart your login session.

Change the following options:

  • 2. Capture cards -> (New capture card) -> Card type: DVB DTV capture card (v3.x) -> Finish
  • 4. Video sources -> (New video source) -> Video source name: FOOBAR, Listings grabber: North America (SchedulesDirect.org) (Internal), User ID: blank, Pass: blank
  • 5. Input connections -> [DVB: /dev/dvb/adapter0/frontend0] -> Video source: FOOBAR -> Scan for channels
    Note: if you see the error "Failed to open the card", make sure the mythv-backend is actually stopped. The GUI said it would stop the backend but it was still running.

After running mythtv-setup, it will ask you if you want to start the backend. Select yes to start the backend. It will also ask you if you want to run mythfilldatabase. Select yes to run mythfilldatabase. This may take a while.

Ensure mythv backend is running

After running mythtv-setup, the mythtv backend should start running.

To check that the backend is running, run:

$ ps -ef | grep myth

If the mythtv backend is not running, start it using the following command:

$ sudo service mythtv-backend start

Troubleshooting mythbackend

If mythbackend doesn't stay running, there may be some configuration that is broken. Check /var/log/syslog. If that does not have enough information, run the backend with the --verbose option:

$ mythbackend --verbose

Run the MythTV frontend

$ mythfrontend

Some keyboard shortcuts

  • P - pause/play
  • SPACE - set/clear bookmark
  • LEFT/RIGHT ARROW - skip back/forward
  • M - menu
  • D - delete

Other stuff

  • You may want to change the theme. I chose the TintedGlass 2.43 theme.
  • To get schedule information, I ended up signing up for a membership at www.schedulesdirect.org. It is $25/year (or ~$2/month). It seems to be the recommended way to get schedule information.

How run mythfrontend on another Ubuntu laptop connected to your LAN (Added 2013-06-07)


Since MythTV has a flexible client/server architecture, you can run the MythTV backend server on one machine and access it from multiple other machines running a Mythtv frontend. These steps assume the remote frontend is running on a laptop with Ubuntu 12.04 and it is connected to your local network (LAN) (not through the internet (though that is possible.).)

UPDATE: Playing 1080p HD content over my $30 Belkin G wireless router (rated at 54 Mbps) had occasional stalls in the playback. Repositioning my router helped, but after a couple days, I decided to order a Netgear N600 Wireless-N Dual Band Router. Hopefully this will solve my problem.

On the Mythtv backend server configured above:

  • Determine the IP address of the Mythtv backend server by running ifconfig
    $ ifconfig
    For me, it is 192.168.2.2. This will be used in the steps below.
  • Follow the instructions here: http://www.mythtv.org/wiki/Mythfrontend
    • Edit /etc/mysql/my.cnf so the the bind-address line is commented out (on Ubuntu 16.04 edit /etc/mysql/mysql.conf.d/mysqld.cnf):
      #bind-address 127.0.0.1
    • Allow remote users access to the database. Note: replace "mypassword" with the value found in ~/.mythtv/mysql.txt.
      $ mysql -u root
mysql> grant all on mythconverg.* to 'mythtv'@'%' identified by 'mypassword';
mysql> flush privileges;
mysql> exit
    • Restart mysql server:
      $ sudo service mysql restart
  • Ensure mythbackend is not using 127.0.0.1.
    • Run mythtv-setup:
      $ mythtv-setup
    • Change the IP address from 127.0.0.1 to 192.168.2.2 (or IP address you determined from above.)

On the laptop:

  • $ sudo apt-get install mythtv-frontend
  • click "yes" to be added to the mythtv group
  • click "yes" to restart your session
  • click "OK" to the msg about logging out of your session
  • logout and login again
  • Run mythfrontend
    $ mythfrontend
  • For the hostname: enter the IP address of the Mythtv server. For me it is 192.168.2.2.
  • Enter the Mysql password. This can be found in ~/.mythtv/mysql.txt (or ~/.mythtv/config.xml or /etc/mythtv/config.xml on newer versions) on the Mythtv server machine. Or you can check the settings of the mythfrontend running on the server machine.

How to watch your recorded videos on your Android phone over the internet


  • This method uses the MythTV Services API
  • PC: Set up a SSH server on your MythTV backend server
  • PC: Get the external IP address of your MythTV backend server
    $ curl http://ifconfig.me
111.222.333.444
  • Android: Install Connectbot on your Android phone and enable port forwarding of 6544. For more info see: http://parker1.co.uk/mythtv_ssh.php
    • Android: Using Connectbot, connect to your MythTV server using the IP address from above (111.222.333.444)
    • Android: Menu -> Port Forwards -> Menu -> Add port forward:
      • Nickname: mythtv
      • Type: local
      • Source port: 6544
      • Destination: localhost:6544
    • Android: Disconnect and reconnect
  • Android: Install and set up MythTV Android Frontend
    • Android: Touch the settings icon -> Away Profiles
      • Name: Away
      • MythTV Master Backend Address: http://localhost:6544/
      Save
    • Android: Away -> Recordings -> Select a show to watch -> watch it

See also:

How to control your DVR from your Android phone

  • Configure your MythTv Frontend on your PC:
    • Setup -> General -> Hit "Next" 6 times -> Check "Enable Network Remote Control"
    • Setup -> Appearance -> Hit "Next" 3 times and
      • Check "Enable LCD device"
      • Check "Display time"
      • Check "Display menus"
      • Check "Display music arstist and title"
      • Check "Display channel information"
  • Install MythDroid on your Android phone
  • Install MDD on your PC
    • Install libimlib2
      $ sudo apt-get install libimlib2-dev
    • Download MDD
      $ wget http://mythdroid.googlecode.com/files/mdd-0.6.2.tgz
    • Install MDD
      $ tar xvf mdd-0.6.2.tgz 
$ cd mdd 
$ perl Build.PL
      Type "y" because you are running this on the PC that runs your MythTv frontend
      $ ./Build installdeps
      Hit ENTER to accept all the defaults
      $ ./Build test 
$ sudo ./Build install
      Type "y" to stop mythfrontend. Then start it again
      $ mythfrontend

How to restore an old database on a fresh install Added 2016-10-16

See https://www.mythtv.org/wiki/Database_Backup_and_Restore

$ wget https://raw.githubusercontent.com/MythTV/mythtv/master/mythtv/programs/scripts/database/mythconverg_restore.pl 
$ chmod a+x mythconverg_restore.pl 
$  ./mythconverg_restore.pl --directory /mnt/hdd2/mythtv/db_backups --filename mythconverg-1317-20161008080149.sql.gz --drop_database --create_database 
$ # if there has been a database schema change, you will need to run mythtv-setup

Help / References

How to prevent nose (unittest) from using the docstring when verbosity >= 2

Date:  |  Tags: python  |  1 Comment

Some of our Python unit tests have docstrings. I find it annoying that, when using a verbosity level >= 2, nose prints the docstring instead of the class name and method name. Here's a hack to prevent it from doing that: Add a shortDescription() method to the test case class that returns None.

Here is an example of normal behavior:

import unittest

class MyTestCase(unittest.TestCase):
    def test_with_docstring(self):
        """Test that something does something
        """

    def test_without_docstring(self):
        pass

$ nosetests --verbosity=2 tmp.py
Test that something does something ... ok
test_without_docstring (tmp.MyTestCase) ... ok

Here is an example with the hack to prevent printing the docstring:

import unittest

class MyTestCase(unittest.TestCase):
    def shortDescription(self):
        return None

    def test_with_docstring(self):
        """Test that something does something
        """

    def test_without_docstring(self):
        pass

$ nosetests --verbosity=2 tmp.py
test_with_docstring (tmp.MyTestCase) ... ok
test_without_docstring (tmp.MyTestCase) ... ok

Hack to share & sync Google contacts between Android phones

Date:  |  Tags: android  |  1 Comment

I want to share and sync (in real time) Google (Gmail) contacts with my wife on our Android 2.3.6 Gingerbread phones. Google does not make this easy to do. Here's the best solution I could come up with (ref whitenack on androidcentral). (Note: these are not our real email addresses.)

  • This contact list resides only on the [email protected] account.
  • Contacts are removed from the "My Contacts" group and instead stored in groups called "Angela" and/or "Eliot". For shared contacts, the contact is in both groups. (Contact groups are like tags. A contact can be in multiple groups at the same time.)
  • Contacts in the "Angela" group show up on Angela's phone and contacts in the "Eliot" group show up on Eliot's phone. Contacts in both groups show up in both phones.
  • On Angela's phone, add the [email protected] account and check the box for syncing Contacts and uncheck the box for syncing Contacts from the [email protected] account.
  • On Angela's phone, check the box for displaying the groups "My Contacts" and "Angela" under the [email protected] account and uncheck all the boxes for displaying contacts on the [email protected] account.
  • On Eliot's phone, check the box for displaying the groups "My Contacts" and "Eliot" under the [email protected] account
  • On both phones, set the account used for creating new contacts to [email protected]: Contacts -> More -> Settings -> Contact storage -> Select the [email protected] account
  • When a *new* contact is added on either of the phones, it will be added to the "My Contacts" group on the [email protected] account. These contacts later need to be moved to the "Angela" and/or "Eliot" groups from the browser while signed in to the [email protected] account.
  • The [email protected] account will not be able to view, add, or edit contacts from the browser (Gmail).

We are able to share and sync contacts in real time, however there are annoyances. The main problem is that contact list lives under one account, so it is not available to the secondary user (my wife) when she is using Gmail or wants to manage contacts in the browser. A second minor annoyance is that our Android phones don't allow us to assign a contact to a group, so all new contacts added from our phones will be added to the generic "My Contacts" group and need to be categorized later from the browser.

I also tried the free Google Apps because it has Contact sharing. However, I could not figure out how to get shared contacts to show up in our phones.

Will upgrading to Android 4.0 ICS help?

Test coverage with nose and coverage.py

Date:  |  Modified:  |  Tags: python, testing  |  4 Comments

It's fun to use nose + coverage.py to show my progress as I write tests. Seeing the bar next to my code change from red to green makes me happy. 100% test coverage does not mean tests are complete. For example, a boolean OR'ed conditional expression may not test all conditions even though the line is marked as covered. Other limitations are discussed here: Flaws in coverage measurement. However, good test coverage is at least a step towards having a good test suite.

Install nose and coverage.py

Activate your virtualenv and pip install nose and coverage.

$ pip install nose 
$ pip install coverage

Run it

Here is the command line I use to run the tests. --with-coverage enables the nose-coverage plugin to check test coverage. --cover-erase erases coverage test results from a previous run. --cover-package specifies which Python package to analyze. Specifiy the package as you would using an import (e.g. dp.blueprints.info.views). If --cover-package is not specified, it will analyze everything. --cover-html enables pretty HTML coverage reports. This example is for the flask-encryptedsession tests.

$ nosetests --with-coverage --cover-erase --cover-package=flask_encryptedsession --cover-html
..........
Name                                      Stmts   Miss  Cover   Missing
-----------------------------------------------------------------------
flask_encryptedsession                        0      0   100%   
flask_encryptedsession.encryptedcookie       41      1    98%   176
flask_encryptedsession.encryptedsession      35      1    97%   75
-----------------------------------------------------------------------
TOTAL                                        76      2    97%   
----------------------------------------------------------------------
Ran 10 tests in 0.188s

OK

Display the HTML report

$ firefox cover/index.html

Get branch coverage

Branch coverage is useful for checking "if" statements without an explicit "else" in the code. I had to install the development version of nose to use this feature: As of version 1.2.0, this feature is available.

$ pip install https://github.com/nose-devs/nose/tarball/master 

$ nosetests --cover-branches --with-coverage --cover-erase --cover-package=flask_encryptedsession --cover-html 
..........
Name                                      Stmts   Miss Branch BrPart  Cover   Missing
-------------------------------------------------------------------------------------
flask_encryptedsession                        0      0      0      0   100%   
flask_encryptedsession.encryptedcookie       41      1     12      1    96%   176
flask_encryptedsession.encryptedsession      35      1      4      1    95%   75
-------------------------------------------------------------------------------------
TOTAL                                        76      2     16      2    96%   
----------------------------------------------------------------------
Ran 10 tests in 0.234s

OK

Pycon 2012 talks that I saw that I enjoyed

Date:  |  Tags: python

The Pycon 2012 videos are up at pyvideo.org. Here are some of the talks I enjoyed that I saw. I know I probably missed some great talks so I will try to watch more online. Let me know if there are some that I should not miss.

Favorite talk of the conference

Other great talks (in chronological order)

A unique Python redis-based queue with delay

Date:  |  Tags: concurrency, python  |  4 Comments

This is a simple Redis-based queue. Two features that I needed were uniqueness (i.e. if an item exists in the queue already, it won't be added again) and a delay, like beanstalkd, where an item must wait a specified time before it can be popped from the queue. There are a number of other Redis-based queues that have many more features but I didn't see one that had these two features together. This 50-line class works for my needs. It may or may not work for you. Feel free to copy this and build on it.

Note: I wrote this in May 2010. I ended up using this solution after trying out beanstalkd and Gearman.

Install

Install on Ubuntu 10.10 Maverick

  • Install the redis server
    $ sudo apt-get install redis-server
  • Install the python redis client
    $ pip install redis
  • Default conf file: /etc/redis/redis.conf
    Default log file: /var/log/redis/redis-server.log
    Default db dir: /var/lib/redis
    Stop redis server: sudo /etc/init.d/redis-server stop
    Start redis server: sudo /etc/init.d/redis-server start

Redis commands used

The queue is based on the redis sorted set data type and uses the following commands:

  • ZADD - Add members to a sorted set, or update its score if it already exists
  • ZRANGEBYSCORE - Return a range of members in a sorted set, by score
  • ZREM - Remove one or more members from a sorted set

Code

import time
import redis


REDIS_ADDRESS = '127.0.0.1'


class UniqueMessageQueueWithDelay(object):
    """A message queue based on the Redis sorted set data type. Duplicate items
    in the queue are not allowed. When a duplicate item is added to the queue,
    the new item is added, and the old duplicate item is removed. A delay may be
    specified when adding items to the queue. Items will only be popped after
    the delay has passed. Pop() is non-blocking, so polling must be used. The
    name of the queue == the Redis key for the sorted set.
    """
    def __init__(self, name):
        self.name = name
        self.redis = redis.Redis(REDIS_ADDRESS)

    def add(self, data, delay=0):
        """Add an item to the queue. delay is in seconds.
        """
        score = time.time() + delay
        self.redis.zadd(self.name, data, score)
        debug('Added %.1f, %s' % (score, data))

    def pop(self):
        """Pop one item from the front of the queue. Items are popped only if
        the delay specified in the add() has passed. Return False if no items
        are available.
        """
        min_score = 0
        max_score = time.time()
        result = self.redis.zrangebyscore(
            self.name, min_score, max_score, start=0, num=1, withscores=False)
        if result == None:
            return False
        if len(result) == 1:
            debug('Popped %s' % result[0])
            return result[0]
        else:
            return False

    def remove(self, data):
        return self.redis.zrem(self.name, data)


def debug(msg):
    print msg


def test_queue():
    u = UniqueMessageQueueWithDelay('myqueue')

    # add items to the queue
    for i in [0, 1, 2, 3, 4, 0, 1]:
        data = 'Item %d' % i
        delay = 5
        u.add(data, delay)
        time.sleep(0.1)

    # get items from the queue
    while True:
        print
        result = u.pop()
        print result
        if result != False:
            u.remove(result)
        time.sleep(1)


if __name__ == '__main__':
    test_queue()

Results:

Added 1320773851.8, Item 0
Added 1320773851.9, Item 1
Added 1320773852.0, Item 2
Added 1320773852.1, Item 3
Added 1320773852.2, Item 4
Added 1320773852.3, Item 0
Added 1320773852.4, Item 1

False

False

False

False

False

Popped Item 2
Item 2

Popped Item 3
Item 3

Popped Item 4
Item 4

Popped Item 0
Item 0

Popped Item 1
Item 1

False

False

False
^CTraceback (most recent call last):
  File "umqwdredisqueue.py", line 102, in 
    test_queue()
  File "umqwdredisqueue.py", line 98, in test_queue
    time.sleep(1)
KeyboardInterrupt

See also

Older Posts
Newer Posts