diff --git a/lib/galaxy/datatypes/assembly.py b/lib/galaxy/datatypes/assembly.py
index 4f047e865fa..e8ec95525cf 100644
--- a/lib/galaxy/datatypes/assembly.py
+++ b/lib/galaxy/datatypes/assembly.py
@@ -3,8 +3,9 @@ velvet datatypes
 James E Johnson - University of Minnesota
 for velvet assembler tool in galaxy
 """
+from __future__ import absolute_import
 
-import data
+from galaxy.datatypes import data
 import logging
 import os
 import re
diff --git a/lib/galaxy/datatypes/chrominfo.py b/lib/galaxy/datatypes/chrominfo.py
index edee9c79f32..73ec75a40fc 100644
--- a/lib/galaxy/datatypes/chrominfo.py
+++ b/lib/galaxy/datatypes/chrominfo.py
@@ -1,9 +1,11 @@
-from tabular import Tabular
-from galaxy.datatypes import metadata
+from __future__ import absolute_import
+
+import galaxy.datatypes.tabular
+import galaxy.datatypes.metadata
 from galaxy.datatypes.metadata import MetadataElement
 
 
-class ChromInfo( Tabular ):
+class ChromInfo( galaxy.datatypes.tabular.Tabular ):
     file_ext = "len"
-    MetadataElement( name="chrom", default=1, desc="Chrom column", param=metadata.ColumnParameter )
-    MetadataElement( name="length", default=2, desc="Length column", param=metadata.ColumnParameter )
+    MetadataElement( name="chrom", default=1, desc="Chrom column", param=galaxy.datatypes.metadata.ColumnParameter )
+    MetadataElement( name="length", default=2, desc="Length column", param=galaxy.datatypes.metadata.ColumnParameter )
diff --git a/lib/galaxy/datatypes/converters/wiggle_to_simple_converter.py b/lib/galaxy/datatypes/converters/wiggle_to_simple_converter.py
index a409745aff8..b41c27b0f98 100644
--- a/lib/galaxy/datatypes/converters/wiggle_to_simple_converter.py
+++ b/lib/galaxy/datatypes/converters/wiggle_to_simple_converter.py
@@ -10,7 +10,7 @@ import sys
 
 import bx.wiggle
 
-from galaxy.tools.exception_handling import UCSCOutWrapper, UCSCLimitException
+from galaxy.util.ucsc import UCSCOutWrapper, UCSCLimitException
 
 
 def stop_err( msg ):
diff --git a/lib/galaxy/datatypes/data.py b/lib/galaxy/datatypes/data.py
index f74227c47e5..af276616b32 100644
--- a/lib/galaxy/datatypes/data.py
+++ b/lib/galaxy/datatypes/data.py
@@ -1,3 +1,4 @@
+from __future__ import absolute_import
 import logging
 import mimetypes
 import os
@@ -7,7 +8,7 @@ import zipfile
 from cgi import escape
 from inspect import isclass
 
-import metadata
+from . import metadata
 from galaxy import util
 from galaxy.datatypes.metadata import MetadataElement  # import directly to maintain ease of use in Datatype class definitions
 from galaxy.util import inflector
@@ -15,7 +16,7 @@ from galaxy.util.bunch import Bunch
 from galaxy.util.odict import odict
 from galaxy.util.sanitize_html import sanitize_html
 
-import dataproviders
+from . import dataproviders
 
 import paste
 
@@ -58,7 +59,7 @@ class Data( object ):
     >>> DataTest.metadata_spec.test.desc
     'test'
     >>> type( DataTest.metadata_spec.test.param )
-    <class 'galaxy.datatypes.metadata.MetadataParameter'>
+    <class 'galaxy.model.metadata.MetadataParameter'>
 
     """
     edam_format = "format_1915"
