Category: Uncategorized

Recently, we published two Python packages to PyPI: bdrxml and bdrcmodels. No one else is using those packages, as far as I know, and it takes some effort to put them up there, but there are benefits from publishing them.

Putting a package on PyPI makes it easier for other code we package up to depend on bdrxml. For our indexing package, we can switch from this:

‘bdrxml @ https://github.com/Brown-University-Library/bdrxml/archive/v1.0a1.zip#sha1=5802ed82ee80a9627657cbb222fe9c056f73ad2c’,

to this:

‘bdrxml>=1.0’,

in setup.py, which is simpler. This also lets us using Python’s package version checking to not pin bdrxml to just one version, which is helpful when we embed the indexing package in another project that may use a different version of bdrxml.

Publishing these first two packages also gave us experience, which will help if we publish more packages to PyPI.

We recently had a couple cases where we wanted to dump the data out of a Django database. In the first case (“tracker”), we were shutting down a legacy application, but needed to preserve the data in a different form for users. In the second case (“deposits”), we were backing up some obsolete data before removing it from the database. We handled the processes in two different ways.

Tracker

For the tracker, we used an export script to extract the data. Here’s a modified version of the script:

def export_data():
    now = datetime.datetime.now()
    dir_name = 'data_%s_%s_%s' % (now.year, now.month, now.day)
    d = os.mkdir(dir_name)
    file_name = os.path.join(dir_name, 'tracker_items.dat')
    with open(file_name, 'wb') as f:
        f.write(u'\u241f'.join([
                    'project name',
                    'container identifier',
                    'container name',
                    'identifier',
                    'name',
                    'dimensions',
                    'note',
                    'create digital surrogate',
                    'qc digital surrogate',
                    'create metadata record',
                    'qc metadata record',
                    'create submission package']).encode('utf8'))
        f.write('\u241e'.encode('utf8'))
        for project in models.Project.objects.all():
            for container in project.container_set.all():
                print(container)
                for item in container.item_set.all():
                    data = u'\u241f'.join([
                        project.name.strip(),
                        container.identifier.strip(),
                        container.name.strip(),
                        item.identifier.strip(),
                        item.name.strip(),
                        item.dimensions.strip(),
                        item.note.strip()
                    ])
                    item_actions = u'\u241f'.join([str(item_action) for item_action in item.itemaction_set.all().order_by('id')])
                    line_data = u'%s\u241f%s\u241e' % (data, item_actions)
                    f.write(line_data.encode('utf8'))

As you can see, we looped through different Django models and pulled out fields, writing everything to a file. We used the Unicode Record and Unit Separators as delimiters. One advantage of using those is that your data can have commas, tabs, newlines, … and it won’t matter. You still don’t have to quote or escape anything.

Then we converted the data to a spreadsheet that users can view and search:

import openpyxl

workbook = openpyxl.Workbook()
worksheet = workbook.active

with open('tracker_items.dat', 'rb') as f:
    data = f.read()
    lines = data.decode('utf8').split('\u241e')
    print(len(lines))
    print(lines[0])
    print(lines[-1])
    for line in lines:
        fields = line.split('\u241f')
        worksheet.append(fields)
workbook.save('tracker_items.xlsx')

Deposits

For the deposits project, we just used the built-in Django dumpdata command:

python manage.py dumpdata -o data_20180727.dat

That output file could be used to load data back into a database if needed.

The BDR provides thumbnails for image and PDF objects. The thumbnail service is set up to check for a thumbnail in storage, then try to generate one from the IIIF image server, and fall back to an icon if needed. Thumbnails are cached by the thumbnail service for 30 days, up to a maximum of 5000 thumbnails. The maximum number of thumbnails was the limiting factor on the thumbnails in the cache – they weren’t being purged from the cache because they were older than 30 days, but because the cache filled up.

We recently had the need to purge some thumbnails, because we updated the images for some objects. We decided to update the thumbnail caching, so that the cache is timestamped. We already use our API to check for permissions on an object before displaying the thumbnail, and we added an API check for when the object in storage was last modified. If the object was modified more recently than the cache timestamp, the cache is stale and we grab an updated thumbnail. This should keep the thumbnail cache from serving stale images.

In the BDR, we can provide streaming versions for audio or video content, in addition to (or instead of) file download. We used Wowza for streaming content in the past, but now we use Panopto.

The process for getting content to Panopto has been manual: download the file from the BDR, upload it to Panopto, set the correct session name in Panopto, and associate the Panopto ID with the BDR record. I’ve been working on automating this process, though.

Here are the steps that the automation code performs:

• download audio or video file from BDR to a temporary file
• hit the Panopto API to upload file
  • create the session and upload in Panopto
  • use the Amazon S3 protocol to upload the file
  • mark the upload as complete so Panopto starts processing the file
• create a streaming object in the BDR with the correct Panopto session ID

We want to make sure that the process can handle large files without running out of memory or taxing the server too much. So, we stream the content to the temporary file in chunks. Then, when we upload the file to Panopto, we’d like to do that in chunks as well, so we’re never reading the whole file into memory – unfortunately, we’re currently running into an error with the multipart upload.

This automation will reduce the amount of manual work we do for streaming content, and could open the door to creating streaming objects automatically on request from non-BDR staff (or even users).

