<s> import xml . sax <EOL> import datetime <EOL> import itertools <EOL> from boto import handler <EOL> from boto import config <EOL> from boto . mturk . price import Price <EOL> import boto . mturk . notification <EOL> from boto . connection import AWSQueryConnection <EOL> from boto . exception import EC2ResponseError <EOL> from boto . resultset import ResultSet <EOL> from boto . mturk . question import QuestionForm , ExternalQuestion <EOL> class MTurkRequestError ( EC2ResponseError ) : <EOL> "<STR_LIT>" <EOL> class MTurkConnection ( AWSQueryConnection ) : <EOL> APIVersion = '<STR_LIT>' <EOL> def __init__ ( self , aws_access_key_id = None , aws_secret_access_key = None , <EOL> is_secure = True , port = None , proxy = None , proxy_port = None , <EOL> proxy_user = None , proxy_pass = None , <EOL> host = None , debug = <NUM_LIT:0> , <EOL> https_connection_factory = None ) : <EOL> if not host : <EOL> if config . has_option ( '<STR_LIT>' , '<STR_LIT>' ) and config . get ( '<STR_LIT>' , '<STR_LIT>' ) == '<STR_LIT:True>' : <EOL> host = '<STR_LIT>' <EOL> else : <EOL> host = '<STR_LIT>' <EOL> AWSQueryConnection . __init__ ( self , aws_access_key_id , <EOL> aws_secret_access_key , <EOL> is_secure , port , proxy , proxy_port , <EOL> proxy_user , proxy_pass , host , debug , <EOL> https_connection_factory ) <EOL> def _required_auth_capability ( self ) : <EOL> return [ '<STR_LIT>' ] <EOL> def get_account_balance ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> params = { } <EOL> return self . _process_request ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , Price ) , <EOL> ( '<STR_LIT>' , Price ) ] ) <EOL> def register_hit_type ( self , title , description , reward , duration , <EOL> keywords = None , approval_delay = None , qual_req = None ) : <EOL> """<STR_LIT>""" <EOL> params = dict ( <EOL> Title = title , <EOL> Description = description , <EOL> AssignmentDurationInSeconds = <EOL> self . duration_as_seconds ( duration ) , <EOL> ) <EOL> params . update ( MTurkConnection . get_price_as_price ( reward ) . get_as_params ( '<STR_LIT>' ) ) <EOL> if keywords : <EOL> params [ '<STR_LIT>' ] = self . get_keywords_as_string ( keywords ) <EOL> if approval_delay is not None : <EOL> d = self . duration_as_seconds ( approval_delay ) <EOL> params [ '<STR_LIT>' ] = d <EOL> if qual_req is not None : <EOL> params . update ( qual_req . get_as_params ( ) ) <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def set_email_notification ( self , hit_type , email , event_types = None ) : <EOL> """<STR_LIT>""" <EOL> return self . _set_notification ( hit_type , '<STR_LIT>' , email , event_types ) <EOL> def set_rest_notification ( self , hit_type , url , event_types = None ) : <EOL> """<STR_LIT>""" <EOL> return self . _set_notification ( hit_type , '<STR_LIT>' , url , event_types ) <EOL> def _set_notification ( self , hit_type , transport , destination , event_types = None ) : <EOL> """<STR_LIT>""" <EOL> assert type ( hit_type ) is str , "<STR_LIT>" <EOL> params = { '<STR_LIT>' : hit_type } <EOL> notification_params = { '<STR_LIT>' : destination , <EOL> '<STR_LIT>' : transport , <EOL> '<STR_LIT>' : boto . mturk . notification . NotificationMessage . NOTIFICATION_VERSION , <EOL> '<STR_LIT>' : True , <EOL> } <EOL> if event_types : <EOL> self . build_list_params ( notification_params , event_types , '<STR_LIT>' ) <EOL> notification_rest_params = { } <EOL> num = <NUM_LIT:1> <EOL> for key in notification_params : <EOL> notification_rest_params [ '<STR_LIT>' % ( num , key ) ] = notification_params [ key ] <EOL> params . update ( notification_rest_params ) <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def create_hit ( self , hit_type = None , question = None , <EOL> lifetime = datetime . timedelta ( days = <NUM_LIT:7> ) , <EOL> max_assignments = <NUM_LIT:1> , <EOL> title = None , description = None , keywords = None , <EOL> reward = None , duration = datetime . timedelta ( days = <NUM_LIT:7> ) , <EOL> approval_delay = None , annotation = None , <EOL> questions = None , qualifications = None , <EOL> response_groups = None ) : <EOL> """<STR_LIT>""" <EOL> neither = question is None and questions is None <EOL> both = question is not None and questions is not None <EOL> if neither or both : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if question : <EOL> questions = [ question ] <EOL> question_param = QuestionForm ( questions ) <EOL> if isinstance ( question , QuestionForm ) : <EOL> question_param = question <EOL> elif isinstance ( question , ExternalQuestion ) : <EOL> question_param = question <EOL> params = { '<STR_LIT>' : question_param . get_as_xml ( ) , <EOL> '<STR_LIT>' : <EOL> self . duration_as_seconds ( lifetime ) , <EOL> '<STR_LIT>' : max_assignments , <EOL> } <EOL> if hit_type : <EOL> params [ '<STR_LIT>' ] = hit_type <EOL> else : <EOL> final_keywords = MTurkConnection . get_keywords_as_string ( keywords ) <EOL> final_price = MTurkConnection . get_price_as_price ( reward ) <EOL> final_duration = self . duration_as_seconds ( duration ) <EOL> additional_params = dict ( <EOL> Title = title , <EOL> Description = description , <EOL> Keywords = final_keywords , <EOL> AssignmentDurationInSeconds = final_duration , <EOL> ) <EOL> additional_params . update ( final_price . get_as_params ( '<STR_LIT>' ) ) <EOL> if approval_delay is not None : <EOL> d = self . duration_as_seconds ( approval_delay ) <EOL> additional_params [ '<STR_LIT>' ] = d <EOL> params . update ( additional_params ) <EOL> if annotation is not None : <EOL> params [ '<STR_LIT>' ] = annotation <EOL> if qualifications is not None : <EOL> params . update ( qualifications . get_as_params ( ) ) <EOL> if response_groups : <EOL> self . build_list_params ( params , response_groups , '<STR_LIT>' ) <EOL> return self . _process_request ( '<STR_LIT>' , params , [ ( '<STR_LIT>' , HIT ) , ] ) <EOL> def change_hit_type_of_hit ( self , hit_id , hit_type ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : hit_id , <EOL> '<STR_LIT>' : hit_type } <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def get_reviewable_hits ( self , hit_type = None , status = '<STR_LIT>' , <EOL> sort_by = '<STR_LIT>' , sort_direction = '<STR_LIT>' , <EOL> page_size = <NUM_LIT:10> , page_number = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : status , <EOL> '<STR_LIT>' : sort_by , <EOL> '<STR_LIT>' : sort_direction , <EOL> '<STR_LIT>' : page_size , <EOL> '<STR_LIT>' : page_number } <EOL> if hit_type is not None : <EOL> params . update ( { '<STR_LIT>' : hit_type } ) <EOL> return self . _process_request ( '<STR_LIT>' , params , [ ( '<STR_LIT>' , HIT ) , ] ) <EOL> @ staticmethod <EOL> def _get_pages ( page_size , total_records ) : <EOL> """<STR_LIT>""" <EOL> pages = total_records / page_size + bool ( total_records % page_size ) <EOL> return range ( <NUM_LIT:1> , pages + <NUM_LIT:1> ) <EOL> def get_all_hits ( self ) : <EOL> """<STR_LIT>""" <EOL> page_size = <NUM_LIT:100> <EOL> search_rs = self . search_hits ( page_size = page_size ) <EOL> total_records = int ( search_rs . TotalNumResults ) <EOL> get_page_hits = lambda ( page ) : self . search_hits ( page_size = page_size , page_number = page ) <EOL> page_nums = self . _get_pages ( page_size , total_records ) <EOL> hit_sets = itertools . imap ( get_page_hits , page_nums ) <EOL> return itertools . chain . from_iterable ( hit_sets ) <EOL> def search_hits ( self , sort_by = '<STR_LIT>' , sort_direction = '<STR_LIT>' , <EOL> page_size = <NUM_LIT:10> , page_number = <NUM_LIT:1> , response_groups = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : sort_by , <EOL> '<STR_LIT>' : sort_direction , <EOL> '<STR_LIT>' : page_size , <EOL> '<STR_LIT>' : page_number } <EOL> if response_groups : <EOL> self . build_list_params ( params , response_groups , '<STR_LIT>' ) <EOL> return self . _process_request ( '<STR_LIT>' , params , [ ( '<STR_LIT>' , HIT ) , ] ) <EOL> def get_assignments ( self , hit_id , status = None , <EOL> sort_by = '<STR_LIT>' , sort_direction = '<STR_LIT>' , <EOL> page_size = <NUM_LIT:10> , page_number = <NUM_LIT:1> , response_groups = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : hit_id , <EOL> '<STR_LIT>' : sort_by , <EOL> '<STR_LIT>' : sort_direction , <EOL> '<STR_LIT>' : page_size , <EOL> '<STR_LIT>' : page_number } <EOL> if status is not None : <EOL> params [ '<STR_LIT>' ] = status <EOL> if response_groups : <EOL> self . build_list_params ( params , response_groups , '<STR_LIT>' ) <EOL> return self . _process_request ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , Assignment ) , ] ) <EOL> def approve_assignment ( self , assignment_id , feedback = None ) : <EOL> """<STR_LIT:U+0020>""" <EOL> params = { '<STR_LIT>' : assignment_id , } <EOL> if feedback : <EOL> params [ '<STR_LIT>' ] = feedback <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def reject_assignment ( self , assignment_id , feedback = None ) : <EOL> """<STR_LIT:U+0020>""" <EOL> params = { '<STR_LIT>' : assignment_id , } <EOL> if feedback : <EOL> params [ '<STR_LIT>' ] = feedback <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def get_hit ( self , hit_id , response_groups = None ) : <EOL> """<STR_LIT:U+0020>""" <EOL> params = { '<STR_LIT>' : hit_id , } <EOL> if response_groups : <EOL> self . build_list_params ( params , response_groups , '<STR_LIT>' ) <EOL> return self . _process_request ( '<STR_LIT>' , params , [ ( '<STR_LIT>' , HIT ) , ] ) <EOL> def set_reviewing ( self , hit_id , revert = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : hit_id , } <EOL> if revert : <EOL> params [ '<STR_LIT>' ] = revert <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def disable_hit ( self , hit_id , response_groups = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : hit_id , } <EOL> if response_groups : <EOL> self . build_list_params ( params , response_groups , '<STR_LIT>' ) <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def dispose_hit ( self , hit_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : hit_id , } <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def expire_hit ( self , hit_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : hit_id , } <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def extend_hit ( self , hit_id , assignments_increment = None , expiration_increment = None ) : <EOL> """<STR_LIT>""" <EOL> if ( assignments_increment is None and expiration_increment is None ) or ( assignments_increment is not None and expiration_increment is not None ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> params = { '<STR_LIT>' : hit_id , } <EOL> if assignments_increment : <EOL> params [ '<STR_LIT>' ] = assignments_increment <EOL> if expiration_increment : <EOL> params [ '<STR_LIT>' ] = expiration_increment <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def get_help ( self , about , help_type = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : about , '<STR_LIT>' : help_type , } <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def grant_bonus ( self , worker_id , assignment_id , bonus_price , reason ) : <EOL> """<STR_LIT>""" <EOL> params = bonus_price . get_as_params ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> params [ '<STR_LIT>' ] = worker_id <EOL> params [ '<STR_LIT>' ] = assignment_id <EOL> params [ '<STR_LIT>' ] = reason <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def block_worker ( self , worker_id , reason ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : worker_id , '<STR_LIT>' : reason } <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def unblock_worker ( self , worker_id , reason ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : worker_id , '<STR_LIT>' : reason } <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def notify_workers ( self , worker_ids , subject , message_text ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : subject , <EOL> '<STR_LIT>' : message_text } <EOL> self . build_list_params ( params , worker_ids , '<STR_LIT>' ) <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def create_qualification_type ( self , <EOL> name , <EOL> description , <EOL> status , <EOL> keywords = None , <EOL> retry_delay = None , <EOL> test = None , <EOL> answer_key = None , <EOL> answer_key_xml = None , <EOL> test_duration = None , <EOL> auto_granted = False , <EOL> auto_granted_value = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:Name>' : name , <EOL> '<STR_LIT>' : description , <EOL> '<STR_LIT>' : status , <EOL> } <EOL> if retry_delay is not None : <EOL> params [ '<STR_LIT>' ] = retry_delay <EOL> if test is not None : <EOL> assert ( isinstance ( test , QuestionForm ) ) <EOL> assert ( test_duration is not None ) <EOL> params [ '<STR_LIT>' ] = test . get_as_xml ( ) <EOL> if test_duration is not None : <EOL> params [ '<STR_LIT>' ] = test_duration <EOL> if answer_key is not None : <EOL> if isinstance ( answer_key , basestring ) : <EOL> params [ '<STR_LIT>' ] = answer_key <EOL> else : <EOL> raise TypeError <EOL> if auto_granted : <EOL> assert ( test is False ) <EOL> params [ '<STR_LIT>' ] = True <EOL> params [ '<STR_LIT>' ] = auto_granted_value <EOL> if keywords : <EOL> params [ '<STR_LIT>' ] = self . get_keywords_as_string ( keywords ) <EOL> return self . _process_request ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , QualificationType ) , ] ) <EOL> def get_qualification_type ( self , qualification_type_id ) : <EOL> params = { '<STR_LIT>' : qualification_type_id } <EOL> return self . _process_request ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , QualificationType ) , ] ) <EOL> def get_qualifications_for_qualification_type ( self , qualification_type_id ) : <EOL> params = { '<STR_LIT>' : qualification_type_id } <EOL> return self . _process_request ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , QualificationType ) , ] ) <EOL> def update_qualification_type ( self , qualification_type_id , <EOL> description = None , <EOL> status = None , <EOL> retry_delay = None , <EOL> test = None , <EOL> answer_key = None , <EOL> test_duration = None , <EOL> auto_granted = None , <EOL> auto_granted_value = None ) : <EOL> params = { '<STR_LIT>' : qualification_type_id } <EOL> if description is not None : <EOL> params [ '<STR_LIT>' ] = description <EOL> if status is not None : <EOL> params [ '<STR_LIT>' ] = status <EOL> if retry_delay is not None : <EOL> params [ '<STR_LIT>' ] = retry_delay <EOL> if test is not None : <EOL> assert ( isinstance ( test , QuestionForm ) ) <EOL> params [ '<STR_LIT>' ] = test . get_as_xml ( ) <EOL> if test_duration is not None : <EOL> params [ '<STR_LIT>' ] = test_duration <EOL> if answer_key is not None : <EOL> if isinstance ( answer_key , basestring ) : <EOL> params [ '<STR_LIT>' ] = answer_key <EOL> else : <EOL> raise TypeError <EOL> if auto_granted is not None : <EOL> params [ '<STR_LIT>' ] = auto_granted <EOL> if auto_granted_value is not None : <EOL> params [ '<STR_LIT>' ] = auto_granted_value <EOL> return self . _process_request ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , QualificationType ) , ] ) <EOL> def dispose_qualification_type ( self , qualification_type_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : qualification_type_id } <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def search_qualification_types ( self , query = None , sort_by = '<STR_LIT:Name>' , <EOL> sort_direction = '<STR_LIT>' , page_size = <NUM_LIT:10> , <EOL> page_number = <NUM_LIT:1> , must_be_requestable = True , <EOL> must_be_owned_by_caller = True ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : query , <EOL> '<STR_LIT>' : sort_by , <EOL> '<STR_LIT>' : sort_direction , <EOL> '<STR_LIT>' : page_size , <EOL> '<STR_LIT>' : page_number , <EOL> '<STR_LIT>' : must_be_requestable , <EOL> '<STR_LIT>' : must_be_owned_by_caller } <EOL> return self . _process_request ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , QualificationType ) , ] ) <EOL> def get_qualification_requests ( self , qualification_type_id , <EOL> sort_by = '<STR_LIT>' , <EOL> sort_direction = '<STR_LIT>' , page_size = <NUM_LIT:10> , <EOL> page_number = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : qualification_type_id , <EOL> '<STR_LIT>' : sort_by , <EOL> '<STR_LIT>' : sort_direction , <EOL> '<STR_LIT>' : page_size , <EOL> '<STR_LIT>' : page_number } <EOL> return self . _process_request ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , QualificationRequest ) , ] ) <EOL> def grant_qualification ( self , qualification_request_id , integer_value = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : qualification_request_id , <EOL> '<STR_LIT>' : integer_value } <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def revoke_qualification ( self , subject_id , qualification_type_id , <EOL> reason = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : subject_id , <EOL> '<STR_LIT>' : qualification_type_id , <EOL> '<STR_LIT>' : reason } <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def assign_qualification ( self , qualification_type_id , worker_id , <EOL> value = <NUM_LIT:1> , send_notification = True ) : <EOL> params = { '<STR_LIT>' : qualification_type_id , <EOL> '<STR_LIT>' : worker_id , <EOL> '<STR_LIT>' : value , <EOL> '<STR_LIT>' : send_notification } <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def get_qualification_score ( self , qualification_type_id , worker_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : qualification_type_id , <EOL> '<STR_LIT>' : worker_id } <EOL> return self . _process_request ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , Qualification ) , ] ) <EOL> def update_qualification_score ( self , qualification_type_id , worker_id , <EOL> value ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : qualification_type_id , <EOL> '<STR_LIT>' : worker_id , <EOL> '<STR_LIT>' : value } <EOL> return self . _process_request ( '<STR_LIT>' , params ) <EOL> def _process_request ( self , request_type , params , marker_elems = None ) : <EOL> """<STR_LIT>""" <EOL> response = self . make_request ( request_type , params , verb = '<STR_LIT:POST>' ) <EOL> return self . _process_response ( response , marker_elems ) <EOL> def _process_response ( self , response , marker_elems = None ) : <EOL> """<STR_LIT>""" <EOL> body = response . read ( ) <EOL> if '<STR_LIT>' not in body : <EOL> rs = ResultSet ( marker_elems ) <EOL> h = handler . XmlHandler ( rs , self ) <EOL> xml . sax . parseString ( body , h ) <EOL> return rs <EOL> else : <EOL> raise MTurkRequestError ( response . status , response . reason , body ) <EOL> @ staticmethod <EOL> def get_keywords_as_string ( keywords ) : <EOL> """<STR_LIT>""" <EOL> if type ( keywords ) is list : <EOL> keywords = '<STR_LIT:U+002CU+0020>' . join ( keywords ) <EOL> if type ( keywords ) is str : <EOL> final_keywords = keywords <EOL> elif type ( keywords ) is unicode : <EOL> final_keywords = keywords . encode ( '<STR_LIT:utf-8>' ) <EOL> elif keywords is None : <EOL> final_keywords = "<STR_LIT>" <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" % type ( keywords ) ) <EOL> return final_keywords <EOL> @ staticmethod <EOL> def get_price_as_price ( reward ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( reward , Price ) : <EOL> final_price = reward <EOL> else : <EOL> final_price = Price ( reward ) <EOL> return final_price <EOL> @ staticmethod <EOL> def duration_as_seconds ( duration ) : <EOL> if isinstance ( duration , datetime . timedelta ) : <EOL> duration = duration . days * <NUM_LIT> + duration . seconds <EOL> try : <EOL> duration = int ( duration ) <EOL> except TypeError : <EOL> raise TypeError ( "<STR_LIT>" % type ( duration ) ) <EOL> return duration <EOL> class BaseAutoResultElement : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , connection ) : <EOL> pass <EOL> def startElement ( self , name , attrs , connection ) : <EOL> return None <EOL> def endElement ( self , name , value , connection ) : <EOL> setattr ( self , name , value ) <EOL> class HIT ( BaseAutoResultElement ) : <EOL> """<STR_LIT>""" <EOL> def _has_expired ( self ) : <EOL> """<STR_LIT>""" <EOL> expired = False <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> now = datetime . datetime . utcnow ( ) <EOL> expiration = datetime . datetime . strptime ( self . Expiration , '<STR_LIT>' ) <EOL> expired = ( now >= expiration ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return expired <EOL> expired = property ( _has_expired ) <EOL> class Qualification ( BaseAutoResultElement ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class QualificationType ( BaseAutoResultElement ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class QualificationRequest ( BaseAutoResultElement ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class Assignment ( BaseAutoResultElement ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , connection ) : <EOL> BaseAutoResultElement . __init__ ( self , connection ) <EOL> self . answers = [ ] <EOL> def endElement ( self , name , value , connection ) : <EOL> if name == '<STR_LIT>' : <EOL> answer_rs = ResultSet ( [ ( '<STR_LIT>' , QuestionFormAnswer ) , ] ) <EOL> h = handler . XmlHandler ( answer_rs , connection ) <EOL> value = connection . get_utf8_value ( value ) <EOL> xml . sax . parseString ( value , h ) <EOL> self . answers . append ( answer_rs ) <EOL> else : <EOL> BaseAutoResultElement . endElement ( self , name , value , connection ) <EOL> class QuestionFormAnswer ( BaseAutoResultElement ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , connection ) : <EOL> BaseAutoResultElement . __init__ ( self , connection ) <EOL> self . fields = [ ] <EOL> self . qid = None <EOL> def endElement ( self , name , value , connection ) : <EOL> if name == '<STR_LIT>' : <EOL> self . qid = value <EOL> elif name in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] and self . qid : <EOL> self . fields . append ( ( self . qid , value ) ) <EOL> elif name == '<STR_LIT>' : <EOL> self . qid = None </s>
