"""
Classes related to parameter validation.
"""
import os, re, logging
from elementtree.ElementTree import XML
from galaxy import model
log = logging.getLogger( __name__ )
class LateValidationError( Exception ):
def __init__( self, message ):
self.message = message
class Validator( object ):
"""
A validator checks that a value meets some conditions OR raises ValueError
"""
@classmethod
def from_element( cls, param, elem ):
type = elem.get( 'type', None )
assert type is not None, "Required 'type' attribute missing from validator"
return validator_types[type].from_element( param, elem )
def validate( self, value, history=None ):
raise TypeError( "Abstract Method" )
class RegexValidator( Validator ):
"""
Validator that evaluates a regular expression
>>> from galaxy.tools.parameters import ToolParameter
>>> p = ToolParameter.build( None, XML( '''
...
... [Ff]oo
...
... ''' ) )
>>> t = p.validate( "Foo" )
>>> t = p.validate( "foo" )
>>> t = p.validate( "Fop" )
Traceback (most recent call last):
...
ValueError: Not gonna happen
"""
@classmethod
def from_element( cls, param, elem ):
return cls( elem.get( 'message' ), elem.text )
def __init__( self, message, expression ):
self.message = message
# Compile later. RE objects used to not be thread safe. Not sure about
# the sre module.
self.expression = expression
def validate( self, value, history=None ):
if re.match( self.expression, value ) is None:
raise ValueError( self.message )
class ExpressionValidator( Validator ):
"""
Validator that evaluates a python expression using the value
>>> from galaxy.tools.parameters import ToolParameter
>>> p = ToolParameter.build( None, XML( '''
...
... value.lower() == "foo"
...
... ''' ) )
>>> t = p.validate( "Foo" )
>>> t = p.validate( "foo" )
>>> t = p.validate( "Fop" )
Traceback (most recent call last):
...
ValueError: Not gonna happen
"""
@classmethod
def from_element( cls, param, elem ):
return cls( elem.get( 'message' ), elem.text, elem.get( 'substitute_value_in_message' ) )
def __init__( self, message, expression, substitute_value_in_message ):
self.message = message
self.substitute_value_in_message = substitute_value_in_message
# Save compiled expression, code objects are thread safe (right?)
self.expression = compile( expression, '', 'eval' )
def validate( self, value, history=None ):
if not( eval( self.expression, dict( value=value ) ) ):
message = self.message
if self.substitute_value_in_message:
message = message % value
raise ValueError( message )
class InRangeValidator( Validator ):
"""
Validator that ensures a number is in a specific range
>>> from galaxy.tools.parameters import ToolParameter
>>> p = ToolParameter.build( None, XML( '''
...
...
...
... ''' ) )
>>> t = p.validate( 10 )
>>> t = p.validate( 15 )
>>> t = p.validate( 20 )
>>> t = p.validate( 21 )
Traceback (most recent call last):
...
ValueError: Not gonna happen
"""
@classmethod
def from_element( cls, param, elem ):
return cls( elem.get( 'message', None ), elem.get( 'min', '-inf' ), elem.get( 'max', '+inf' ) )
def __init__( self, message, range_min, range_max ):
self.min = float( range_min )
self.max = float( range_max )
self.message = message or ( "Value must be between %f and %f" % ( self.min, self.max ) )
def validate( self, value, history=None ):
if not( self.min <= float( value ) <= self.max ):
raise ValueError( self.message )
class LengthValidator( Validator ):
"""
Validator that ensures the length of the provided string (value) is in a specific range
>>> from galaxy.tools.parameters import ToolParameter
>>> p = ToolParameter.build( None, XML( '''
...
...
...
... ''' ) )
>>> t = p.validate( "foo" )
>>> t = p.validate( "bar" )
>>> t = p.validate( "f" )
Traceback (most recent call last):
...
ValueError: Must have length of at least 2
>>> t = p.validate( "foobarbaz" )
Traceback (most recent call last):
...
ValueError: Must have length no more than 8
"""
@classmethod
def from_element( cls, param, elem ):
return cls( elem.get( 'message', None ), elem.get( 'min', None ), elem.get( 'max', None ) )
def __init__( self, message, length_min, length_max ):
self.message = message
if length_min is not None:
length_min = int( length_min )
if length_max is not None:
length_max = int( length_max )
self.min = length_min
self.max = length_max
def validate( self, value, history=None ):
if self.min is not None and len( value ) < self.min:
raise ValueError( self.message or ( "Must have length of at least %d" % self.min ) )
if self.max is not None and len( value ) > self.max:
raise ValueError( self.message or ( "Must have length no more than %d" % self.max ) )
class DatasetOkValidator( Validator ):
"""
Validator that checks if a dataset is in an 'ok' state
"""
def __init__( self, message=None ):
self.message = message
@classmethod
def from_element( cls, param, elem ):
return cls( elem.get( 'message', None ) )
def validate( self, value, history=None ):
if value and value.state != model.Dataset.states.OK:
if self.message is None:
self.message = "The selected dataset is still being generated, select another dataset or wait until it is completed"
raise ValueError( self.message )
class MetadataValidator( Validator ):
"""
Validator that checks for missing metadata
"""
def __init__( self, message = None, check = "", skip = "" ):
self.message = message
self.check = check.split( "," )
self.skip = skip.split( "," )
@classmethod
def from_element( cls, param, elem ):
return cls( message=elem.get( 'message', None ), check=elem.get( 'check', "" ), skip=elem.get( 'skip', "" ) )
def validate( self, value, history=None ):
if value and value.missing_meta( check = self.check, skip = self.skip ):
if self.message is None:
self.message = "Metadata missing, click the pencil icon in the history item to edit / save the metadata attributes"
raise ValueError( self.message )
class UnspecifiedBuildValidator( Validator ):
"""
Validator that checks for missing metadata
"""
def __init__( self, message=None ):
if message is None:
self.message = "Unspecified genome build, click the pencil icon in the history item to set the genome build"
else:
self.message = message
@classmethod
def from_element( cls, param, elem ):
return cls( elem.get( 'message', None ) )
def validate( self, value, history=None ):
#if value is None, we cannot validate
if value:
dbkey = value.metadata.dbkey
if isinstance( dbkey, list ):
dbkey = dbkey[0]
if dbkey == '?':
raise ValueError( self.message )
class NoOptionsValidator( Validator ):
"""Validator that checks for empty select list"""
def __init__( self, message=None ):
self.message = message
@classmethod
def from_element( cls, param, elem ):
return cls( elem.get( 'message', None ) )
def validate( self, value, history=None ):
if value is None:
if self.message is None:
self.message = "No options available for selection"
raise ValueError( self.message )
class EmptyTextfieldValidator( Validator ):
"""Validator that checks for empty text field"""
def __init__( self, message=None ):
self.message = message
@classmethod
def from_element( cls, param, elem ):
return cls( elem.get( 'message', None ) )
def validate( self, value, history=None ):
if value == '':
if self.message is None:
self.message = "Field requires a value"
raise ValueError( self.message )
class MetadataInFileColumnValidator( Validator ):
"""
Validator that checks if the value for a dataset's metadata item exists in a file.
"""
@classmethod
def from_element( cls, param, elem ):
filename = elem.get( "filename", None )
if filename:
filename = "%s/%s" % ( param.tool.app.config.tool_data_path, filename.strip() )
metadata_name = elem.get( "metadata_name", None )
if metadata_name:
metadata_name = metadata_name.strip()
metadata_column = int( elem.get( "metadata_column", 0 ) )
message = elem.get( "message", "Value for metadata %s was not found in %s." % ( metadata_name, filename ) )
line_startswith = elem.get( "line_startswith", None )
if line_startswith:
line_startswith = line_startswith.strip()
return cls( filename, metadata_name, metadata_column, message, line_startswith )
def __init__( self, filename, metadata_name, metadata_column, message="Value for metadata not found.", line_startswith=None ):
self.metadata_name = metadata_name
self.message = message
self.valid_values = []
for line in open( filename ):
if line_startswith is None or line.startswith( line_startswith ):
fields = line.split( '\t' )
if metadata_column < len( fields ):
self.valid_values.append( fields[metadata_column].strip() )
def validate( self, value, history = None ):
if not value: return
if hasattr( value, "metadata" ):
if value.metadata.spec[self.metadata_name].param.to_string( value.metadata.get( self.metadata_name ) ) in self.valid_values:
return
raise ValueError( self.message )
validator_types = dict( expression=ExpressionValidator,
regex=RegexValidator,
in_range=InRangeValidator,
length=LengthValidator,
metadata=MetadataValidator,
unspecified_build=UnspecifiedBuildValidator,
no_options=NoOptionsValidator,
empty_field=EmptyTextfieldValidator,
dataset_metadata_in_file=MetadataInFileColumnValidator,
dataset_ok_validator=DatasetOkValidator )
def get_suite():
"""Get unittest suite for this module"""
import doctest, sys
return doctest.DocTestSuite( sys.modules[__name__] )