Many applications make calls to external services, or other services that are part of the application. Testing those HTTP calls can be challenging, but there are some different options available in Python.

Mocking

One option for testing your HTTP calls is to mock out your function that makes the HTTP call. This way, your function doesn’t make the HTTP call, since it’s replaced by a mock function that just returns whatever you want it to.

Here’s an example of mocking out your HTTP call:

import requests

class SomeClass:

  def __init__(self):
    self.data = self._fetch_data()

  def _fetch_data(self):
    r = requests.get('https://repository.library.brown.edu/api/collections/')
    return r.json()

  def get_collection_ids(self):
    return [c['id'] for c in self.data['collections']]

from unittest.mock import patch
MOCK_DATA = {'collections': [{'id': 1}, {'id': 2}]}

with patch.object(SomeClass, '_fetch_data', return_value=MOCK_DATA) as mock_method:
  thing = SomeClass()
  assert thing.get_collection_ids() == [1, 2]

Another mocking option is the responses package. Responses mocks out the requests library specifically, so if you’re using requests, you can tell the responses package what you want each requests call to return.

Here’s an example using the responses package (SomeClass is defined the same way as in the first example):

import responses
import json
MOCK_JSON_DATA = json.dumps({'collections': [{'id': 1}, {'id': 2}]})

@responses.activate
def test_some_class():
  responses.add(responses.GET,  'https://repository.library.brown.edu/api/collections/',
 body=MOCK_JSON_DATA,
 status=200,
 content_type='application/json'
 )
  thing = SomeClass()
  assert thing.get_collection_ids() == [1, 2]

test_some_class()

Record & Replay Data

A different type of solution is to use a package to record the responses from your HTTP calls, and then replay those responses automatically for you.

• VCR.py – VCR.py is a Python version of the Ruby VCR library, and it supports various HTTP clients, including requests.

Here’s a VCR.py example, again using SomeClass from the first example:

import vcr 
IDS = [674, 278, 280, 282, 719, 300, 715, 659, 468, 720, 716, 687, 286, 288, 290, 296, 298, 671, 733, 672, 334, 328, 622, 318, 330, 332, 625, 740, 626, 336, 340, 338, 725, 724, 342, 549, 284, 457, 344, 346, 370, 350, 656, 352, 354, 356, 358, 406, 663, 710, 624, 362, 721, 700, 661, 364, 660, 718, 744, 702, 688, 366, 667]

with vcr.use_cassette('vcr_cassettes/cassette.yaml'):
  thing = SomeClass()
  fetched_ids = thing.get_collection_ids()
  assert sorted(fetched_ids) == sorted(IDS)

• betamax – From the documentation: “Betamax is a VCR imitation for requests.” Note that it is more limited than VCR.py, since it only works for the requests package.

Here’s a betamax example (note: I modified the code in order to test it – maybe there’s a way to test the code with betamax without modifying it?):

import requests

class SomeClass:
    def __init__(self, session=None):
        self.data = self._fetch_data(session)

    def _fetch_data(self, session=None):
         if session:
             r = session.get('https://repository.library.brown.edu/api/collections/')
         else:
             r = requests.get('https://repository.library.brown.edu/api/collections/')
         return r.json()

    def get_collection_ids(self):
        return [c['id'] for c in self.data['collections']]


import betamax
CASSETTE_LIBRARY_DIR = 'betamax_cassettes'
IDS = [674, 278, 280, 282, 719, 300, 715, 659, 468, 720, 716, 687, 286, 288, 290, 296, 298, 671, 733, 672, 334, 328, 622, 318, 330, 332, 625, 740, 626, 336, 340, 338, 725, 724, 342, 549, 284, 457, 344, 346, 370, 350, 656, 352, 354, 356, 358, 406, 663, 710, 624, 362, 721, 700, 661, 364, 660, 718, 744, 702, 688, 366, 667]

session = requests.Session()
recorder = betamax.Betamax(
 session, cassette_library_dir=CASSETTE_LIBRARY_DIR
 )

with recorder.use_cassette('our-first-recorded-session', record='none'):
    thing = SomeClass(session)
    fetched_ids = thing.get_collection_ids()
    assert sorted(fetched_ids) == sorted(IDS)

Integration Test

Note that with all the solutions I listed above, it’s probably safest to cover the HTTP calls with an integration test that interacts with the real service, in addition to whatever you do in your unit tests.

Another possible solution is to test as much as possible with unit tests without testing the HTTP call, and then just rely on the integration test(s) to test the HTTP call. If you’ve constructed your application so that the HTTP call is only a small, isolated part of the code, this may be a reasonable option.

Here’s an example where the class fetches the data if needed, but the data can easily be put into the class for testing the rest of the functionality (without any mocking or external packages):

import requests

class SomeClass:

    def __init__(self):
        self._data = None

    @property
    def data(self):
        if not self._data:
            r = requests.get('https://repository.library.brown.edu/api/collections/')
            self._data = r.json()
        return self._data

    def get_collection_ids(self):
        return [c['id'] for c in self.data['collections']]


import json
MOCK_DATA = {'collections': [{'id': 1}, {'id': 2}]}

def test_some_class():
    thing = SomeClass()
    thing._data = MOCK_DATA
    assert thing.get_collection_ids() == [1, 2]

test_some_class()