diff --git a/lib/galaxy/datatypes/metadata.py b/lib/galaxy/datatypes/metadata.py
index ec83adb1115..db4ba5a3caa 100644
--- a/lib/galaxy/datatypes/metadata.py
+++ b/lib/galaxy/datatypes/metadata.py
@@ -1,874 +1,46 @@
-"""
-Galaxy Metadata
-
+""" Expose the model metadata module as a datatype module also,
+allowing it to live in galaxy.model means the model module doesn't
+have any dependencies on th datatypes module. This module will need
+to remain here for datatypes living in the tool shed so we might as
+well keep and use this interface from the datatypes module.
 """
 
-import copy
-import cPickle
-import json
-import os
-import shutil
-import sys
-import tempfile
-import weakref
-
-from os.path import abspath
-
-from sqlalchemy.orm import object_session
-
-import galaxy.model
-from galaxy.util import listify
-from galaxy.util.object_wrapper import sanitize_lists_to_string
-from galaxy.util import stringify_dictionary_keys
-from galaxy.util import string_as_bool
-from galaxy.util import in_directory
-from galaxy.util.json import safe_dumps
-from galaxy.util.odict import odict
-from galaxy.web import form_builder
-
-import logging
-log = logging.getLogger(__name__)
-
-STATEMENTS = "__galaxy_statements__"  # this is the name of the property in a Datatype class where new metadata spec element Statements are stored
-
-
-class Statement( object ):
-    """
-    This class inserts its target into a list in the surrounding
-    class.  the data.Data class has a metaclass which executes these
-    statements.  This is how we shove the metadata element spec into
-    the class.
-    """
-    def __init__( self, target ):
-        self.target = target
-
-    def __call__( self, *args, **kwargs ):
-        # get the locals dictionary of the frame object one down in the call stack (i.e. the Datatype class calling MetadataElement)
-        class_locals = sys._getframe( 1 ).f_locals
-        # get and set '__galaxy_statments__' to an empty list if not in locals dict
-        statements = class_locals.setdefault( STATEMENTS, [] )
-        # add Statement containing info to populate a MetadataElementSpec
-        statements.append( ( self, args, kwargs ) )
-
-    @classmethod
-    def process( cls, element ):
-        for statement, args, kwargs in getattr( element, STATEMENTS, [] ):
-            statement.target( element, *args, **kwargs )  # statement.target is MetadataElementSpec, element is a Datatype class
-
-
-class MetadataCollection( object ):
-    """
-    MetadataCollection is not a collection at all, but rather a proxy
-    to the real metadata which is stored as a Dictionary. This class
-    handles processing the metadata elements when they are set and
-    retrieved, returning default values in cases when metadata is not set.
-    """
-    def __init__(self, parent ):
-        self.parent = parent
-        # initialize dict if needed
-        if self.parent._metadata is None:
-            self.parent._metadata = {}
-
-    def get_parent( self ):
-        if "_parent" in self.__dict__:
-            return self.__dict__["_parent"]()
-        return None
-
-    def set_parent( self, parent ):
-        # use weakref to prevent a circular reference interfering with garbage
-        # collection: hda/lda (parent) <--> MetadataCollection (self) ; needs to be
-        # hashable, so cannot use proxy.
-        self.__dict__["_parent"] = weakref.ref( parent )
-    parent = property( get_parent, set_parent )
-
-    @property
-    def spec( self ):
-        return self.parent.datatype.metadata_spec
-
-    def __iter__( self ):
-        return self.parent._metadata.__iter__()
-
-    def get( self, key, default=None ):
-        try:
-            return self.__getattr__( key ) or default
-        except:
-            return default
-
-    def items(self):
-        return iter( [ ( k, self.get( k ) ) for k in self.spec.iterkeys() ] )
-
-    def __str__(self):
-        return dict( self.items() ).__str__()
-
-    def __nonzero__( self ):
-        return bool( self.parent._metadata )
-
-    def __getattr__( self, name ):
-        if name in self.spec:
-            if name in self.parent._metadata:
-                return self.spec[name].wrap( self.parent._metadata[name], object_session( self.parent ) )
-            return self.spec[name].wrap( self.spec[name].default, object_session( self.parent ) )
-        if name in self.parent._metadata:
-            return self.parent._metadata[name]
-
-    def __setattr__( self, name, value ):
-        if name == "parent":
-            return self.set_parent( value )
-        else:
-            if name in self.spec:
-                self.parent._metadata[name] = self.spec[name].unwrap( value )
-            else:
-                self.parent._metadata[name] = value
-
-    def remove_key( self, name ):
-        if name in self.parent._metadata:
-            del self.parent._metadata[name]
-        else:
-            log.info( "Attempted to delete invalid key '%s' from MetadataCollection" % name )
-
-    def element_is_set( self, name ):
-        return bool( self.parent._metadata.get( name, False ) )
-
-    def get_html_by_name( self, name, **kwd ):
-        if name in self.spec:
-            rval = self.spec[name].param.get_html( value=getattr( self, name ), context=self, **kwd )
-            if rval is None:
-                return self.spec[name].no_value
-            return rval
-
-    def make_dict_copy( self, to_copy ):
-        """Makes a deep copy of input iterable to_copy according to self.spec"""
-        rval = {}
-        for key, value in to_copy.items():
-            if key in self.spec:
-                rval[key] = self.spec[key].param.make_copy( value, target_context=self, source_context=to_copy )
-        return rval
-
-    def from_JSON_dict( self, filename=None, path_rewriter=None, json_dict=None ):
-        dataset = self.parent
-        if filename is not None:
-            log.debug( 'loading metadata from file for: %s %s' % ( dataset.__class__.__name__, dataset.id ) )
-            JSONified_dict = json.load( open( filename ) )
-        elif json_dict is not None:
-            log.debug( 'loading metadata from dict for: %s %s' % ( dataset.__class__.__name__, dataset.id ) )
-            if isinstance( json_dict, basestring ):
-                JSONified_dict = json.loads( json_dict )
-            elif isinstance( json_dict, dict ):
-                JSONified_dict = json_dict
-            else:
-                raise ValueError( "json_dict must be either a dictionary or a string, got %s." % ( type( json_dict ) ) )
-        else:
-            raise ValueError( "You must provide either a filename or a json_dict" )
-        for name, spec in self.spec.items():
-            if name in JSONified_dict:
-                from_ext_kwds = {}
-                external_value = JSONified_dict[ name ]
-                param = spec.param
-                if isinstance( param, FileParameter ):
-                    from_ext_kwds[ 'path_rewriter' ] = path_rewriter
-                dataset._metadata[ name ] = param.from_external_value( external_value, dataset, **from_ext_kwds )
-            elif name in dataset._metadata:
-                # if the metadata value is not found in our externally set metadata but it has a value in the 'old'
-                # metadata associated with our dataset, we'll delete it from our dataset's metadata dict
-                del dataset._metadata[ name ]
-        if '__extension__' in JSONified_dict:
-            dataset.extension = JSONified_dict['__extension__']
-
-    def to_JSON_dict( self, filename=None ):
-        # galaxy.model.customtypes.json_encoder.encode()
-        meta_dict = {}
-        dataset_meta_dict = self.parent._metadata
-        for name, spec in self.spec.items():
-            if name in dataset_meta_dict:
-                meta_dict[ name ] = spec.param.to_external_value( dataset_meta_dict[ name ] )
-        if '__extension__' in dataset_meta_dict:
-            meta_dict[ '__extension__' ] = dataset_meta_dict['__extension__']
-        if filename is None:
-            return json.dumps( meta_dict )
-        json.dump( meta_dict, open( filename, 'wb+' ) )
-
-    def __getstate__( self ):
-        # cannot pickle a weakref item (self._parent), when
-        # data._metadata_collection is None, it will be recreated on demand
-        return None
-
-
-class MetadataSpecCollection( odict ):
-    """
-    A simple extension of dict which allows cleaner access to items
-    and allows the values to be iterated over directly as if it were a
-    list.  append() is also implemented for simplicity and does not
-    "append".
-    """
-    def __init__( self, dict=None ):
-        odict.__init__( self, dict=None )
-
-    def append( self, item ):
-        self[item.name] = item
-
-    def iter( self ):
-        return self.itervalues()
-
-    def __getattr__( self, name ):
-        return self.get( name )
-
-    def __repr__( self ):
-        # force elements to draw with __str__ for sphinx-apidoc
-        return ', '.join([ item.__str__() for item in self.iter() ])
-
-
-class MetadataParameter( object ):
-    def __init__( self, spec ):
-        self.spec = spec
-
-    def get_html_field( self, value=None, context=None, other_values=None, **kwd ):
-        context = context or {}
-        other_values = other_values or {}
-        return form_builder.TextField( self.spec.name, value=value )
-
-    def get_html( self, value, context=None, other_values=None, **kwd ):
-        """
-        The "context" is simply the metadata collection/bunch holding
-        this piece of metadata. This is passed in to allow for
-        metadata to validate against each other (note: this could turn
-        into a huge, recursive mess if not done with care). For
-        example, a column assignment should validate against the
-        number of columns in the dataset.
-        """
-        context = context or {}
-        other_values = other_values or {}
-
-        if self.spec.get("readonly"):
-            return value
-        if self.spec.get("optional"):
-            checked = False
-            if value:
-                checked = "true"
-            checkbox = form_builder.CheckboxField( "is_" + self.spec.name, checked=checked )
-            return checkbox.get_html() + self.get_html_field( value=value, context=context, other_values=other_values, **kwd ).get_html()
-        else:
-            return self.get_html_field( value=value, context=context, other_values=other_values, **kwd ).get_html()
-
-    def to_string( self, value ):
-        return str( value )
-
-    def to_safe_string( self, value ):
-        return sanitize_lists_to_string( self.to_string( value ) )
-
-    def make_copy( self, value, target_context=None, source_context=None ):
-        return copy.deepcopy( value )
-
-    @classmethod
-    def marshal( cls, value ):
-        """
-        This method should/can be overridden to convert the incoming
-        value to whatever type it is supposed to be.
-        """
-        return value
-
-    def validate( self, value ):
-        """
-        Throw an exception if the value is invalid.
-        """
-        pass
-
-    def unwrap( self, form_value ):
-        """
-        Turns a value into its storable form.
-        """
-        value = self.marshal( form_value )
-        self.validate( value )
-        return value
-
-    def wrap( self, value, session ):
-        """
-        Turns a value into its usable form.
-        """
-        return value
-
-    def from_external_value( self, value, parent ):
-        """
-        Turns a value read from an external dict into its value to be pushed directly into the metadata dict.
-        """
-        return value
-
-    def to_external_value( self, value ):
-        """
-        Turns a value read from a metadata into its value to be pushed directly into the external dict.
-        """
-        return value
-
-
-class MetadataElementSpec( object ):
-    """
-    Defines a metadata element and adds it to the metadata_spec (which
-    is a MetadataSpecCollection) of datatype.
-    """
-    def __init__( self, datatype, name=None, desc=None,
-                  param=MetadataParameter, default=None, no_value=None,
-                  visible=True, set_in_upload=False, **kwargs ):
-        self.name = name
-        self.desc = desc or name
-        self.default = default
-        self.no_value = no_value
-        self.visible = visible
-        self.set_in_upload = set_in_upload
-        # Catch-all, allows for extra attributes to be set
-        self.__dict__.update(kwargs)
-        # set up param last, as it uses values set above
-        self.param = param( self )
-        # add spec element to the spec
-        datatype.metadata_spec.append( self )
-
-    def get( self, name, default=None ):
-        return self.__dict__.get(name, default)
-
-    def wrap( self, value, session ):
-        """
-        Turns a stored value into its usable form.
-        """
-        return self.param.wrap( value, session )
-
-    def unwrap( self, value ):
-        """
-        Turns an incoming value into its storable form.
-        """
-        return self.param.unwrap( value )
-
-    def __str__( self ):
-        # TODO??: assuming param is the class of this MetadataElementSpec - add the plain class name for that
-        spec_dict = dict( param_class=self.param.__class__.__name__ )
-        spec_dict.update( self.__dict__ )
-        return ( "{name} ({param_class}): {desc}, defaults to '{default}'".format( **spec_dict ) )
-
-# create a statement class that, when called,
-#   will add a new MetadataElementSpec to a class's metadata_spec
-MetadataElement = Statement( MetadataElementSpec )
-
-
-"""
-MetadataParameter sub-classes.
-"""
-
-
-class SelectParameter( MetadataParameter ):
-    def __init__( self, spec ):
-        MetadataParameter.__init__( self, spec )
-        self.values = self.spec.get( "values" )
-        self.multiple = string_as_bool( self.spec.get( "multiple" ) )
-
-    def to_string( self, value ):
-        if value in [ None, [] ]:
-            return str( self.spec.no_value )
-        if not isinstance( value, list ):
-            value = [value]
-        return ",".join( map( str, value ) )
-
-    def get_html_field( self, value=None, context=None, other_values=None, values=None, **kwd ):
-        context = context or {}
-        other_values = other_values or {}
-
-        field = form_builder.SelectField( self.spec.name, multiple=self.multiple, display=self.spec.get("display") )
-        if self.values:
-            value_list = self.values
-        elif values:
-            value_list = values
-        elif value:
-            value_list = [ ( v, v ) for v in listify( value )]
-        else:
-            value_list = []
-        for val, label in value_list:
-            try:
-                if ( self.multiple and val in value ) or ( not self.multiple and val == value ):
-                    field.add_option( label, val, selected=True )
-                else:
-                    field.add_option( label, val, selected=False )
-            except TypeError:
-                field.add_option( val, label, selected=False )
-        return field
-
-    def get_html( self, value, context=None, other_values=None, values=None, **kwd ):
-        context = context or {}
-        other_values = other_values or {}
-
-        if self.spec.get("readonly"):
-            if value in [ None, [] ]:
-                return str( self.spec.no_value )
-            return ", ".join( map( str, value ) )
-        return MetadataParameter.get_html( self, value, context=context, other_values=other_values, values=values, **kwd )
-
-    def wrap( self, value, session ):
-        # do we really need this (wasteful)? - yes because we are not sure that
-        # all existing selects have been stored previously as lists. Also this
-        # will handle the case where defaults/no_values are specified and are
-        # single non-list values.
-        value = self.marshal( value )
-        if self.multiple:
-            return value
-        elif value:
-            return value[0]  # single select, only return the first value
-        return None
-
-    @classmethod
-    def marshal( cls, value ):
-        # Store select as list, even if single item
-        if value is None:
-            return []
-        if not isinstance( value, list ):
-            return [value]
-        return value
-
-
-class DBKeyParameter( SelectParameter ):
-
-    def get_html_field( self, value=None, context=None, other_values=None, values=None, **kwd):
-        context = context or {}
-        other_values = other_values or {}
-        try:
-            values = kwd['trans'].app.genome_builds.get_genome_build_names( kwd['trans'] )
-        except KeyError:
-            pass
-        return super(DBKeyParameter, self).get_html_field( value, context, other_values, values, **kwd)
-
-    def get_html( self, value=None, context=None, other_values=None, values=None, **kwd):
-        context = context or {}
-        other_values = other_values or {}
-        try:
-            values = kwd['trans'].app.genome_builds.get_genome_build_names( kwd['trans'] )
-        except KeyError:
-            pass
-        return super(DBKeyParameter, self).get_html( value, context, other_values, values, **kwd)
-
-
-class RangeParameter( SelectParameter ):
-
-    def __init__( self, spec ):
-        SelectParameter.__init__( self, spec )
-        # The spec must be set with min and max values
-        self.min = spec.get( "min" ) or 1
-        self.max = spec.get( "max" ) or 1
-        self.step = self.spec.get( "step" ) or 1
-
-    def get_html_field( self, value=None, context=None, other_values=None, values=None, **kwd ):
-        context = context or {}
-        other_values = other_values or {}
-
-        if values is None:
-            values = zip( range( self.min, self.max, self.step ), range( self.min, self.max, self.step ))
-        return SelectParameter.get_html_field( self, value=value, context=context, other_values=other_values, values=values, **kwd )
-
-    def get_html( self, value, context=None, other_values=None, values=None, **kwd ):
-        context = context or {}
-        other_values = other_values or {}
-
-        if values is None:
-            values = zip( range( self.min, self.max, self.step ), range( self.min, self.max, self.step ))
-        return SelectParameter.get_html( self, value, context=context, other_values=other_values, values=values, **kwd )
-
-    @classmethod
-    def marshal( cls, value ):
-        value = SelectParameter.marshal( value )
-        values = [ int(x) for x in value ]
-        return values
-
-
-class ColumnParameter( RangeParameter ):
-
-    def get_html_field( self, value=None, context=None, other_values=None, values=None, **kwd ):
-        context = context or {}
-        other_values = other_values or {}
-
-        if values is None and context:
-            column_range = range( 1, ( context.columns or 0 ) + 1, 1 )
-            values = zip( column_range, column_range )
-        return RangeParameter.get_html_field( self, value=value, context=context, other_values=other_values, values=values, **kwd )
-
-    def get_html( self, value, context=None, other_values=None, values=None, **kwd ):
-        context = context or {}
-        other_values = other_values or {}
-
-        if values is None and context:
-            column_range = range( 1, ( context.columns or 0 ) + 1, 1 )
-            values = zip( column_range, column_range )
-        return RangeParameter.get_html( self, value, context=context, other_values=other_values, values=values, **kwd )
-
-
-class ColumnTypesParameter( MetadataParameter ):
-
-    def to_string( self, value ):
-        return ",".join( map( str, value ) )
-
-
-class ListParameter( MetadataParameter ):
-
-    def to_string( self, value ):
-        return ",".join( [str(x) for x in value] )
-
-
-class DictParameter( MetadataParameter ):
-
-    def to_string( self, value ):
-        return json.dumps( value )
-
-    def to_safe_string( self, value ):
-        # We do not sanitize json dicts
-        return safe_dumps( value )
-
-
-class PythonObjectParameter( MetadataParameter ):
-
-    def to_string( self, value ):
-        if not value:
-            return self.spec._to_string( self.spec.no_value )
-        return self.spec._to_string( value )
-
-    def get_html_field( self, value=None, context=None, other_values=None, **kwd ):
-        context = context or {}
-        other_values = other_values or {}
-        return form_builder.TextField( self.spec.name, value=self._to_string( value ) )
-
-    def get_html( self, value=None, context=None, other_values=None, **kwd ):
-        context = context or {}
-        other_values = other_values or {}
-        return str( self )
-
-    @classmethod
-    def marshal( cls, value ):
-        return value
-
-
-class FileParameter( MetadataParameter ):
-
-    def to_string( self, value ):
-        if not value:
-            return str( self.spec.no_value )
-        return value.file_name
-
-    def to_safe_string( self, value ):
-        # We do not sanitize file names
-        return self.to_string( value )
-
-    def get_html_field( self, value=None, context=None, other_values=None, **kwd ):
-        context = context or {}
-        other_values = other_values or {}
-        return form_builder.TextField( self.spec.name, value=str( value.id ) )
-
-    def get_html( self, value=None, context=None, other_values=None, **kwd ):
-        context = context or {}
-        other_values = other_values or {}
-        return "<div>No display available for Metadata Files</div>"
-
-    def wrap( self, value, session ):
-        if value is None:
-            return None
-        if isinstance( value, galaxy.model.MetadataFile ) or isinstance( value, MetadataTempFile ):
-            return value
-        mf = session.query( galaxy.model.MetadataFile ).get( value )
-        return mf
-
-    def make_copy( self, value, target_context, source_context ):
-        value = self.wrap( value, object_session( target_context.parent ) )
-        if value:
-            new_value = galaxy.model.MetadataFile( dataset=target_context.parent, name=self.spec.name )
-            object_session( target_context.parent ).add( new_value )
-            object_session( target_context.parent ).flush()
-            shutil.copy( value.file_name, new_value.file_name )
-            return self.unwrap( new_value )
-        return None
-
-    @classmethod
-    def marshal( cls, value ):
-        if isinstance( value, galaxy.model.MetadataFile ):
-            value = value.id
-        return value
-
-    def from_external_value( self, value, parent, path_rewriter=None ):
-        """
-        Turns a value read from a external dict into its value to be pushed directly into the metadata dict.
-        """
-        if MetadataTempFile.is_JSONified_value( value ):
-            value = MetadataTempFile.from_JSON( value )
-        if isinstance( value, MetadataTempFile ):
-            mf = parent.metadata.get( self.spec.name, None)
-            if mf is None:
-                mf = self.new_file( dataset=parent, **value.kwds )
-            # Ensure the metadata file gets updated with content
-            file_name = value.file_name
-            if path_rewriter:
-                # Job may have run with a different (non-local) tmp/working
-                # directory. Correct.
-                file_name = path_rewriter( file_name )
-            parent.dataset.object_store.update_from_file( mf,
-                                                          file_name=file_name,
-                                                          extra_dir='_metadata_files',
-                                                          extra_dir_at_root=True,
-                                                          alt_name=os.path.basename(mf.file_name) )
-            os.unlink( file_name )
-            value = mf.id
-        return value
-
-    def to_external_value( self, value ):
-        """
-        Turns a value read from a metadata into its value to be pushed directly into the external dict.
-        """
-        if isinstance( value, galaxy.model.MetadataFile ):
-            value = value.id
-        elif isinstance( value, MetadataTempFile ):
-            value = MetadataTempFile.to_JSON( value )
-        return value
-
-    def new_file( self, dataset=None, **kwds ):
-        if object_session( dataset ):
-            mf = galaxy.model.MetadataFile( name=self.spec.name, dataset=dataset, **kwds )
-            object_session( dataset ).add( mf )
-            object_session( dataset ).flush()  # flush to assign id
-            return mf
-        else:
-            # we need to make a tmp file that is accessable to the head node,
-            # we will be copying its contents into the MetadataFile objects filename after restoring from JSON
-            # we do not include 'dataset' in the kwds passed, as from_JSON_value() will handle this for us
-            return MetadataTempFile( **kwds )
-
-
-# This class is used when a database file connection is not available
-class MetadataTempFile( object ):
-    tmp_dir = 'database/tmp'  # this should be overwritten as necessary in calling scripts
-
-    def __init__( self, **kwds ):
-        self.kwds = kwds
-        self._filename = None
-
-    @property
-    def file_name( self ):
-        if self._filename is None:
-            # we need to create a tmp file, accessable across all nodes/heads, save the name, and return it
-            self._filename = abspath( tempfile.NamedTemporaryFile( dir=self.tmp_dir, prefix="metadata_temp_file_" ).name )
-            open( self._filename, 'wb+' )  # create an empty file, so it can't be reused using tempfile
-        return self._filename
-
-    def to_JSON( self ):
-        return { '__class__': self.__class__.__name__,
-                 'filename': self.file_name,
-                 'kwds': self.kwds }
-
-    @classmethod
-    def from_JSON( cls, json_dict ):
-        # need to ensure our keywords are not unicode
-        rval = cls( **stringify_dictionary_keys( json_dict['kwds'] ) )
-        rval._filename = json_dict['filename']
-        return rval
-
-    @classmethod
-    def is_JSONified_value( cls, value ):
-        return ( isinstance( value, dict ) and value.get( '__class__', None ) == cls.__name__ )
-
-    @classmethod
-    def cleanup_from_JSON_dict_filename( cls, filename ):
-        try:
-            for key, value in json.load( open( filename ) ).items():
-                if cls.is_JSONified_value( value ):
-                    value = cls.from_JSON( value )
-                if isinstance( value, cls ) and os.path.exists( value.file_name ):
-                    log.debug( 'Cleaning up abandoned MetadataTempFile file: %s' % value.file_name )
-                    os.unlink( value.file_name )
-        except Exception as e:
-            log.debug( 'Failed to cleanup MetadataTempFile temp files from %s: %s' % ( filename, e ) )
-
-
-class JobExternalOutputMetadataWrapper( object ):
-    """
-    Class with methods allowing set_meta() to be called externally to the
-    Galaxy head.
-    This class allows access to external metadata filenames for all outputs
-    associated with a job.
-    We will use JSON as the medium of exchange of information, except for the
-    DatasetInstance object which will use pickle (in the future this could be
-    JSONified as well)
-    """
-
-    def __init__( self, job ):
-        self.job_id = job.id
-
-    def get_output_filenames_by_dataset( self, dataset, sa_session ):
-        if isinstance( dataset, galaxy.model.HistoryDatasetAssociation ):
-            return sa_session.query( galaxy.model.JobExternalOutputMetadata ) \
-                             .filter_by( job_id=self.job_id,
-                                         history_dataset_association_id=dataset.id,
-                                         is_valid=True ) \
-                             .first()  # there should only be one or None
-        elif isinstance( dataset, galaxy.model.LibraryDatasetDatasetAssociation ):
-            return sa_session.query( galaxy.model.JobExternalOutputMetadata ) \
-                             .filter_by( job_id=self.job_id,
-                                         library_dataset_dataset_association_id=dataset.id,
-                                         is_valid=True ) \
-                             .first()  # there should only be one or None
-        return None
-
-    def get_dataset_metadata_key( self, dataset ):
-        # Set meta can be called on library items and history items,
-        # need to make different keys for them, since ids can overlap
-        return "%s_%d" % ( dataset.__class__.__name__, dataset.id )
-
-    def invalidate_external_metadata( self, datasets, sa_session ):
-        for dataset in datasets:
-            jeom = self.get_output_filenames_by_dataset( dataset, sa_session )
-            # shouldn't be more than one valid, but you never know
-            while jeom:
-                jeom.is_valid = False
-                sa_session.add( jeom )
-                sa_session.flush()
-                jeom = self.get_output_filenames_by_dataset( dataset, sa_session )
-
-    def setup_external_metadata( self, datasets, sa_session, exec_dir=None,
-                                 tmp_dir=None, dataset_files_path=None,
-                                 output_fnames=None, config_root=None,
-                                 config_file=None, datatypes_config=None,
-                                 job_metadata=None, compute_tmp_dir=None,
-                                 include_command=True, max_metadata_value_size=0,
-                                 kwds=None):
-        kwds = kwds or {}
-        if tmp_dir is None:
-            tmp_dir = MetadataTempFile.tmp_dir
-        else:
-            MetadataTempFile.tmp_dir = tmp_dir
-
-        if not os.path.exists(tmp_dir):
-            os.makedirs(tmp_dir)
-
-        # path is calculated for Galaxy, may be different on compute - rewrite
-        # for the compute server.
-        def metadata_path_on_compute(path):
-            compute_path = path
-            if compute_tmp_dir and tmp_dir and in_directory(path, tmp_dir):
-                path_relative = os.path.relpath(path, tmp_dir)
-                compute_path = os.path.join(compute_tmp_dir, path_relative)
-            return compute_path
-
-        # fill in metadata_files_dict and return the command with args required to set metadata
-        def __metadata_files_list_to_cmd_line( metadata_files ):
-            def __get_filename_override():
-                if output_fnames:
-                    for dataset_path in output_fnames:
-                        if dataset_path.real_path == metadata_files.dataset.file_name:
-                            return dataset_path.false_path or dataset_path.real_path
-                return ""
-            line = '"%s,%s,%s,%s,%s,%s"' % (
-                metadata_path_on_compute(metadata_files.filename_in),
-                metadata_path_on_compute(metadata_files.filename_kwds),
-                metadata_path_on_compute(metadata_files.filename_out),
-                metadata_path_on_compute(metadata_files.filename_results_code),
-                __get_filename_override(),
-                metadata_path_on_compute(metadata_files.filename_override_metadata),
-            )
-            return line
-        if not isinstance( datasets, list ):
-            datasets = [ datasets ]
-        if exec_dir is None:
-            exec_dir = os.path.abspath( os.getcwd() )
-        if dataset_files_path is None:
-            dataset_files_path = galaxy.model.Dataset.file_path
-        if config_root is None:
-            config_root = os.path.abspath( os.getcwd() )
-        if datatypes_config is None:
-            raise Exception( 'In setup_external_metadata, the received datatypes_config is None.' )
-            datatypes_config = 'datatypes_conf.xml'
-        metadata_files_list = []
-        for dataset in datasets:
-            key = self.get_dataset_metadata_key( dataset )
-            # future note:
-            # wonkiness in job execution causes build command line to be called more than once
-            # when setting metadata externally, via 'auto-detect' button in edit attributes, etc.,
-            # we don't want to overwrite (losing the ability to cleanup) our existing dataset keys and files,
-            # so we will only populate the dictionary once
-            metadata_files = self.get_output_filenames_by_dataset( dataset, sa_session )
-            if not metadata_files:
-                job = sa_session.query( galaxy.model.Job ).get( self.job_id )
-                metadata_files = galaxy.model.JobExternalOutputMetadata( job=job, dataset=dataset )
-                # we are using tempfile to create unique filenames, tempfile always returns an absolute path
-                # we will use pathnames relative to the galaxy root, to accommodate instances where the galaxy root
-                # is located differently, i.e. on a cluster node with a different filesystem structure
-
-                # file to store existing dataset
-                metadata_files.filename_in = abspath( tempfile.NamedTemporaryFile( dir=tmp_dir, prefix="metadata_in_%s_" % key ).name )
-
-                # FIXME: HACK
-                # sqlalchemy introduced 'expire_on_commit' flag for sessionmaker at version 0.5x
-                # This may be causing the dataset attribute of the dataset_association object to no-longer be loaded into memory when needed for pickling.
-                # For now, we'll simply 'touch' dataset_association.dataset to force it back into memory.
-                dataset.dataset  # force dataset_association.dataset to be loaded before pickling
-                # A better fix could be setting 'expire_on_commit=False' on the session, or modifying where commits occur, or ?
-
-                # Touch also deferred column
-                dataset._metadata
-
-                cPickle.dump( dataset, open( metadata_files.filename_in, 'wb+' ) )
-                # file to store metadata results of set_meta()
-                metadata_files.filename_out = abspath( tempfile.NamedTemporaryFile( dir=tmp_dir, prefix="metadata_out_%s_" % key ).name )
-                open( metadata_files.filename_out, 'wb+' )  # create the file on disk, so it cannot be reused by tempfile (unlikely, but possible)
-                # file to store a 'return code' indicating the results of the set_meta() call
-                # results code is like (True/False - if setting metadata was successful/failed , exception or string of reason of success/failure )
-                metadata_files.filename_results_code = abspath( tempfile.NamedTemporaryFile( dir=tmp_dir, prefix="metadata_results_%s_" % key ).name )
-                # create the file on disk, so it cannot be reused by tempfile (unlikely, but possible)
-                json.dump( ( False, 'External set_meta() not called' ), open( metadata_files.filename_results_code, 'wb+' ) )
-                # file to store kwds passed to set_meta()
-                metadata_files.filename_kwds = abspath( tempfile.NamedTemporaryFile( dir=tmp_dir, prefix="metadata_kwds_%s_" % key ).name )
-                json.dump( kwds, open( metadata_files.filename_kwds, 'wb+' ), ensure_ascii=True )
-                # existing metadata file parameters need to be overridden with cluster-writable file locations
-                metadata_files.filename_override_metadata = abspath( tempfile.NamedTemporaryFile( dir=tmp_dir, prefix="metadata_override_%s_" % key ).name )
-                open( metadata_files.filename_override_metadata, 'wb+' )  # create the file on disk, so it cannot be reused by tempfile (unlikely, but possible)
-                override_metadata = []
-                for meta_key, spec_value in dataset.metadata.spec.iteritems():
-                    if isinstance( spec_value.param, FileParameter ) and dataset.metadata.get( meta_key, None ) is not None:
-                        metadata_temp = MetadataTempFile()
-                        shutil.copy( dataset.metadata.get( meta_key, None ).file_name, metadata_temp.file_name )
-                        override_metadata.append( ( meta_key, metadata_temp.to_JSON() ) )
-                json.dump( override_metadata, open( metadata_files.filename_override_metadata, 'wb+' ) )
-                # add to session and flush
-                sa_session.add( metadata_files )
-                sa_session.flush()
-            metadata_files_list.append( metadata_files )
-        args = '"%s" "%s" %s %s' % ( datatypes_config,
-                                     job_metadata,
-                                     " ".join( map( __metadata_files_list_to_cmd_line, metadata_files_list ) ),
-                                     max_metadata_value_size)
-        if include_command:
-            # return command required to build
-            fd, fp = tempfile.mkstemp( suffix='.py', dir=tmp_dir, prefix="set_metadata_" )
-            metadata_script_file = abspath( fp )
-            os.fdopen( fd, 'w' ).write( 'from galaxy_ext.metadata.set_metadata import set_metadata; set_metadata()' )
-            return 'python "%s" %s' % ( metadata_path_on_compute(metadata_script_file), args )
-        else:
-            # return args to galaxy_ext.metadata.set_metadata required to build
-            return args
-
-    def external_metadata_set_successfully( self, dataset, sa_session ):
-        metadata_files = self.get_output_filenames_by_dataset( dataset, sa_session )
-        if not metadata_files:
-            return False  # this file doesn't exist
-        rval, rstring = json.load( open( metadata_files.filename_results_code ) )
-        if not rval:
-            log.debug( 'setting metadata externally failed for %s %s: %s' % ( dataset.__class__.__name__, dataset.id, rstring ) )
-        return rval
-
-    def cleanup_external_metadata( self, sa_session ):
-        log.debug( 'Cleaning up external metadata files' )
-        for metadata_files in sa_session.query( galaxy.model.Job ).get( self.job_id ).external_output_metadata:
-            # we need to confirm that any MetadataTempFile files were removed, if not we need to remove them
-            # can occur if the job was stopped before completion, but a MetadataTempFile is used in the set_meta
-            MetadataTempFile.cleanup_from_JSON_dict_filename( metadata_files.filename_out )
-            dataset_key = self.get_dataset_metadata_key( metadata_files.dataset )
-            for key, fname in [ ( 'filename_in', metadata_files.filename_in ),
-                                ( 'filename_out', metadata_files.filename_out ),
-                                ( 'filename_results_code', metadata_files.filename_results_code ),
-                                ( 'filename_kwds', metadata_files.filename_kwds ),
-                                ( 'filename_override_metadata', metadata_files.filename_override_metadata ) ]:
-                try:
-                    os.remove( fname )
-                except Exception as e:
-                    log.debug( 'Failed to cleanup external metadata file (%s) for %s: %s' % ( key, dataset_key, e ) )
-
-    def set_job_runner_external_pid( self, pid, sa_session ):
-        for metadata_files in sa_session.query( galaxy.model.Job ).get( self.job_id ).external_output_metadata:
-            metadata_files.job_runner_external_pid = pid
-            sa_session.add( metadata_files )
-            sa_session.flush()
+from galaxy.model.metadata import (
+    Statement,
+    MetadataElement,
+    MetadataCollection,
+    MetadataSpecCollection,
+    MetadataParameter,
+    MetadataElementSpec,
+    SelectParameter,
+    DBKeyParameter,
+    RangeParameter,
+    ColumnParameter,
+    ColumnTypesParameter,
+    ListParameter,
+    DictParameter,
+    PythonObjectParameter,
+    FileParameter,
+    MetadataTempFile,
+    JobExternalOutputMetadataWrapper,
+)
+
+__all__ = [
+    "Statement",
+    "MetadataElement",
+    "MetadataCollection",
+    "MetadataSpecCollection",
+    "MetadataParameter",
+    "MetadataElementSpec",
+    "SelectParameter",
+    "DBKeyParameter",
+    "RangeParameter",
+    "ColumnParameter",
+    "ColumnTypesParameter",
+    "ListParameter",
+    "DictParameter",
+    "PythonObjectParameter",
+    "FileParameter",
+    "MetadataTempFile",
+    "JobExternalOutputMetadataWrapper",
+]
diff --git a/lib/galaxy/datatypes/registry.py b/lib/galaxy/datatypes/registry.py
index 17c4b70c122..e16a02637bc 100644
--- a/lib/galaxy/datatypes/registry.py
+++ b/lib/galaxy/datatypes/registry.py
@@ -1,23 +1,25 @@
 """
 Provides mapping between extensions and datatypes, mime-types, etc.
 """
+from __future__ import absolute_import
+
 import os
 import tempfile
 import logging
 import imp
-import data
-import tabular
-import interval
-import images
-import sequence
-import qualityscore
-import xml
-import coverage
-import tracks
-import binary
+from . import data
+from . import tabular
+from . import interval
+from . import images
+from . import sequence
+from . import qualityscore
+from . import xml
+from . import coverage
+from . import tracks
+from . import binary
 import galaxy.util
 from galaxy.util.odict import odict
-from display_applications.application import DisplayApplication
+from .display_applications.application import DisplayApplication
 
 
 class ConfigurationError( Exception ):
diff --git a/lib/galaxy/datatypes/sniff.py b/lib/galaxy/datatypes/sniff.py
index 5683c804e37..2dbebd9ff18 100644
--- a/lib/galaxy/datatypes/sniff.py
+++ b/lib/galaxy/datatypes/sniff.py
@@ -1,11 +1,12 @@
 """
 File format detector
 """
+from __future__ import absolute_import
+
 import gzip
 import logging
 import os
 import re
-import registry
 import shutil
 import sys
 import tempfile
@@ -253,75 +254,75 @@ def is_column_based( fname, sep='\t', skip=0, is_multi_byte=False ):
     return True
 
 
-def guess_ext( fname, sniff_order=None, is_multi_byte=False ):
+def guess_ext( fname, sniff_order, is_multi_byte=False ):
     """
     Returns an extension that can be used in the datatype factory to
     generate a data for the 'fname' file
 
     >>> fname = get_test_fname('megablast_xml_parser_test1.blastxml')
