Source code for galaxy.datatypes.dataproviders.decorators

DataProvider related decorators.

# I'd like to decorate the factory methods that give data_providers by the name they can be accessed from. e.g.:
# @provides( 'id_seq' ) # where 'id_seq' is some 'data_format' string/alias
# def get_id_seq_provider( dataset, **settings ):

# then in some central dispatch (maybe data.Data), have it look up the proper method by the data_format string

# also it would be good to have this decorator maintain a list of available providers (for a datatype)

# i don't particularly want to cut up method names ( get_([\w_]*)_provider )
# adapted from:
#    /questions/14095616/python-can-i-programmatically-decorate-class-methods-from-a-class-instance

import copy
import logging
from functools import wraps
from urllib.parse import unquote

log = logging.getLogger(__name__)

_DATAPROVIDER_METHOD_NAME_KEY = "_dataprovider_name"

[docs]def has_dataproviders(cls): """ Wraps a class (generally a Datatype), finds methods within that have been decorated with `@dataprovider` and adds them, by their name, to a map in the class. This allows a class to maintain a name -> method map, effectively 'registering' dataprovider factory methods:: @has_dataproviders class MyDtype( data.Data ): @dataprovider_factory( 'bler' ) def provide_some_bler( self, dataset, **settings ): '''blerblerbler''' dataset_source = providers.DatasetDataProvider( dataset ) # ... chain other, intermidiate providers here return providers.BlerDataProvider( dataset_source, **settings ) # use the base method in data.Data provider = dataset.datatype.dataprovider( dataset, 'bler', my_setting='blah', ... ) # OR directly from the map provider = dataset.datatype.dataproviders[ 'bler' ]( dataset, my_setting='blah', ... ) """ # init the class dataproviders map if necc. if not hasattr(cls, _DATAPROVIDER_CLASS_MAP_KEY): setattr(cls, _DATAPROVIDER_CLASS_MAP_KEY, {}) else: # need to deepcopy or subclasses will modify super.dataproviders as well existing_dataproviders = getattr(cls, _DATAPROVIDER_CLASS_MAP_KEY) copied_dataproviders = copy.deepcopy(existing_dataproviders) setattr(cls, _DATAPROVIDER_CLASS_MAP_KEY, copied_dataproviders) dataproviders = getattr(cls, _DATAPROVIDER_CLASS_MAP_KEY) # scan for methods with dataprovider names and add them to the map # note: this has a 'cascading' effect # where it's possible to override a super's provider with a sub's for attr_key, attr_value in cls.__dict__.items(): # can't use isinstance( attr_value, MethodType ) bc of wrapping if ( (callable(attr_value)) and (not attr_key.startswith("__")) and (getattr(attr_value, _DATAPROVIDER_METHOD_NAME_KEY, None)) ): name = getattr(attr_value, _DATAPROVIDER_METHOD_NAME_KEY) dataproviders[name] = attr_value return cls
[docs]def dataprovider_factory(name, settings=None): """ Wraps a class method and marks it as a dataprovider factory and creates a function to parse query strings to __init__ arguments as the `parse_query_string_settings` attribute of the factory function. An example use of the `parse_query_string_settings`:: kwargs = dataset.datatype.dataproviders[ provider ].parse_query_string_settings( query_kwargs ) return list( dataset.datatype.dataprovider( dataset, provider, **kwargs ) ) :param name: what name/key to register the factory under in `cls.dataproviders` :type name: any hashable var :param settings: dictionary containing key/type pairs for parsing query strings to __init__ arguments :type settings: dictionary """ # TODO:?? use *args for settings allowing mulitple dictionaries # make a function available through the name->provider dispatch to parse query strings # callable like: # settings_dict = dataproviders[ provider_name ].parse_query_string_settings( query_kwargs ) # TODO: ugh - overly complicated but the best I could think of def parse_query_string_settings(query_kwargs): return _parse_query_string_settings(query_kwargs, settings) def named_dataprovider_factory(func): setattr(func, _DATAPROVIDER_METHOD_NAME_KEY, name) func.parse_query_string_settings = parse_query_string_settings func.settings = settings # TODO: I want a way to inherit settings from the previous provider( this_name ) instead of defining over and over @wraps(func) def wrapped_dataprovider_factory(self, *args, **kwargs): return func(self, *args, **kwargs) return wrapped_dataprovider_factory return named_dataprovider_factory
def _parse_query_string_settings(query_kwargs, settings=None): """ Parse the values in `query_kwargs` from strings to the proper types listed in the same key in `settings`. """ # TODO: this was a relatively late addition: review and re-think def list_from_query_string(s): # assume csv return s.split(",") parsers = { "int": int, "float": float, "bool": bool, "list:str": lambda s: list_from_query_string(s), "list:escaped": lambda s: [unquote(e) for e in list_from_query_string(s)], "list:int": lambda s: [int(i) for i in list_from_query_string(s)], } settings = settings or {} # yay! yet another set of query string parsers! <-- sarcasm # work through the keys in settings finding matching keys in query_kwargs # if found in both, get the expected/needed type from settings and store the new parsed value # if we can't parse it (no parser, bad value), delete the key from query_kwargs so the provider will use the defaults for key in settings: if key in query_kwargs: # TODO: this would be the place to sanitize any strings query_value = query_kwargs[key] needed_type = settings[key] if needed_type != "str": try: query_kwargs[key] = parsers[needed_type](query_value) except (KeyError, ValueError): del query_kwargs[key] # TODO:?? do we want to remove query_kwarg entries NOT in settings? return query_kwargs