A simple python sandbox for Celery and not

At the end of this post you will find the source code of a python sandbox that I write for have a general sandbox that work for all my celery tasks.
Anyway this is a general python code and you can use it also without Celery but here I will give you also some example to how to use it in easy and fast way with Celery.

The sandbox is wrote as a Context Manager (under suggestion of asksol 😉 ) with an entry point that will create the folders and an exit point that will clean it.
The class create a folder structure like this:


On the exit point only the taskpid folder will be erased, in this way you can launch several tasks of the same type at the same time
Let’s see how you can use it for you celery task.

In my personal opinion the best way to do it is to create a base class for all your tasks that extend the celery task class and inside it implement some general useful methods like the sandbox behaviour.
Following you can see an example of a general base class that use my sandbox context manager.

import settings
from celery.task import Task
from sandBox import SandBox

class toforgeTask(Task):

    def __call__(self, *args, **kwargs):
        """In celery task this function call the run method, here you can
        set some environment variable before the run of the task"""

        self.sandbox = SandBox(self.name, self.request.id,
        return self.run(*args, **kwargs)

    def after_return(self, status, retval, task_id, args, kwargs, einfo):
        #: If the sandbox is initialized we clean it
        if self.sandbox.isInitialized():

    def run(self):
        return "This is a Base Task Class not runnable"

As you can read on the comments the __call__ method is called before the task execution.
Here we call the constructor of our sanbox but we don’t initialize it because maybe not all our tasks needs sandbox 😉 .
For the same reason in the method after_return, called form celery at the end of a task whatever is the status (RETRY,FAILURE, ecc.), we will clean the sandbox only if we used it.
The settings module that you see in my general task is the django settings that I use for store conf vars from my task like the PATH_SANDBOX the radix folder for your sandbox.

Now you can write a task by extend the base task that we created and maybe it will look like this

from toforgeTask import toforgeTask

class sampleDownloader(toforgeTask):

    def run(self, album_id, **kwargs):
        self.logger.info("Download a test file %s " % (album_id))

        #: initialize tha sanbox

        #: download a file

        #: do some stuff with the file in the sandbox
        #: you can open files in the standard way just use
        #: the self.sandbox.path variable

        return """File downloaded, but you will not found it because at the
               end of the task the sandbox will be cleaned"""

Following you can see the source code of the protagonist of this post, the sandBox.py file.

import errno
import os
import shutil
import urllib2

class SandBox(object):
    """A simple sandbox context manager for easy create and clean of a
    sandbox folder plus some useful methods for file download and write

    Mauro Rocco

    def __init__(self, name, id, prefix, debug = False):

        #: name of the folder, commonly the name of the task
        self.name = name

        #: pid of the task, the name of the end folder where the sandbox
        #: writes the files
        self.id = id

        #: the base path for your sandbox
        self.prefix = prefix

        #: te complete path used in the context
        self.path = None

        #: debug option for print some output information
        self.debug = debug

    def __enter__(self):
        self.path = os.path.join(self.prefix, self.name, self.id)

        except OSError, exc:
            if exc.errno != errno.EEXIST:

        return self

    def __exit__(self, *exc_info):
        except OSError, exc:
            if exc.errno != errno.ENOENT:

    def _debugPrint(self,message):
        if self.debug:
            print message

    def _createFile(self, fileName, mode, override = False):
        if not self.isInitialized():

        if override:
            return open(os.path.join(self.path,fileName), mode)
        except IOError, exc:
            if exc.errno == errno.ENOENT:
                return open(os.path.join(self.path,fileName), mode)
        raise Exception("File already exist in sandbox: %s "
                        % (os.path.join(self.path,fileName)))

    def isInitialized(self):
        """Check if the sandbox is initialized"""
        if self.path:
            return True
        return False

    def getSandboxContent(self):
        """Return a list of all folders and files
        in the current sandbox folder"""
        if not self.isInitialized():
        tree = []
        for dirname, dirnames, filenames in os.walk(self.path):
            for subdirname in dirnames:
                tree.append(os.path.join(dirname, subdirname))
            for filename in filenames:
                tree.append(os.path.join(dirname, filename))
        return tree

    def writeFile(self,content, fileName, mode = "w", override = False):
        """write a file in the sandbox folder"""
        f = self._createFile(fileName, mode, override)

    def writeFileFromUrl(self,url,fileName, override = False, httpproxy=None,
                         ftpproxy=None, gopherproxy=None):
        """Download a file from the given url  and write it
         on the sandbox with the given name"""

        if httpproxy:

        if ftpproxy:

        if gopherproxy:

        outfile=self._createFile(fileName, "wb", override)

        proxy = urllib2.ProxyHandler(proxies)
        opener = urllib2.build_opener(proxy)

        instream = urllib2.urlopen(url)
        length = instream.info().getheader("Content-Length")

        if length==None:

        self._debugPrint("Downloading %s (%s bytes) ..."
                         % (instream.url, length))


        for line in instream:

            if length!="?":
                self._debugPrint("Downloading... %.02f/%.02f kb (%d%%)"
                                 % (fileName,

        self._debugPrint("File downloaded successful")

Enjoy 🙂