-    >>> guess_ext(fname)
+    >>> from galaxy.datatypes import registry
+    >>> datatypes_registry = registry.Registry()
+    >>> datatypes_registry.load_datatypes()
+    >>> sniff_order = datatypes_registry.sniff_order
+    >>> guess_ext(fname, sniff_order)
     'xml'
     >>> fname = get_test_fname('interval.interval')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'interval'
     >>> fname = get_test_fname('interval1.bed')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'bed'
     >>> fname = get_test_fname('test_tab.bed')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'bed'
     >>> fname = get_test_fname('sequence.maf')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'maf'
     >>> fname = get_test_fname('sequence.fasta')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'fasta'
     >>> fname = get_test_fname('file.html')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'html'
     >>> fname = get_test_fname('test.gtf')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'gtf'
     >>> fname = get_test_fname('test.gff')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'gff'
     >>> fname = get_test_fname('gff_version_3.gff')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'gff3'
     >>> fname = get_test_fname('temp.txt')
     >>> file(fname, 'wt').write("a\\t2\\nc\\t1\\nd\\t0")
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'tabular'
     >>> fname = get_test_fname('temp.txt')
     >>> file(fname, 'wt').write("a 1 2 x\\nb 3 4 y\\nc 5 6 z")
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'txt'
     >>> fname = get_test_fname('test_tab1.tabular')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'tabular'
     >>> fname = get_test_fname('alignment.lav')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'lav'
     >>> fname = get_test_fname('1.sff')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'sff'
     >>> fname = get_test_fname('1.bam')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'bam'
     >>> fname = get_test_fname('3unsorted.bam')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'bam'
     >>> fname = get_test_fname('test.idpDB')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'idpdb'
     >>> fname = get_test_fname('test.mz5')
-    >>> guess_ext(fname)
+    >>> guess_ext(fname, sniff_order)
     'h5'
     """
-    if sniff_order is None:
-        datatypes_registry = registry.Registry()
-        datatypes_registry.load_datatypes()
-        sniff_order = datatypes_registry.sniff_order
     for datatype in sniff_order:
         """
         Some classes may not have a sniff function, which is ok.  In fact, the
diff --git a/lib/galaxy/datatypes/tabular.py b/lib/galaxy/datatypes/tabular.py
index ebc10108878..07495e49c23 100644
--- a/lib/galaxy/datatypes/tabular.py
+++ b/lib/galaxy/datatypes/tabular.py
@@ -1,6 +1,8 @@
 """
 Tabular datatype
 """
+from __future__ import absolute_import
+
 import csv
 import gzip
 import logging
@@ -18,7 +20,7 @@ from galaxy.datatypes.metadata import MetadataElement
 from galaxy.datatypes.sniff import get_headers
 from galaxy.util.json import dumps
 
-import dataproviders
+from . import dataproviders
 
 log = logging.getLogger(__name__)
 
@@ -461,15 +463,15 @@ class Sam( Tabular ):
                     break  # EOF
                 if line:
                     if line[0] != '@':
-                        linePieces = line.split('\t')
-                        if len(linePieces) < 11:
+                        line_pieces = line.split('\t')
+                        if len(line_pieces) < 11:
                             return False
                         try:
-                            int(linePieces[1])
-                            int(linePieces[3])
-                            int(linePieces[4])
-                            int(linePieces[7])
-                            int(linePieces[8])
+                            int(line_pieces[1])
+                            int(line_pieces[3])
+                            int(line_pieces[4])
+                            int(line_pieces[7])
+                            int(line_pieces[8])
                         except ValueError:
                             return False
                         count += 1
@@ -792,18 +794,18 @@ class Eland( Tabular ):
                 if not line:
                     break  # EOF
                 if line:
-                    linePieces = line.split('\t')
-                    if len(linePieces) != 22:
+                    line_pieces = line.split('\t')
+                    if len(line_pieces) != 22:
                         return False
                     try:
-                        if long(linePieces[1]) < 0:
+                        if long(line_pieces[1]) < 0:
                             raise Exception('Out of range')
-                        if long(linePieces[2]) < 0:
+                        if long(line_pieces[2]) < 0:
                             raise Exception('Out of range')
-                        if long(linePieces[3]) < 0:
+                        if long(line_pieces[3]) < 0:
                             raise Exception('Out of range')
-                        int(linePieces[4])
-                        int(linePieces[5])
+                        int(line_pieces[4])
+                        int(line_pieces[5])
                         # can get a lot more specific
                     except ValueError:
                         fh.close()
@@ -838,13 +840,13 @@ class Eland( Tabular ):
             # Otherwise, read the whole thing and set num data lines.
             for i, line in enumerate(dataset_fh):
                 if line:
-                    linePieces = line.split('\t')
-                    if len(linePieces) != 22:
+                    line_pieces = line.split('\t')
+                    if len(line_pieces) != 22:
                         raise Exception('%s:%d:Corrupt line!' % (dataset.file_name, i))
-                    lanes[linePieces[2]] = 1
-                    tiles[linePieces[3]] = 1
-                    barcodes[linePieces[6]] = 1
-                    reads[linePieces[7]] = 1
+                    lanes[line_pieces[2]] = 1
+                    tiles[line_pieces[3]] = 1
+                    barcodes[line_pieces[6]] = 1
+                    reads[line_pieces[7]] = 1
                 pass
             dataset.metadata.data_lines = i + 1
             dataset_fh.close()
diff --git a/lib/galaxy/metadata b/lib/galaxy/metadata
deleted file mode 120000
index 5eecd174bbb..00000000000
--- a/lib/galaxy/metadata
+++ /dev/null
@@ -1 +0,0 @@
-../galaxy_ext/metadata
\ No newline at end of file
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index e1e82c18370..f2b291419b7 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -27,12 +27,10 @@ try:
 except ImportError:
     pexpect = None
 
-import galaxy.datatypes
-import galaxy.datatypes.registry
 import galaxy.model.orm.now
+import galaxy.model.metadata
 import galaxy.security.passwords
 import galaxy.util
-from galaxy.datatypes.metadata import MetadataCollection
 from galaxy.model.item_attrs import UsesAnnotations
 from galaxy.util.dictifiable import Dictifiable
 from galaxy.security import get_permitted_actions
@@ -50,9 +48,7 @@ from galaxy.web.form_builder import (AddressField, CheckboxField, HistoryField,
 
 log = logging.getLogger( __name__ )
 
-datatypes_registry = galaxy.datatypes.registry.Registry()
-# Default Value Required for unit tests
-datatypes_registry.load_datatypes()
+_datatypes_registry = None
 
 # When constructing filters with in for a fixed set of ids, maximum
 # number of items to place in the IN statement. Different databases
@@ -80,12 +76,18 @@ class ConverterDependencyException(Exception):
         return repr(self.value)
 
 
+def _get_datatypes_registry():
+    if _datatypes_registry is None:
+        raise Exception("galaxy.model.set_datatypes_registry must be called before performing certain DatasetInstance operations.")
+    return _datatypes_registry
+
+
 def set_datatypes_registry( d_registry ):
     """
     Set up datatypes_registry
     """
-    global datatypes_registry
-    datatypes_registry = d_registry
+    global _datatypes_registry
+    _datatypes_registry = d_registry
 
 
 class HasName:
@@ -1858,13 +1860,13 @@ class DatasetInstance( object ):
 
     @property
     def datatype( self ):
-        return datatypes_registry.get_datatype_by_extension( self.extension )
+        return _get_datatypes_registry().get_datatype_by_extension( self.extension )
 
     def get_metadata( self ):
         # using weakref to store parent (to prevent circ ref),
         #   does a Session.clear() cause parent to be invalidated, while still copying over this non-database attribute?
         if not hasattr( self, '_metadata_collection' ) or self._metadata_collection.parent != self:
-            self._metadata_collection = MetadataCollection( self )
+            self._metadata_collection = galaxy.model.metadata.MetadataCollection( self )
         return self._metadata_collection
 
     def set_metadata( self, bunch ):
@@ -1892,7 +1894,7 @@ class DatasetInstance( object ):
 
     def change_datatype( self, new_ext ):
         self.clear_associated_files()
-        datatypes_registry.change_datatype( self, new_ext )
+        _get_datatypes_registry().change_datatype( self, new_ext )
 
     def get_size( self, nice_size=False ):
         """Returns the size of the data on disk"""
@@ -1929,7 +1931,7 @@ class DatasetInstance( object ):
     def get_mime( self ):
         """Returns the mime type of the data"""
         try:
-            return datatypes_registry.get_mimetype_by_extension( self.extension.lower() )
+            return _get_datatypes_registry().get_mimetype_by_extension( self.extension.lower() )
         except AttributeError:
             # extension is None
             return 'data'
@@ -2054,14 +2056,14 @@ class DatasetInstance( object ):
         return None
 
     def get_converter_types(self):
-        return self.datatype.get_converter_types( self, datatypes_registry )
+        return self.datatype.get_converter_types( self, _get_datatypes_registry() )
 
     def can_convert_to(self, format):
         return format in self.get_converter_types()
 
     def find_conversion_destination( self, accepted_formats, **kwd ):
         """Returns ( target_ext, existing converted dataset )"""
-        return self.datatype.find_conversion_destination( self, accepted_formats, datatypes_registry, **kwd )
+        return self.datatype.find_conversion_destination( self, accepted_formats, _get_datatypes_registry(), **kwd )
 
     def add_validation_error( self, validation_error ):
         self.validation_errors.append( validation_error )
diff --git a/lib/galaxy/model/metadata.py b/lib/galaxy/model/metadata.py
new file mode 100644
index 00000000000..49f8f79b457
--- /dev/null
+++ b/lib/galaxy/model/metadata.py
@@ -0,0 +1,894 @@
+"""
+Galaxy Metadata
+
+"""
+
+import copy
+import cPickle
+import json
+import os
+import shutil
+import sys
+import tempfile
+import weakref
+
+from os.path import abspath
+
+from sqlalchemy.orm import object_session
+
+import galaxy.model
+from galaxy.util import listify
+from galaxy.util.object_wrapper import sanitize_lists_to_string
+from galaxy.util import stringify_dictionary_keys
+from galaxy.util import string_as_bool
+from galaxy.util import in_directory
+from galaxy.util.json import safe_dumps
+from galaxy.util.odict import odict
+from galaxy.web import form_builder
+
+import logging
+log = logging.getLogger(__name__)
+
+STATEMENTS = "__galaxy_statements__"  # this is the name of the property in a Datatype class where new metadata spec element Statements are stored
+
+
+class Statement( object ):
+    """
+    This class inserts its target into a list in the surrounding
+    class.  the data.Data class has a metaclass which executes these
+    statements.  This is how we shove the metadata element spec into
+    the class.
+    """
+    def __init__( self, target ):
+        self.target = target
+
+    def __call__( self, *args, **kwargs ):
+        # get the locals dictionary of the frame object one down in the call stack (i.e. the Datatype class calling MetadataElement)
+        class_locals = sys._getframe( 1 ).f_locals
+        # get and set '__galaxy_statments__' to an empty list if not in locals dict
+        statements = class_locals.setdefault( STATEMENTS, [] )
+        # add Statement containing info to populate a MetadataElementSpec
+        statements.append( ( self, args, kwargs ) )
+
+    @classmethod
+    def process( cls, element ):
+        for statement, args, kwargs in getattr( element, STATEMENTS, [] ):
+            statement.target( element, *args, **kwargs )  # statement.target is MetadataElementSpec, element is a Datatype class
+
+
+class MetadataCollection( object ):
+    """
+    MetadataCollection is not a collection at all, but rather a proxy
+    to the real metadata which is stored as a Dictionary. This class
+    handles processing the metadata elements when they are set and
+    retrieved, returning default values in cases when metadata is not set.
+    """
+    def __init__(self, parent ):
+        self.parent = parent
+        # initialize dict if needed
+        if self.parent._metadata is None:
+            self.parent._metadata = {}
+
+    def get_parent( self ):
+        if "_parent" in self.__dict__:
+            return self.__dict__["_parent"]()
+        return None
+
+    def set_parent( self, parent ):
+        # use weakref to prevent a circular reference interfering with garbage
+        # collection: hda/lda (parent) <--> MetadataCollection (self) ; needs to be
+        # hashable, so cannot use proxy.
+        self.__dict__["_parent"] = weakref.ref( parent )
+    parent = property( get_parent, set_parent )
+
+    @property
+    def spec( self ):
+        return self.parent.datatype.metadata_spec
+
+    def __iter__( self ):
+        return self.parent._metadata.__iter__()
+
+    def get( self, key, default=None ):
+        try:
+            return self.__getattr__( key ) or default
+        except:
+            return default
+
+    def items(self):
+        return iter( [ ( k, self.get( k ) ) for k in self.spec.iterkeys() ] )
+
+    def __str__(self):
+        return dict( self.items() ).__str__()
+
+    def __nonzero__( self ):
+        return bool( self.parent._metadata )
+
+    def __getattr__( self, name ):
+        if name in self.spec:
+            if name in self.parent._metadata:
+                return self.spec[name].wrap( self.parent._metadata[name], object_session( self.parent ) )
+            return self.spec[name].wrap( self.spec[name].default, object_session( self.parent ) )
+        if name in self.parent._metadata:
+            return self.parent._metadata[name]
+
+    def __setattr__( self, name, value ):
+        if name == "parent":
+            return self.set_parent( value )
+        else:
+            if name in self.spec:
+                self.parent._metadata[name] = self.spec[name].unwrap( value )
+            else:
+                self.parent._metadata[name] = value
+
+    def remove_key( self, name ):
+        if name in self.parent._metadata:
+            del self.parent._metadata[name]
+        else:
+            log.info( "Attempted to delete invalid key '%s' from MetadataCollection" % name )
+
+    def element_is_set( self, name ):
+        return bool( self.parent._metadata.get( name, False ) )
+
+    def get_html_by_name( self, name, **kwd ):
+        if name in self.spec:
+            rval = self.spec[name].param.get_html( value=getattr( self, name ), context=self, **kwd )
+            if rval is None:
+                return self.spec[name].no_value
+            return rval
+
+    def make_dict_copy( self, to_copy ):
+        """Makes a deep copy of input iterable to_copy according to self.spec"""
+        rval = {}
+        for key, value in to_copy.items():
+            if key in self.spec:
+                rval[key] = self.spec[key].param.make_copy( value, target_context=self, source_context=to_copy )
+        return rval
+
+    def from_JSON_dict( self, filename=None, path_rewriter=None, json_dict=None ):
+        dataset = self.parent
+        if filename is not None:
+            log.debug( 'loading metadata from file for: %s %s' % ( dataset.__class__.__name__, dataset.id ) )
+            JSONified_dict = json.load( open( filename ) )
+        elif json_dict is not None:
+            log.debug( 'loading metadata from dict for: %s %s' % ( dataset.__class__.__name__, dataset.id ) )
+            if isinstance( json_dict, basestring ):
+                JSONified_dict = json.loads( json_dict )
+            elif isinstance( json_dict, dict ):
+                JSONified_dict = json_dict
+            else:
+                raise ValueError( "json_dict must be either a dictionary or a string, got %s." % ( type( json_dict ) ) )
+        else:
+            raise ValueError( "You must provide either a filename or a json_dict" )
+        for name, spec in self.spec.items():
+            if name in JSONified_dict:
+                from_ext_kwds = {}
+                external_value = JSONified_dict[ name ]
+                param = spec.param
+                if isinstance( param, FileParameter ):
+                    from_ext_kwds[ 'path_rewriter' ] = path_rewriter
+                dataset._metadata[ name ] = param.from_external_value( external_value, dataset, **from_ext_kwds )
+            elif name in dataset._metadata:
+                # if the metadata value is not found in our externally set metadata but it has a value in the 'old'
+                # metadata associated with our dataset, we'll delete it from our dataset's metadata dict
+                del dataset._metadata[ name ]
+        if '__extension__' in JSONified_dict:
+            dataset.extension = JSONified_dict['__extension__']
+
+    def to_JSON_dict( self, filename=None ):
+        # galaxy.model.customtypes.json_encoder.encode()
+        meta_dict = {}
+        dataset_meta_dict = self.parent._metadata
+        for name, spec in self.spec.items():
+            if name in dataset_meta_dict:
+                meta_dict[ name ] = spec.param.to_external_value( dataset_meta_dict[ name ] )
+        if '__extension__' in dataset_meta_dict:
+            meta_dict[ '__extension__' ] = dataset_meta_dict['__extension__']
+        if filename is None:
+            return json.dumps( meta_dict )
+        json.dump( meta_dict, open( filename, 'wb+' ) )
+
+    def __getstate__( self ):
+        # cannot pickle a weakref item (self._parent), when
+        # data._metadata_collection is None, it will be recreated on demand
+        return None
+
+
+class MetadataSpecCollection( odict ):
+    """
+    A simple extension of dict which allows cleaner access to items
+    and allows the values to be iterated over directly as if it were a
+    list.  append() is also implemented for simplicity and does not
+    "append".
+    """
+    def __init__( self, dict=None ):
+        odict.__init__( self, dict=None )
+
+    def append( self, item ):
+        self[item.name] = item
+
+    def iter( self ):
+        return self.itervalues()
+
+    def __getattr__( self, name ):
+        return self.get( name )
+
+    def __repr__( self ):
+        # force elements to draw with __str__ for sphinx-apidoc
+        return ', '.join([ item.__str__() for item in self.iter() ])
+
+
+class MetadataParameter( object ):
+    def __init__( self, spec ):
+        self.spec = spec
+
+    def get_html_field( self, value=None, context=None, other_values=None, **kwd ):
+        context = context or {}
+        other_values = other_values or {}
+        return form_builder.TextField( self.spec.name, value=value )
+
+    def get_html( self, value, context=None, other_values=None, **kwd ):
+        """
+        The "context" is simply the metadata collection/bunch holding
+        this piece of metadata. This is passed in to allow for
+        metadata to validate against each other (note: this could turn
+        into a huge, recursive mess if not done with care). For
+        example, a column assignment should validate against the
+        number of columns in the dataset.
+        """
+        context = context or {}
+        other_values = other_values or {}
+
+        if self.spec.get("readonly"):
+            return value
+        if self.spec.get("optional"):
+            checked = False
+            if value:
+                checked = "true"
+            checkbox = form_builder.CheckboxField( "is_" + self.spec.name, checked=checked )
+            return checkbox.get_html() + self.get_html_field( value=value, context=context, other_values=other_values, **kwd ).get_html()
+        else:
+            return self.get_html_field( value=value, context=context, other_values=other_values, **kwd ).get_html()
+
+    def to_string( self, value ):
+        return str( value )
+
+    def to_safe_string( self, value ):
+        return sanitize_lists_to_string( self.to_string( value ) )
+
+    def make_copy( self, value, target_context=None, source_context=None ):
+        return copy.deepcopy( value )
+
+    @classmethod
+    def marshal( cls, value ):
+        """
+        This method should/can be overridden to convert the incoming
+        value to whatever type it is supposed to be.
+        """
+        return value
+
+    def validate( self, value ):
+        """
+        Throw an exception if the value is invalid.
+        """
+        pass
+
+    def unwrap( self, form_value ):
+        """
+        Turns a value into its storable form.
+        """
+        value = self.marshal( form_value )
+        self.validate( value )
+        return value
+
+    def wrap( self, value, session ):
+        """
+        Turns a value into its usable form.
+        """
+        return value
+
+    def from_external_value( self, value, parent ):
+        """
+        Turns a value read from an external dict into its value to be pushed directly into the metadata dict.
+        """
+        return value
+
+    def to_external_value( self, value ):
+        """
+        Turns a value read from a metadata into its value to be pushed directly into the external dict.
+        """
+        return value
+
+
+class MetadataElementSpec( object ):
+    """
+    Defines a metadata element and adds it to the metadata_spec (which
+    is a MetadataSpecCollection) of datatype.
+    """
+    def __init__( self, datatype, name=None, desc=None,
+                  param=MetadataParameter, default=None, no_value=None,
+                  visible=True, set_in_upload=False, **kwargs ):
+        self.name = name
+        self.desc = desc or name
+        self.default = default
+        self.no_value = no_value
+        self.visible = visible
+        self.set_in_upload = set_in_upload
+        # Catch-all, allows for extra attributes to be set
+        self.__dict__.update(kwargs)
+        # set up param last, as it uses values set above
+        self.param = param( self )
+        # add spec element to the spec
+        datatype.metadata_spec.append( self )
+
+    def get( self, name, default=None ):
+        return self.__dict__.get(name, default)
+
+    def wrap( self, value, session ):
+        """
+        Turns a stored value into its usable form.
+        """
+        return self.param.wrap( value, session )
+
+    def unwrap( self, value ):
+        """
+        Turns an incoming value into its storable form.
+        """
+        return self.param.unwrap( value )
+
+    def __str__( self ):
+        # TODO??: assuming param is the class of this MetadataElementSpec - add the plain class name for that
+        spec_dict = dict( param_class=self.param.__class__.__name__ )
+        spec_dict.update( self.__dict__ )
+        return ( "{name} ({param_class}): {desc}, defaults to '{default}'".format( **spec_dict ) )
+
+# create a statement class that, when called,
+#   will add a new MetadataElementSpec to a class's metadata_spec
+MetadataElement = Statement( MetadataElementSpec )
+
+
+"""
+MetadataParameter sub-classes.
+"""
+
+
+class SelectParameter( MetadataParameter ):
+    def __init__( self, spec ):
+        MetadataParameter.__init__( self, spec )
+        self.values = self.spec.get( "values" )
+        self.multiple = string_as_bool( self.spec.get( "multiple" ) )
+
+    def to_string( self, value ):
+        if value in [ None, [] ]:
+            return str( self.spec.no_value )
+        if not isinstance( value, list ):
+            value = [value]
+        return ",".join( map( str, value ) )
+
+    def get_html_field( self, value=None, context=None, other_values=None, values=None, **kwd ):
+        context = context or {}
+        other_values = other_values or {}
+
+        field = form_builder.SelectField( self.spec.name, multiple=self.multiple, display=self.spec.get("display") )
+        if self.values:
+            value_list = self.values
+        elif values:
+            value_list = values
+        elif value:
+            value_list = [ ( v, v ) for v in listify( value )]
+        else:
+            value_list = []
+        for val, label in value_list:
+            try:
+                if ( self.multiple and val in value ) or ( not self.multiple and val == value ):
+                    field.add_option( label, val, selected=True )
+                else:
+                    field.add_option( label, val, selected=False )
+            except TypeError:
+                field.add_option( val, label, selected=False )
+        return field
+
+    def get_html( self, value, context=None, other_values=None, values=None, **kwd ):
+        context = context or {}
+        other_values = other_values or {}
+
+        if self.spec.get("readonly"):
+            if value in [ None, [] ]:
+                return str( self.spec.no_value )
+            return ", ".join( map( str, value ) )
+        return MetadataParameter.get_html( self, value, context=context, other_values=other_values, values=values, **kwd )
+
+    def wrap( self, value, session ):
+        # do we really need this (wasteful)? - yes because we are not sure that
+        # all existing selects have been stored previously as lists. Also this
+        # will handle the case where defaults/no_values are specified and are
+        # single non-list values.
+        value = self.marshal( value )
+        if self.multiple:
+            return value
+        elif value:
+            return value[0]  # single select, only return the first value
+        return None
+
+    @classmethod
+    def marshal( cls, value ):
+        # Store select as list, even if single item
+        if value is None:
+            return []
+        if not isinstance( value, list ):
+            return [value]
+        return value
+
+
+class DBKeyParameter( SelectParameter ):
+
+    def get_html_field( self, value=None, context=None, other_values=None, values=None, **kwd):
+        context = context or {}
+        other_values = other_values or {}
+        try:
+            values = kwd['trans'].app.genome_builds.get_genome_build_names( kwd['trans'] )
+        except KeyError:
+            pass
+        return super(DBKeyParameter, self).get_html_field( value, context, other_values, values, **kwd)
+
+    def get_html( self, value=None, context=None, other_values=None, values=None, **kwd):
+        context = context or {}
+        other_values = other_values or {}
+        try:
+            values = kwd['trans'].app.genome_builds.get_genome_build_names( kwd['trans'] )
+        except KeyError:
+            pass
+        return super(DBKeyParameter, self).get_html( value, context, other_values, values, **kwd)
+
+
+class RangeParameter( SelectParameter ):
+
+    def __init__( self, spec ):
+        SelectParameter.__init__( self, spec )
+        # The spec must be set with min and max values
+        self.min = spec.get( "min" ) or 1
+        self.max = spec.get( "max" ) or 1
+        self.step = self.spec.get( "step" ) or 1
+
+    def get_html_field( self, value=None, context=None, other_values=None, values=None, **kwd ):
+        context = context or {}
+        other_values = other_values or {}
+
+        if values is None:
+            values = zip( range( self.min, self.max, self.step ), range( self.min, self.max, self.step ))
+        return SelectParameter.get_html_field( self, value=value, context=context, other_values=other_values, values=values, **kwd )
+
+    def get_html( self, value, context=None, other_values=None, values=None, **kwd ):
+        context = context or {}
+        other_values = other_values or {}
+
+        if values is None:
+            values = zip( range( self.min, self.max, self.step ), range( self.min, self.max, self.step ))
+        return SelectParameter.get_html( self, value, context=context, other_values=other_values, values=values, **kwd )
+
+    @classmethod
+    def marshal( cls, value ):
+        value = SelectParameter.marshal( value )
+        values = [ int(x) for x in value ]
+        return values
+
+
+class ColumnParameter( RangeParameter ):
+
+    def get_html_field( self, value=None, context=None, other_values=None, values=None, **kwd ):
+        context = context or {}
+        other_values = other_values or {}
+
+        if values is None and context:
+            column_range = range( 1, ( context.columns or 0 ) + 1, 1 )
+            values = zip( column_range, column_range )
+        return RangeParameter.get_html_field( self, value=value, context=context, other_values=other_values, values=values, **kwd )
+
+    def get_html( self, value, context=None, other_values=None, values=None, **kwd ):
+        context = context or {}
+        other_values = other_values or {}
+
+        if values is None and context:
+            column_range = range( 1, ( context.columns or 0 ) + 1, 1 )
+            values = zip( column_range, column_range )
+        return RangeParameter.get_html( self, value, context=context, other_values=other_values, values=values, **kwd )
+
+
+class ColumnTypesParameter( MetadataParameter ):
+
+    def to_string( self, value ):
+        return ",".join( map( str, value ) )
+
+
+class ListParameter( MetadataParameter ):
+
+    def to_string( self, value ):
+        return ",".join( [str(x) for x in value] )
+
+
+class DictParameter( MetadataParameter ):
+
+    def to_string( self, value ):
+        return json.dumps( value )
+
+    def to_safe_string( self, value ):
+        # We do not sanitize json dicts
+        return safe_dumps( value )
+
+
+class PythonObjectParameter( MetadataParameter ):
+
+    def to_string( self, value ):
+        if not value:
+            return self.spec._to_string( self.spec.no_value )
+        return self.spec._to_string( value )
+
+    def get_html_field( self, value=None, context=None, other_values=None, **kwd ):
+        context = context or {}
+        other_values = other_values or {}
+        return form_builder.TextField( self.spec.name, value=self._to_string( value ) )
+
+    def get_html( self, value=None, context=None, other_values=None, **kwd ):
+        context = context or {}
+        other_values = other_values or {}
+        return str( self )
+
+    @classmethod
+    def marshal( cls, value ):
+        return value
+
+
+class FileParameter( MetadataParameter ):
+
+    def to_string( self, value ):
+        if not value:
+            return str( self.spec.no_value )
+        return value.file_name
+
+    def to_safe_string( self, value ):
+        # We do not sanitize file names
+        return self.to_string( value )
+
+    def get_html_field( self, value=None, context=None, other_values=None, **kwd ):
+        context = context or {}
+        other_values = other_values or {}
+        return form_builder.TextField( self.spec.name, value=str( value.id ) )
+
+    def get_html( self, value=None, context=None, other_values=None, **kwd ):
+        context = context or {}
+        other_values = other_values or {}
+        return "<div>No display available for Metadata Files</div>"
+
+    def wrap( self, value, session ):
+        if value is None:
+            return None
+        if isinstance( value, galaxy.model.MetadataFile ) or isinstance( value, MetadataTempFile ):
+            return value
+        mf = session.query( galaxy.model.MetadataFile ).get( value )
+        return mf
+
+    def make_copy( self, value, target_context, source_context ):
+        value = self.wrap( value, object_session( target_context.parent ) )
+        if value:
+            new_value = galaxy.model.MetadataFile( dataset=target_context.parent, name=self.spec.name )
+            object_session( target_context.parent ).add( new_value )
+            object_session( target_context.parent ).flush()
+            shutil.copy( value.file_name, new_value.file_name )
+            return self.unwrap( new_value )
+        return None
+
+    @classmethod
+    def marshal( cls, value ):
+        if isinstance( value, galaxy.model.MetadataFile ):
+            value = value.id
+        return value
+
+    def from_external_value( self, value, parent, path_rewriter=None ):
+        """
+        Turns a value read from a external dict into its value to be pushed directly into the metadata dict.
+        """
+        if MetadataTempFile.is_JSONified_value( value ):
+            value = MetadataTempFile.from_JSON( value )
+        if isinstance( value, MetadataTempFile ):
+            mf = parent.metadata.get( self.spec.name, None)
+            if mf is None:
+                mf = self.new_file( dataset=parent, **value.kwds )
+            # Ensure the metadata file gets updated with content
+            file_name = value.file_name
+            if path_rewriter:
+                # Job may have run with a different (non-local) tmp/working
+                # directory. Correct.
+                file_name = path_rewriter( file_name )
+            parent.dataset.object_store.update_from_file( mf,
+                                                          file_name=file_name,
+                                                          extra_dir='_metadata_files',
+                                                          extra_dir_at_root=True,
+                                                          alt_name=os.path.basename(mf.file_name) )
+            os.unlink( file_name )
+            value = mf.id
+        return value
+
+    def to_external_value( self, value ):
+        """
+        Turns a value read from a metadata into its value to be pushed directly into the external dict.
+        """
+        if isinstance( value, galaxy.model.MetadataFile ):
+            value = value.id
+        elif isinstance( value, MetadataTempFile ):
+            value = MetadataTempFile.to_JSON( value )
+        return value
+
+    def new_file( self, dataset=None, **kwds ):
+        if object_session( dataset ):
+            mf = galaxy.model.MetadataFile( name=self.spec.name, dataset=dataset, **kwds )
+            object_session( dataset ).add( mf )
+            object_session( dataset ).flush()  # flush to assign id
+            return mf
+        else:
+            # we need to make a tmp file that is accessable to the head node,
+            # we will be copying its contents into the MetadataFile objects filename after restoring from JSON
+            # we do not include 'dataset' in the kwds passed, as from_JSON_value() will handle this for us
+            return MetadataTempFile( **kwds )
+
+
+# This class is used when a database file connection is not available
+class MetadataTempFile( object ):
+    tmp_dir = 'database/tmp'  # this should be overwritten as necessary in calling scripts
+
+    def __init__( self, **kwds ):
+        self.kwds = kwds
+        self._filename = None
+
+    @property
+    def file_name( self ):
+        if self._filename is None:
+            # we need to create a tmp file, accessable across all nodes/heads, save the name, and return it
+            self._filename = abspath( tempfile.NamedTemporaryFile( dir=self.tmp_dir, prefix="metadata_temp_file_" ).name )
+            open( self._filename, 'wb+' )  # create an empty file, so it can't be reused using tempfile
+        return self._filename
+
+    def to_JSON( self ):
+        return { '__class__': self.__class__.__name__,
+                 'filename': self.file_name,
+                 'kwds': self.kwds }
+
+    @classmethod
+    def from_JSON( cls, json_dict ):
+        # need to ensure our keywords are not unicode
+        rval = cls( **stringify_dictionary_keys( json_dict['kwds'] ) )
+        rval._filename = json_dict['filename']
+        return rval
+
+    @classmethod
+    def is_JSONified_value( cls, value ):
+        return ( isinstance( value, dict ) and value.get( '__class__', None ) == cls.__name__ )
+
+    @classmethod
+    def cleanup_from_JSON_dict_filename( cls, filename ):
+        try:
+            for key, value in json.load( open( filename ) ).items():
+                if cls.is_JSONified_value( value ):
+                    value = cls.from_JSON( value )
+                if isinstance( value, cls ) and os.path.exists( value.file_name ):
+                    log.debug( 'Cleaning up abandoned MetadataTempFile file: %s' % value.file_name )
+                    os.unlink( value.file_name )
+        except Exception as e:
+            log.debug( 'Failed to cleanup MetadataTempFile temp files from %s: %s' % ( filename, e ) )
+
+
+class JobExternalOutputMetadataWrapper( object ):
+    """
+    Class with methods allowing set_meta() to be called externally to the
+    Galaxy head.
+    This class allows access to external metadata filenames for all outputs
+    associated with a job.
+    We will use JSON as the medium of exchange of information, except for the
+    DatasetInstance object which will use pickle (in the future this could be
+    JSONified as well)
+    """
+
+    def __init__( self, job ):
+        self.job_id = job.id
+
+    def get_output_filenames_by_dataset( self, dataset, sa_session ):
+        if isinstance( dataset, galaxy.model.HistoryDatasetAssociation ):
+            return sa_session.query( galaxy.model.JobExternalOutputMetadata ) \
+                             .filter_by( job_id=self.job_id,
+                                         history_dataset_association_id=dataset.id,
+                                         is_valid=True ) \
+                             .first()  # there should only be one or None
+        elif isinstance( dataset, galaxy.model.LibraryDatasetDatasetAssociation ):
+            return sa_session.query( galaxy.model.JobExternalOutputMetadata ) \
+                             .filter_by( job_id=self.job_id,
+                                         library_dataset_dataset_association_id=dataset.id,
+                                         is_valid=True ) \
+                             .first()  # there should only be one or None
+        return None
+
+    def get_dataset_metadata_key( self, dataset ):
+        # Set meta can be called on library items and history items,
+        # need to make different keys for them, since ids can overlap
+        return "%s_%d" % ( dataset.__class__.__name__, dataset.id )
+
+    def invalidate_external_metadata( self, datasets, sa_session ):
+        for dataset in datasets:
+            jeom = self.get_output_filenames_by_dataset( dataset, sa_session )
+            # shouldn't be more than one valid, but you never know
+            while jeom:
+                jeom.is_valid = False
+                sa_session.add( jeom )
+                sa_session.flush()
+                jeom = self.get_output_filenames_by_dataset( dataset, sa_session )
+
+    def setup_external_metadata( self, datasets, sa_session, exec_dir=None,
+                                 tmp_dir=None, dataset_files_path=None,
+                                 output_fnames=None, config_root=None,
+                                 config_file=None, datatypes_config=None,
+                                 job_metadata=None, compute_tmp_dir=None,
+                                 include_command=True, max_metadata_value_size=0,
+                                 kwds=None):
+        kwds = kwds or {}
+        if tmp_dir is None:
+            tmp_dir = MetadataTempFile.tmp_dir
+        else:
+            MetadataTempFile.tmp_dir = tmp_dir
+
+        if not os.path.exists(tmp_dir):
+            os.makedirs(tmp_dir)
+
+        # path is calculated for Galaxy, may be different on compute - rewrite
+        # for the compute server.
+        def metadata_path_on_compute(path):
+            compute_path = path
+            if compute_tmp_dir and tmp_dir and in_directory(path, tmp_dir):
+                path_relative = os.path.relpath(path, tmp_dir)
+                compute_path = os.path.join(compute_tmp_dir, path_relative)
+            return compute_path
+
+        # fill in metadata_files_dict and return the command with args required to set metadata
+        def __metadata_files_list_to_cmd_line( metadata_files ):
+            def __get_filename_override():
+                if output_fnames:
+                    for dataset_path in output_fnames:
+                        if dataset_path.real_path == metadata_files.dataset.file_name:
+                            return dataset_path.false_path or dataset_path.real_path
+                return ""
+            line = '"%s,%s,%s,%s,%s,%s"' % (
+                metadata_path_on_compute(metadata_files.filename_in),
+                metadata_path_on_compute(metadata_files.filename_kwds),
+                metadata_path_on_compute(metadata_files.filename_out),
+                metadata_path_on_compute(metadata_files.filename_results_code),
+                __get_filename_override(),
+                metadata_path_on_compute(metadata_files.filename_override_metadata),
+            )
+            return line
+        if not isinstance( datasets, list ):
+            datasets = [ datasets ]
+        if exec_dir is None:
+            exec_dir = os.path.abspath( os.getcwd() )
+        if dataset_files_path is None:
+            dataset_files_path = galaxy.model.Dataset.file_path
+        if config_root is None:
+            config_root = os.path.abspath( os.getcwd() )
+        if datatypes_config is None:
+            raise Exception( 'In setup_external_metadata, the received datatypes_config is None.' )
+            datatypes_config = 'datatypes_conf.xml'
+        metadata_files_list = []
+        for dataset in datasets:
+            key = self.get_dataset_metadata_key( dataset )
+            # future note:
+            # wonkiness in job execution causes build command line to be called more than once
+            # when setting metadata externally, via 'auto-detect' button in edit attributes, etc.,
+            # we don't want to overwrite (losing the ability to cleanup) our existing dataset keys and files,
+            # so we will only populate the dictionary once
+            metadata_files = self.get_output_filenames_by_dataset( dataset, sa_session )
+            if not metadata_files:
+                job = sa_session.query( galaxy.model.Job ).get( self.job_id )
+                metadata_files = galaxy.model.JobExternalOutputMetadata( job=job, dataset=dataset )
+                # we are using tempfile to create unique filenames, tempfile always returns an absolute path
+                # we will use pathnames relative to the galaxy root, to accommodate instances where the galaxy root
+                # is located differently, i.e. on a cluster node with a different filesystem structure
+
+                # file to store existing dataset
+                metadata_files.filename_in = abspath( tempfile.NamedTemporaryFile( dir=tmp_dir, prefix="metadata_in_%s_" % key ).name )
+
+                # FIXME: HACK
+                # sqlalchemy introduced 'expire_on_commit' flag for sessionmaker at version 0.5x
+                # This may be causing the dataset attribute of the dataset_association object to no-longer be loaded into memory when needed for pickling.
+                # For now, we'll simply 'touch' dataset_association.dataset to force it back into memory.
+                dataset.dataset  # force dataset_association.dataset to be loaded before pickling
+                # A better fix could be setting 'expire_on_commit=False' on the session, or modifying where commits occur, or ?
+
+                # Touch also deferred column
+                dataset._metadata
+
+                cPickle.dump( dataset, open( metadata_files.filename_in, 'wb+' ) )
+                # file to store metadata results of set_meta()
+                metadata_files.filename_out = abspath( tempfile.NamedTemporaryFile( dir=tmp_dir, prefix="metadata_out_%s_" % key ).name )
+                open( metadata_files.filename_out, 'wb+' )  # create the file on disk, so it cannot be reused by tempfile (unlikely, but possible)
+                # file to store a 'return code' indicating the results of the set_meta() call
+                # results code is like (True/False - if setting metadata was successful/failed , exception or string of reason of success/failure )
+                metadata_files.filename_results_code = abspath( tempfile.NamedTemporaryFile( dir=tmp_dir, prefix="metadata_results_%s_" % key ).name )
+                # create the file on disk, so it cannot be reused by tempfile (unlikely, but possible)
+                json.dump( ( False, 'External set_meta() not called' ), open( metadata_files.filename_results_code, 'wb+' ) )
+                # file to store kwds passed to set_meta()
+                metadata_files.filename_kwds = abspath( tempfile.NamedTemporaryFile( dir=tmp_dir, prefix="metadata_kwds_%s_" % key ).name )
+                json.dump( kwds, open( metadata_files.filename_kwds, 'wb+' ), ensure_ascii=True )
+                # existing metadata file parameters need to be overridden with cluster-writable file locations
+                metadata_files.filename_override_metadata = abspath( tempfile.NamedTemporaryFile( dir=tmp_dir, prefix="metadata_override_%s_" % key ).name )
+                open( metadata_files.filename_override_metadata, 'wb+' )  # create the file on disk, so it cannot be reused by tempfile (unlikely, but possible)
+                override_metadata = []
+                for meta_key, spec_value in dataset.metadata.spec.iteritems():
+                    if isinstance( spec_value.param, FileParameter ) and dataset.metadata.get( meta_key, None ) is not None:
+                        metadata_temp = MetadataTempFile()
+                        shutil.copy( dataset.metadata.get( meta_key, None ).file_name, metadata_temp.file_name )
+                        override_metadata.append( ( meta_key, metadata_temp.to_JSON() ) )
+                json.dump( override_metadata, open( metadata_files.filename_override_metadata, 'wb+' ) )
+                # add to session and flush
+                sa_session.add( metadata_files )
+                sa_session.flush()
+            metadata_files_list.append( metadata_files )
+        args = '"%s" "%s" %s %s' % ( datatypes_config,
+                                     job_metadata,
+                                     " ".join( map( __metadata_files_list_to_cmd_line, metadata_files_list ) ),
+                                     max_metadata_value_size)
+        if include_command:
+            # return command required to build
+            fd, fp = tempfile.mkstemp( suffix='.py', dir=tmp_dir, prefix="set_metadata_" )
+            metadata_script_file = abspath( fp )
+            os.fdopen( fd, 'w' ).write( 'from galaxy_ext.metadata.set_metadata import set_metadata; set_metadata()' )
+            return 'python "%s" %s' % ( metadata_path_on_compute(metadata_script_file), args )
+        else:
+            # return args to galaxy_ext.metadata.set_metadata required to build
+            return args
+
+    def external_metadata_set_successfully( self, dataset, sa_session ):
+        metadata_files = self.get_output_filenames_by_dataset( dataset, sa_session )
+        if not metadata_files:
+            return False  # this file doesn't exist
+        rval, rstring = json.load( open( metadata_files.filename_results_code ) )
+        if not rval:
+            log.debug( 'setting metadata externally failed for %s %s: %s' % ( dataset.__class__.__name__, dataset.id, rstring ) )
+        return rval
+
+    def cleanup_external_metadata( self, sa_session ):
+        log.debug( 'Cleaning up external metadata files' )
+        for metadata_files in sa_session.query( galaxy.model.Job ).get( self.job_id ).external_output_metadata:
+            # we need to confirm that any MetadataTempFile files were removed, if not we need to remove them
+            # can occur if the job was stopped before completion, but a MetadataTempFile is used in the set_meta
+            MetadataTempFile.cleanup_from_JSON_dict_filename( metadata_files.filename_out )
+            dataset_key = self.get_dataset_metadata_key( metadata_files.dataset )
+            for key, fname in [ ( 'filename_in', metadata_files.filename_in ),
+                                ( 'filename_out', metadata_files.filename_out ),
+                                ( 'filename_results_code', metadata_files.filename_results_code ),
+                                ( 'filename_kwds', metadata_files.filename_kwds ),
+                                ( 'filename_override_metadata', metadata_files.filename_override_metadata ) ]:
+                try:
+                    os.remove( fname )
+                except Exception as e:
+                    log.debug( 'Failed to cleanup external metadata file (%s) for %s: %s' % ( key, dataset_key, e ) )
+
+    def set_job_runner_external_pid( self, pid, sa_session ):
+        for metadata_files in sa_session.query( galaxy.model.Job ).get( self.job_id ).external_output_metadata:
+            metadata_files.job_runner_external_pid = pid
+            sa_session.add( metadata_files )
+            sa_session.flush()
+
+__all__ = [
+    "Statement",
+    "MetadataElement",
+    "MetadataCollection",
+    "MetadataSpecCollection",
+    "MetadataParameter",
+    "MetadataElementSpec",
+    "SelectParameter",
+    "DBKeyParameter",
+    "RangeParameter",
+    "ColumnParameter",
+    "ColumnTypesParameter",
+    "ListParameter",
+    "DictParameter",
+    "PythonObjectParameter",
+    "FileParameter",
+    "MetadataTempFile",
+    "JobExternalOutputMetadataWrapper",
+]
diff --git a/lib/galaxy/model/migrate/versions/0005_cleanup_datasets_fix.py b/lib/galaxy/model/migrate/versions/0005_cleanup_datasets_fix.py
index 2a8042eb25e..65f86d6d3a7 100644
--- a/lib/galaxy/model/migrate/versions/0005_cleanup_datasets_fix.py
+++ b/lib/galaxy/model/migrate/versions/0005_cleanup_datasets_fix.py
@@ -7,7 +7,7 @@ import time
 from sqlalchemy import and_, Boolean, Column, DateTime, false, ForeignKey, Integer, MetaData, not_, Numeric, Table, TEXT, true
 from sqlalchemy.orm import backref, mapper, relation, scoped_session, sessionmaker
 
-from galaxy.datatypes.metadata import MetadataCollection
+from galaxy.model.metadata import MetadataCollection
 from galaxy.model.custom_types import MetadataType, TrimmedString
 from galaxy.util.bunch import Bunch
 
diff --git a/lib/galaxy/tools/exception_handling.py b/lib/galaxy/tools/exception_handling.py
index ad62ea763b2..a9017ad3d3d 100644
--- a/lib/galaxy/tools/exception_handling.py
+++ b/lib/galaxy/tools/exception_handling.py
@@ -1,38 +1,6 @@
-"""
-Exceptions and handlers for tools.
+# We put a tool that references this package into the tool shed
+# so we have to provide this legacy location for import indefinitely
+# it seems.
+from galaxy.util.ucsc import UCSCOutWrapper, UCSCLimitException
 
-FIXME: These are used by tool scripts, not the framework, and should not live
-       in this package.
-"""
-
-
-class UCSCLimitException( Exception ):
-    pass
-
-
-class UCSCOutWrapper( object ):
-    """File-like object that throws an exception if it encounters the UCSC limit error lines"""
-    def __init__( self, other ):
-        self.other = iter( other )
-        # Need one line of lookahead to be sure we are hitting the limit message
-        self.lookahead = None
-
-    def __iter__( self ):
-        return self
-
-    def next( self ):
-        if self.lookahead is None:
-            line = self.other.next()
-        else:
-            line = self.lookahead
-            self.lookahead = None
-        if line.startswith( "----------" ):
-            next_line = self.other.next()
-            if next_line.startswith( "Reached output limit" ):
-                raise UCSCLimitException( next_line.strip() )
-            else:
-                self.lookahead = next_line
-        return line
-
-    def readline(self):
-        return self.next()
+__all__ = ['UCSCOutWrapper', 'UCSCLimitException']
diff --git a/lib/galaxy/util/none_like.py b/lib/galaxy/util/none_like.py
index 254bddbfdf2..5b3d41dbb8b 100644
--- a/lib/galaxy/util/none_like.py
+++ b/lib/galaxy/util/none_like.py
@@ -1,7 +1,7 @@
 """
 Objects with No values
 """
-from galaxy.datatypes.metadata import MetadataCollection
+from galaxy.model.metadata import MetadataCollection
 from galaxy.datatypes.registry import Registry
 
 
diff --git a/lib/galaxy/util/ucsc.py b/lib/galaxy/util/ucsc.py
new file mode 100644
index 00000000000..4869a6ec870
--- /dev/null
+++ b/lib/galaxy/util/ucsc.py
@@ -0,0 +1,35 @@
+"""
+Utilities for dealing with UCSC data.
+"""
+
+
+class UCSCLimitException( Exception ):
+    pass
+
+
+class UCSCOutWrapper( object ):
+    """File-like object that throws an exception if it encounters the UCSC limit error lines"""
+    def __init__( self, other ):
+        self.other = iter( other )
+        # Need one line of lookahead to be sure we are hitting the limit message
+        self.lookahead = None
+
+    def __iter__( self ):
+        return self
+
+    def next( self ):
+        if self.lookahead is None:
+            line = self.other.next()
+        else:
+            line = self.lookahead
+            self.lookahead = None
+        if line.startswith( "----------" ):
+            next_line = self.other.next()
+            if next_line.startswith( "Reached output limit" ):
+                raise UCSCLimitException( next_line.strip() )
+            else:
+                self.lookahead = next_line
+        return line
+
+    def readline(self):
+        return self.next()
diff --git a/scripts/set_metadata.py b/scripts/set_metadata.py
deleted file mode 100644
index 60c04a12dc6..00000000000
--- a/scripts/set_metadata.py
+++ /dev/null
@@ -1,153 +0,0 @@
-"""
-Execute an external process to set_meta() on a provided list of pickled datasets.
-
-This should not be called directly!  Use the set_metadata.sh script in Galaxy's
-top level directly.
-
-"""
-
-import logging
-logging.basicConfig()
-log = logging.getLogger( __name__ )
-
-import cPickle
-import json
-import os
-import sys
-
-# ensure supported version
-from check_python import check_python
-try:
-    check_python()
-except:
-    sys.exit(1)
-
-new_path = [ os.path.join( os.getcwd(), "lib" ) ]
-new_path.extend( sys.path[ 1: ] )  # remove scripts/ from the path
-sys.path = new_path
-
-import galaxy.model.mapping  # need to load this before we unpickle, in order to setup properties assigned by the mappers
-
-# This looks REAL stupid, but it is REQUIRED in order for SA to insert
-# parameters into the classes defined by the mappers --> it appears that
-# instantiating ANY mapper'ed class would suffice here
-galaxy.model.Job()
-
-from galaxy.util import stringify_dictionary_keys
-from sqlalchemy.orm import clear_mappers
-from galaxy.objectstore import build_object_store_from_config
-from galaxy import config
-from galaxy.util.properties import load_app_properties
-
-
-def set_meta_with_tool_provided( dataset_instance, file_dict, set_meta_kwds ):
-    # This method is somewhat odd, in that we set the metadata attributes from tool,
-    # then call set_meta, then set metadata attributes from tool again.
-    # This is intentional due to interplay of overwrite kwd, the fact that some metadata
-    # parameters may rely on the values of others, and that we are accepting the
-    # values provided by the tool as Truth.
-    for metadata_name, metadata_value in file_dict.get( 'metadata', {} ).iteritems():
-        setattr( dataset_instance.metadata, metadata_name, metadata_value )
-    dataset_instance.datatype.set_meta( dataset_instance, **set_meta_kwds )
-    for metadata_name, metadata_value in file_dict.get( 'metadata', {} ).iteritems():
-        setattr( dataset_instance.metadata, metadata_name, metadata_value )
-
-
-def __main__():
-    file_path = sys.argv.pop( 1 )
-    tool_job_working_directory = tmp_dir = sys.argv.pop( 1 )  # this is also the job_working_directory now
-    galaxy.model.Dataset.file_path = file_path
-    galaxy.datatypes.metadata.MetadataTempFile.tmp_dir = tmp_dir
-
-    config_root = sys.argv.pop( 1 )
-    config_file_name = sys.argv.pop( 1 )
-    if not os.path.isabs( config_file_name ):
-        config_file_name = os.path.join( config_root, config_file_name )
-
-    # Set up reference to object store
-    # First, read in the main config file for Galaxy; this is required because
-    # the object store configuration is stored there
-    conf_dict = load_app_properties( ini_file=config_file_name )
-    # config object is required by ObjectStore class so create it now
-    universe_config = config.Configuration(**conf_dict)
-    universe_config.ensure_tempdir()
-    object_store = build_object_store_from_config(universe_config)
-    galaxy.model.Dataset.object_store = object_store
-
-    # Set up datatypes registry
-    datatypes_config = sys.argv.pop( 1 )
-    datatypes_registry = galaxy.datatypes.registry.Registry()
-    datatypes_registry.load_datatypes( root_dir=config_root, config=datatypes_config )
-    galaxy.model.set_datatypes_registry( datatypes_registry )
-
-    job_metadata = sys.argv.pop( 1 )
-    existing_job_metadata_dict = {}
-    new_job_metadata_dict = {}
-    if job_metadata != "None" and os.path.exists( job_metadata ):
-        for line in open( job_metadata, 'r' ):
-            try:
-                line = stringify_dictionary_keys( json.loads( line ) )
-                if line['type'] == 'dataset':
-                    existing_job_metadata_dict[ line['dataset_id'] ] = line
-                elif line['type'] == 'new_primary_dataset':
-                    new_job_metadata_dict[ line[ 'filename' ] ] = line
-            except:
-                continue
-
-    for filenames in sys.argv[1:]:
-        fields = filenames.split( ',' )
-        filename_in = fields.pop( 0 )
-        filename_kwds = fields.pop( 0 )
-        filename_out = fields.pop( 0 )
-        filename_results_code = fields.pop( 0 )
-        dataset_filename_override = fields.pop( 0 )
-        # Need to be careful with the way that these parameters are populated from the filename splitting,
-        # because if a job is running when the server is updated, any existing external metadata command-lines
-        # will not have info about the newly added override_metadata file
-        if fields:
-            override_metadata = fields.pop( 0 )
-        else:
-            override_metadata = None
-        set_meta_kwds = stringify_dictionary_keys( json.load( open( filename_kwds ) ) )  # load kwds; need to ensure our keywords are not unicode
-        try:
-            dataset = cPickle.load( open( filename_in ) )  # load DatasetInstance
-            if dataset_filename_override:
-                dataset.dataset.external_filename = dataset_filename_override
-            files_path = os.path.abspath(os.path.join( tool_job_working_directory, "dataset_%s_files" % (dataset.dataset.id) ))
-            dataset.dataset.external_extra_files_path = files_path
-            if dataset.dataset.id in existing_job_metadata_dict:
-                dataset.extension = existing_job_metadata_dict[ dataset.dataset.id ].get( 'ext', dataset.extension )
-            # Metadata FileParameter types may not be writable on a cluster node, and are therefore temporarily substituted with MetadataTempFiles
-            if override_metadata:
-                override_metadata = json.load( open( override_metadata ) )
-                for metadata_name, metadata_file_override in override_metadata:
-                    if galaxy.datatypes.metadata.MetadataTempFile.is_JSONified_value( metadata_file_override ):
-                        metadata_file_override = galaxy.datatypes.metadata.MetadataTempFile.from_JSON( metadata_file_override )
-                    setattr( dataset.metadata, metadata_name, metadata_file_override )
-            file_dict = existing_job_metadata_dict.get( dataset.dataset.id, {} )
-            set_meta_with_tool_provided( dataset, file_dict, set_meta_kwds )
-            dataset.metadata.to_JSON_dict( filename_out )  # write out results of set_meta
-            json.dump( ( True, 'Metadata has been set successfully' ), open( filename_results_code, 'wb+' ) )  # setting metadata has succeeded
-        except Exception, e:
-            json.dump( ( False, str( e ) ), open( filename_results_code, 'wb+' ) )  # setting metadata has failed somehow
-
-    for i, ( filename, file_dict ) in enumerate( new_job_metadata_dict.iteritems(), start=1 ):
-        new_dataset = galaxy.model.Dataset( id=-i, external_filename=os.path.join( tool_job_working_directory, file_dict[ 'filename' ] ) )
-        extra_files = file_dict.get( 'extra_files', None )
-        if extra_files is not None:
-            new_dataset._extra_files_path = os.path.join( tool_job_working_directory, extra_files )
-        new_dataset.state = new_dataset.states.OK
-        new_dataset_instance = galaxy.model.HistoryDatasetAssociation( id=-i, dataset=new_dataset, extension=file_dict.get( 'ext', 'data' ) )
-        set_meta_with_tool_provided( new_dataset_instance, file_dict, set_meta_kwds )
-        # storing metadata in external form, need to turn back into dict, then later jsonify
-        file_dict[ 'metadata' ] = json.loads( new_dataset_instance.metadata.to_JSON_dict() )
-    if existing_job_metadata_dict or new_job_metadata_dict:
-        with open( job_metadata, 'wb' ) as job_metadata_fh:
-            for value in existing_job_metadata_dict.values() + new_job_metadata_dict.values():
-                job_metadata_fh.write( "%s\n" % ( json.dumps( value ) ) )
-
-    clear_mappers()
-    # Shut down any additional threads that might have been created via the ObjectStore
-    object_store.shutdown()
-
-__main__()
diff --git a/set_metadata.sh b/set_metadata.sh
deleted file mode 100755
index bbc5fdbef1c..00000000000
--- a/set_metadata.sh
+++ /dev/null
@@ -1,6 +0,0 @@
-#!/bin/sh
-
-[ -d .venv ] && . .venv/bin/activate
-
-cd `dirname $0`
-python ./scripts/set_metadata.py $@
diff --git a/test/unit/tools/test_column_parameters.py b/test/unit/tools/test_column_parameters.py
index 44312dd1226..ac2cd968fe0 100644
--- a/test/unit/tools/test_column_parameters.py
+++ b/test/unit/tools/test_column_parameters.py
@@ -5,6 +5,7 @@ test_select_parameters.py.
 from galaxy.util import bunch
 from galaxy import model
 from .test_parameter_parsing import BaseParameterTestCase
+from tools_support import datatypes_registry
 
 
 class DataColumnParameterTestCase( BaseParameterTestCase ):
@@ -99,6 +100,6 @@ class DataColumnParameterTestCase( BaseParameterTestCase ):
             template_xml = '''<param name="my_name" type="%s" %s %s %s %s></param>'''
             param_str = template_xml % ( self.type, data_ref_text, multi_text, optional_text, self.other_attributes )
             self._param = self._parameter_for( xml=param_str )
-            self._param.ref_input = bunch.Bunch(formats=[model.datatypes_registry.get_datatype_by_extension("tabular")])
+            self._param.ref_input = bunch.Bunch(formats=[datatypes_registry.get_datatype_by_extension("tabular")])
 
         return self._param
diff --git a/test/unit/tools_support.py b/test/unit/tools_support.py
index 3d435ca1f68..c8f5a34211e 100644
--- a/test/unit/tools_support.py
+++ b/test/unit/tools_support.py
@@ -18,6 +18,11 @@ from galaxy.util.dbkeys import GenomeBuilds
 from galaxy.jobs import NoopQueue
 from galaxy.tools.parser import get_tool_source
 from galaxy.tools.deps.containers import NullContainerFinder
+import galaxy.datatypes.registry
+
+datatypes_registry = galaxy.datatypes.registry.Registry()
+datatypes_registry.load_datatypes()
+galaxy.model.set_datatypes_registry(datatypes_registry)
 
 
 class UsesApp( object ):
diff --git a/tools/data_source/data_source.py b/tools/data_source/data_source.py
index ccb0213c116..2d3dbcb9e5c 100644
--- a/tools/data_source/data_source.py
+++ b/tools/data_source/data_source.py
@@ -7,7 +7,6 @@ import sys
 import urllib
 from json import loads, dumps
 
-# need to import TOOL_PROVIDED_JOB_METADATA_FILE (which will import galaxy.model) before sniff to resolve a circular import between galaxy.datatypes.registry and galaxy.model
 from galaxy.jobs import TOOL_PROVIDED_JOB_METADATA_FILE
 from galaxy.datatypes import sniff
 from galaxy.datatypes.registry import Registry
diff --git a/tools/data_source/upload.py b/tools/data_source/upload.py
index 625a3cd974e..dd735fc6765 100644
--- a/tools/data_source/upload.py
+++ b/tools/data_source/upload.py
@@ -14,8 +14,6 @@ import urllib
 import zipfile
 
 from galaxy import util
-# need to import model before sniff to resolve a circular import dependency
-import galaxy.model  # noqa
 from galaxy.datatypes import sniff
 from galaxy.datatypes.binary import Binary
 from galaxy.datatypes.registry import Registry
@@ -126,7 +124,7 @@ def add_file( dataset, registry, json_file, output_path ):
     # Is dataset content multi-byte?
     elif dataset.is_multi_byte:
         data_type = 'multi-byte char'
-        ext = sniff.guess_ext( dataset.path, is_multi_byte=True )
+        ext = sniff.guess_ext( dataset.path, registry.sniff_order, is_multi_byte=True )
     # Is dataset content supported sniffable binary?
     else:
         # FIXME: This ignores the declared sniff order in datatype_conf.xml
diff --git a/tools/filters/wiggle_to_simple.py b/tools/filters/wiggle_to_simple.py
index 58760ca4eac..ca18084a1ea 100755
--- a/tools/filters/wiggle_to_simple.py
+++ b/tools/filters/wiggle_to_simple.py
@@ -9,7 +9,7 @@ import sys
 
 import bx.wiggle
 
-from galaxy.tools.exception_handling import UCSCLimitException, UCSCOutWrapper
+from galaxy.util.ucsc import UCSCLimitException, UCSCOutWrapper
 
 
 def stop_err( msg ):
diff --git a/tools/genomespace/genomespace_file_browser.py b/tools/genomespace/genomespace_file_browser.py
index 32eeb1718c1..8e9e7370131 100644
--- a/tools/genomespace/genomespace_file_browser.py
+++ b/tools/genomespace/genomespace_file_browser.py
@@ -9,7 +9,7 @@ import urllib2
 import urlparse
 
 
-import galaxy.model # need to import model before sniff to resolve a circular import dependency
+import galaxy.model  # import no longer needed, remove next time this file is modified
 from galaxy.datatypes import sniff
 from galaxy.datatypes.registry import Registry
 
diff --git a/tools/genomespace/genomespace_importer.py b/tools/genomespace/genomespace_importer.py
index 3daf0f51c06..e1613d36593 100644
--- a/tools/genomespace/genomespace_importer.py
+++ b/tools/genomespace/genomespace_importer.py
@@ -10,7 +10,7 @@ import urllib
 import urllib2
 import urlparse
 
-import galaxy.model # need to import model before sniff to resolve a circular import dependency
+import galaxy.model # import no longer needed, remove next time this file is modified
 from galaxy.datatypes import sniff
 from galaxy.datatypes.registry import Registry
 
diff --git a/tools/stats/aggregate_scores_in_intervals.py b/tools/stats/aggregate_scores_in_intervals.py
index f5a8ec796c6..f527f37e339 100755
--- a/tools/stats/aggregate_scores_in_intervals.py
+++ b/tools/stats/aggregate_scores_in_intervals.py
@@ -22,7 +22,7 @@ from bx.binned_array import BinnedArray, FileBinnedArray
 from bx.bitset_builders import binned_bitsets_from_file
 from bx.cookbook import doc_optparse
 
-from galaxy.tools.exception_handling import UCSCLimitException, UCSCOutWrapper
+from galaxy.util.ucsc import UCSCLimitException, UCSCOutWrapper
 
 
 class PositionalScoresOnDisk: