signature:
def courses ( self , request , enterprise_customer , pk = None ) : # pylint : disable = invalid - name
    """Retrieve the list of courses contained within this catalog .
 Only courses with active course runs are returned . A course run is considered active if it is currently
 open for enrollment , or will open in the future ."""

implementation:
catalog_api = CourseCatalogApiClient ( request . user , enterprise_customer . site )
courses = catalog_api . get_paginated_catalog_courses ( pk , request . GET )
# If the API returned an empty response , that means pagination has ended .
# An empty response can also mean that there was a problem fetching data from catalog API .
self . ensure_data_exists ( request , courses , error_message = ( "Unable to fetch API response for catalog courses from endpoint '{endpoint}'. " "The resource you are looking for does not exist." . format ( endpoint = request . get_full_path ( ) ) ) )
serializer = serializers . EnterpriseCatalogCoursesReadOnlySerializer ( courses )
# Add enterprise related context for the courses .
serializer . update_enterprise_courses ( enterprise_customer , catalog_id = pk )
return get_paginated_response ( serializer . data , request )


signature:
def get_required_query_params ( self , request ) :
    """Gets ` ` email ` ` , ` ` enterprise _ name ` ` , and ` ` number _ of _ codes ` ` ,
 which are the relevant parameters for this API endpoint .
 : param request : The request to this endpoint .
 : return : The ` ` email ` ` , ` ` enterprise _ name ` ` , and ` ` number _ of _ codes ` ` from the request ."""

implementation:
email = get_request_value ( request , self . REQUIRED_PARAM_EMAIL , '' )
enterprise_name = get_request_value ( request , self . REQUIRED_PARAM_ENTERPRISE_NAME , '' )
number_of_codes = get_request_value ( request , self . OPTIONAL_PARAM_NUMBER_OF_CODES , '' )
if not ( email and enterprise_name ) :
    raise CodesAPIRequestError ( self . get_missing_params_message ( [ ( self . REQUIRED_PARAM_EMAIL , bool ( email ) ) , ( self . REQUIRED_PARAM_ENTERPRISE_NAME , bool ( enterprise_name ) ) , ] ) )
return email , enterprise_name , number_of_codes


signature:
def get_missing_params_message ( self , parameter_state ) :
    """Get a user - friendly message indicating a missing parameter for the API endpoint ."""

implementation:
params = ', ' . join ( name for name , present in parameter_state if not present )
return self . MISSING_REQUIRED_PARAMS_MSG . format ( params )


signature:
def post ( self , request ) :
    """POST / enterprise / api / v1 / request _ codes
 Requires a JSON object of the following format :
 > > > " email " : " bob @ alice . com " ,
 > > > " enterprise _ name " : " IBM " ,
 > > > " number _ of _ codes " : " 50"
 Keys :
 * email *
 Email of the customer who has requested more codes .
 * enterprise _ name *
 The name of the enterprise requesting more codes .
 * number _ of _ codes *
 The number of codes requested ."""

implementation:
try :
    email , enterprise_name , number_of_codes = self . get_required_query_params ( request )
except CodesAPIRequestError as invalid_request :
    return Response ( { 'error' : str ( invalid_request ) } , status = HTTP_400_BAD_REQUEST )
subject_line = _ ( 'Code Management - Request for Codes by {token_enterprise_name}' ) . format ( token_enterprise_name = enterprise_name )
msg_with_codes = _ ( '{token_email} from {token_enterprise_name} has requested {token_number_codes} additional ' 'codes. Please reach out to them.' ) . format ( token_email = email , token_enterprise_name = enterprise_name , token_number_codes = number_of_codes )
msg_without_codes = _ ( '{token_email} from {token_enterprise_name} has requested additional codes.' ' Please reach out to them.' ) . format ( token_email = email , token_enterprise_name = enterprise_name )
app_config = apps . get_app_config ( "enterprise" )
from_email_address = app_config . customer_success_email
cs_email = app_config . customer_success_email
data = { self . REQUIRED_PARAM_EMAIL : email , self . REQUIRED_PARAM_ENTERPRISE_NAME : enterprise_name , self . OPTIONAL_PARAM_NUMBER_OF_CODES : number_of_codes , }
try :
    mail . send_mail ( subject_line , msg_with_codes if number_of_codes else msg_without_codes , from_email_address , [ cs_email ] , fail_silently = False )
    return Response ( data , status = HTTP_200_OK )
except SMTPException :
    error_message = _ ( '[Enterprise API] Failure in sending e-mail to {token_cs_email} for {token_email}' ' from {token_enterprise_name}' ) . format ( token_cs_email = cs_email , token_email = email , token_enterprise_name = enterprise_name )
    LOGGER . error ( error_message )
    return Response ( { 'error' : str ( 'Request codes email could not be sent' ) } , status = HTTP_500_INTERNAL_SERVER_ERROR )


signature:
def transform_title ( self , content_metadata_item ) :
    """Return the title of the content item ."""

implementation:
title_with_locales = [ ]
for locale in self . enterprise_configuration . get_locales ( ) :
    title_with_locales . append ( { 'locale' : locale , 'value' : content_metadata_item . get ( 'title' , '' ) } )
return title_with_locales


signature:
def transform_description ( self , content_metadata_item ) :
    """Return the description of the content item ."""

implementation:
description_with_locales = [ ]
for locale in self . enterprise_configuration . get_locales ( ) :
    description_with_locales . append ( { 'locale' : locale , 'value' : ( content_metadata_item . get ( 'full_description' ) or content_metadata_item . get ( 'short_description' ) or content_metadata_item . get ( 'title' , '' ) ) } )
return description_with_locales


signature:
def transform_image ( self , content_metadata_item ) :
    """Return the image URI of the content item ."""

implementation:
image_url = ''
if content_metadata_item [ 'content_type' ] in [ 'course' , 'program' ] :
    image_url = content_metadata_item . get ( 'card_image_url' )
elif content_metadata_item [ 'content_type' ] == 'courserun' :
    image_url = content_metadata_item . get ( 'image_url' )
return image_url


signature:
def transform_launch_points ( self , content_metadata_item ) :
    """Return the content metadata item launch points .
 SAPSF allows you to transmit an arry of content launch points which
 are meant to represent sections of a content item which a learner can
 launch into from SAPSF . Currently , we only provide a single launch
 point for a content item ."""

implementation:
return [ { 'providerID' : self . enterprise_configuration . provider_id , 'launchURL' : content_metadata_item [ 'enrollment_url' ] , 'contentTitle' : content_metadata_item [ 'title' ] , 'contentID' : self . get_content_id ( content_metadata_item ) , 'launchType' : 3 , # This tells SAPSF to launch the course in a new browser window .
'mobileEnabled' : True , # Always return True per ENT - 1401
'mobileLaunchURL' : content_metadata_item [ 'enrollment_url' ] , } ]


signature:
def transform_courserun_title ( self , content_metadata_item ) :
    """Return the title of the courserun content item ."""

implementation:
title = content_metadata_item . get ( 'title' ) or ''
course_run_start = content_metadata_item . get ( 'start' )
if course_run_start :
    if course_available_for_enrollment ( content_metadata_item ) :
        title += ' ({starts}: {:%B %Y})' . format ( parse_lms_api_datetime ( course_run_start ) , starts = _ ( 'Starts' ) )
    else :
        title += ' ({:%B %Y} - {enrollment_closed})' . format ( parse_lms_api_datetime ( course_run_start ) , enrollment_closed = _ ( 'Enrollment Closed' ) )
title_with_locales = [ ]
content_metadata_language_code = transform_language_code ( content_metadata_item . get ( 'content_language' , '' ) )
for locale in self . enterprise_configuration . get_locales ( default_locale = content_metadata_language_code ) :
    title_with_locales . append ( { 'locale' : locale , 'value' : title } )
return title_with_locales


signature:
def transform_courserun_description ( self , content_metadata_item ) :
    """Return the description of the courserun content item ."""

implementation:
description_with_locales = [ ]
content_metadata_language_code = transform_language_code ( content_metadata_item . get ( 'content_language' , '' ) )
for locale in self . enterprise_configuration . get_locales ( default_locale = content_metadata_language_code ) :
    description_with_locales . append ( { 'locale' : locale , 'value' : ( content_metadata_item [ 'full_description' ] or content_metadata_item [ 'short_description' ] or content_metadata_item [ 'title' ] or '' ) } )
return description_with_locales


signature:
def transform_courserun_schedule ( self , content_metadata_item ) :
    """Return the schedule of the courseun content item ."""

implementation:
start = content_metadata_item . get ( 'start' ) or UNIX_MIN_DATE_STRING
end = content_metadata_item . get ( 'end' ) or UNIX_MAX_DATE_STRING
return [ { 'startDate' : parse_datetime_to_epoch_millis ( start ) , 'endDate' : parse_datetime_to_epoch_millis ( end ) , 'active' : current_time_is_in_interval ( start , end ) } ]


signature:
def get_content_id ( self , content_metadata_item ) :
    """Return the id for the given content _ metadata _ item , ` uuid ` for programs or ` key ` for other content"""

implementation:
content_id = content_metadata_item . get ( 'key' , '' )
if content_metadata_item [ 'content_type' ] == 'program' :
    content_id = content_metadata_item . get ( 'uuid' , '' )
return content_id


signature:
def parse_datetime_to_epoch ( datestamp , magnitude = 1.0 ) :
    """Convert an ISO - 8601 datetime string to a Unix epoch timestamp in some magnitude .
 By default , returns seconds ."""

implementation:
parsed_datetime = parse_lms_api_datetime ( datestamp )
time_since_epoch = parsed_datetime - UNIX_EPOCH
return int ( time_since_epoch . total_seconds ( ) * magnitude )


signature:
def current_time_is_in_interval ( start , end ) :
    """Determine whether the current time is on the interval [ start , end ] ."""

implementation:
interval_start = parse_lms_api_datetime ( start or UNIX_MIN_DATE_STRING )
interval_end = parse_lms_api_datetime ( end or UNIX_MAX_DATE_STRING )
return interval_start <= timezone . now ( ) <= interval_end


signature:
def chunks ( dictionary , chunk_size ) :
    """Yield successive n - sized chunks from dictionary ."""

implementation:
iterable = iter ( dictionary )
for __ in range ( 0 , len ( dictionary ) , chunk_size ) :
    yield { key : dictionary [ key ] for key in islice ( iterable , chunk_size ) }


signature:
def strfdelta ( tdelta , fmt = '{D:02}d {H:02}h {M:02}m {S:02}s' , input_type = 'timedelta' ) :
    """Convert a datetime . timedelta object or a regular number to a custom - formatted string .
 This function works like the strftime ( ) method works for datetime . datetime
 objects .
 The fmt argument allows custom formatting to be specified . Fields can
 include seconds , minutes , hours , days , and weeks . Each field is optional .
 Arguments :
 tdelta ( datetime . timedelta , int ) : time delta object containing the duration or an integer
 to go with the input _ type .
 fmt ( str ) : Expected format of the time delta . place holders can only be one of the following .
 1 . D to extract days from time delta
 2 . H to extract hours from time delta
 3 . M to extract months from time delta
 4 . S to extract seconds from timedelta
 input _ type ( str ) : The input _ type argument allows tdelta to be a regular number instead of the
 default , which is a datetime . timedelta object .
 Valid input _ type strings :
 1 . ' s ' , ' seconds ' ,
 2 . ' m ' , ' minutes ' ,
 3 . ' h ' , ' hours ' ,
 4 . ' d ' , ' days ' ,
 5 . ' w ' , ' weeks '
 Returns :
 ( str ) : timedelta object interpolated into a string following the given format .
 Examples :
 ' { D : 02 } d { H : 02 } h { M : 02 } m { S : 02 } s ' - - > ' 05d 08h 04m 02s ' ( default )
 ' { W } w { D } d { H } : { M : 02 } : { S : 02 } ' - - > ' 4w 5d 8:04:02'
 ' { D : 2 } d { H : 2 } : { M : 02 } : { S : 02 } ' - - > ' 5d 8:04:02'
 ' { H } h { S } s ' - - > ' 72h 800s '"""

implementation:
# Convert tdelta to integer seconds .
if input_type == 'timedelta' :
    remainder = int ( tdelta . total_seconds ( ) )
elif input_type in [ 's' , 'seconds' ] :
    remainder = int ( tdelta )
elif input_type in [ 'm' , 'minutes' ] :
    remainder = int ( tdelta ) * 60
elif input_type in [ 'h' , 'hours' ] :
    remainder = int ( tdelta ) * 3600
elif input_type in [ 'd' , 'days' ] :
    remainder = int ( tdelta ) * 86400
elif input_type in [ 'w' , 'weeks' ] :
    remainder = int ( tdelta ) * 604800
else :
    raise ValueError ( 'input_type is not valid. Valid input_type strings are: "timedelta", "s", "m", "h", "d", "w"' )
f = Formatter ( )
desired_fields = [ field_tuple [ 1 ] for field_tuple in f . parse ( fmt ) ]
possible_fields = ( 'W' , 'D' , 'H' , 'M' , 'S' )
constants = { 'W' : 604800 , 'D' : 86400 , 'H' : 3600 , 'M' : 60 , 'S' : 1 }
values = { }
for field in possible_fields :
    if field in desired_fields and field in constants :
        values [ field ] , remainder = divmod ( remainder , constants [ field ] )
return f . format ( fmt , ** values )


signature:
def transform_description ( self , content_metadata_item ) :
    """Return the transformed version of the course description .
 We choose one value out of the course ' s full description , short description , and title
 depending on availability and length limits ."""

implementation:
full_description = content_metadata_item . get ( 'full_description' ) or ''
if 0 < len ( full_description ) <= self . LONG_STRING_LIMIT : # pylint : disable = len - as - condition
    return full_description
return content_metadata_item . get ( 'short_description' ) or content_metadata_item . get ( 'title' ) or ''


signature:
def logo_path ( instance , filename ) :
    """Delete the file if it already exist and returns the enterprise customer logo image path .
 Arguments :
 instance ( : class : ` . EnterpriseCustomerBrandingConfiguration ` ) : EnterpriseCustomerBrandingConfiguration object
 filename ( str ) : file to upload
 Returns :
 path : path of image file e . g . enterprise / branding / < model . id > / < model _ id > _ logo . < ext > . lower ( )"""

implementation:
extension = os . path . splitext ( filename ) [ 1 ] . lower ( )
instance_id = str ( instance . id )
fullname = os . path . join ( "enterprise/branding/" , instance_id , instance_id + "_logo" + extension )
if default_storage . exists ( fullname ) :
    default_storage . delete ( fullname )
return fullname


signature:
def get_link_by_email ( self , user_email ) :
    """Return link by email ."""

implementation:
try :
    user = User . objects . get ( email = user_email )
    try :
        return self . get ( user_id = user . id )
    except EnterpriseCustomerUser . DoesNotExist :
        pass
except User . DoesNotExist :
    pass
try :
    return PendingEnterpriseCustomerUser . objects . get ( user_email = user_email )
except PendingEnterpriseCustomerUser . DoesNotExist :
    pass
return None


signature:
def link_user ( self , enterprise_customer , user_email ) :
    """Link user email to Enterprise Customer .
 If : class : ` django . contrib . auth . models . User ` instance with specified email does not exist ,
 : class : ` . PendingEnterpriseCustomerUser ` instance is created instead ."""

implementation:
try :
    existing_user = User . objects . get ( email = user_email )
    self . get_or_create ( enterprise_customer = enterprise_customer , user_id = existing_user . id )
except User . DoesNotExist :
    PendingEnterpriseCustomerUser . objects . get_or_create ( enterprise_customer = enterprise_customer , user_email = user_email )


signature:
def unlink_user ( self , enterprise_customer , user_email ) :
    """Unlink user email from Enterprise Customer .
 If : class : ` django . contrib . auth . models . User ` instance with specified email does not exist ,
 : class : ` . PendingEnterpriseCustomerUser ` instance is deleted instead .
 Raises EnterpriseCustomerUser . DoesNotExist if instance of : class : ` django . contrib . auth . models . User ` with
 specified email exists and corresponding : class : ` . EnterpriseCustomerUser ` instance does not .
 Raises PendingEnterpriseCustomerUser . DoesNotExist exception if instance of
 : class : ` django . contrib . auth . models . User ` with specified email exists and corresponding
 : class : ` . PendingEnterpriseCustomerUser ` instance does not ."""

implementation:
try :
    existing_user = User . objects . get ( email = user_email )
    # not capturing DoesNotExist intentionally to signal to view that link does not exist
    link_record = self . get ( enterprise_customer = enterprise_customer , user_id = existing_user . id )
    link_record . delete ( )
    if update_user : # Remove the SailThru flags for enterprise learner .
        update_user . delay ( sailthru_vars = { 'is_enterprise_learner' : False , 'enterprise_name' : None , } , email = user_email )
except User . DoesNotExist : # not capturing DoesNotExist intentionally to signal to view that link does not exist
    pending_link = PendingEnterpriseCustomerUser . objects . get ( enterprise_customer = enterprise_customer , user_email = user_email )
    pending_link . delete ( )
LOGGER . info ( 'Enterprise learner {%s} successfully unlinked from Enterprise Customer {%s}' , user_email , enterprise_customer . name )


signature:
def enterprise_customer_uuid ( self ) :
    """Get the enterprise customer uuid linked to the user ."""

implementation:
try :
    enterprise_user = EnterpriseCustomerUser . objects . get ( user_id = self . user . id )
except ObjectDoesNotExist :
    LOGGER . warning ( 'User {} has a {} assignment but is not linked to an enterprise!' . format ( self . __class__ , self . user . id ) )
    return None
except MultipleObjectsReturned :
    LOGGER . warning ( 'User {} is linked to multiple enterprises, which is not yet supported!' . format ( self . user . id ) )
    return None
return str ( enterprise_user . enterprise_customer . uuid )


signature:
def get_data_sharing_consent ( username , enterprise_customer_uuid , course_id = None , program_uuid = None ) :
    """Get the data sharing consent object associated with a certain user , enterprise customer , and other scope .
 : param username : The user that grants consent
 : param enterprise _ customer _ uuid : The consent requester
 : param course _ id ( optional ) : A course ID to which consent may be related
 : param program _ uuid ( optional ) : A program to which consent may be related
 : return : The data sharing consent object , or None if the enterprise customer for the given UUID does not exist ."""

implementation:
EnterpriseCustomer = apps . get_model ( 'enterprise' , 'EnterpriseCustomer' )
# pylint : disable = invalid - name
try :
    if course_id :
        return get_course_data_sharing_consent ( username , course_id , enterprise_customer_uuid )
    return get_program_data_sharing_consent ( username , program_uuid , enterprise_customer_uuid )
except EnterpriseCustomer . DoesNotExist :
    return None


signature:
def get_course_data_sharing_consent ( username , course_id , enterprise_customer_uuid ) :
    """Get the data sharing consent object associated with a certain user of a customer for a course .
 : param username : The user that grants consent .
 : param course _ id : The course for which consent is granted .
 : param enterprise _ customer _ uuid : The consent requester .
 : return : The data sharing consent object"""

implementation:
# Prevent circular imports .
DataSharingConsent = apps . get_model ( 'consent' , 'DataSharingConsent' )
# pylint : disable = invalid - name
return DataSharingConsent . objects . proxied_get ( username = username , course_id = course_id , enterprise_customer__uuid = enterprise_customer_uuid )


signature:
def get_program_data_sharing_consent ( username , program_uuid , enterprise_customer_uuid ) :
    """Get the data sharing consent object associated with a certain user of a customer for a program .
 : param username : The user that grants consent .
 : param program _ uuid : The program for which consent is granted .
 : param enterprise _ customer _ uuid : The consent requester .
 : return : The data sharing consent object"""

implementation:
enterprise_customer = get_enterprise_customer ( enterprise_customer_uuid )
discovery_client = CourseCatalogApiServiceClient ( enterprise_customer . site )
course_ids = discovery_client . get_program_course_keys ( program_uuid )
child_consents = ( get_data_sharing_consent ( username , enterprise_customer_uuid , course_id = individual_course_id ) for individual_course_id in course_ids )
return ProxyDataSharingConsent . from_children ( program_uuid , * child_consents )


signature:
def send_course_enrollment_statement ( lrs_configuration , course_enrollment ) :
    """Send xAPI statement for course enrollment .
 Arguments :
 lrs _ configuration ( XAPILRSConfiguration ) : XAPILRSConfiguration instance where to send statements .
 course _ enrollment ( CourseEnrollment ) : Course enrollment object ."""

implementation:
user_details = LearnerInfoSerializer ( course_enrollment . user )
course_details = CourseInfoSerializer ( course_enrollment . course )
statement = LearnerCourseEnrollmentStatement ( course_enrollment . user , course_enrollment . course , user_details . data , course_details . data , )
EnterpriseXAPIClient ( lrs_configuration ) . save_statement ( statement )


signature:
def send_course_completion_statement ( lrs_configuration , user , course_overview , course_grade ) :
    """Send xAPI statement for course completion .
 Arguments :
 lrs _ configuration ( XAPILRSConfiguration ) : XAPILRSConfiguration instance where to send statements .
 user ( User ) : Django User object .
 course _ overview ( CourseOverview ) : Course over view object containing course details .
 course _ grade ( CourseGrade ) : course grade object ."""

implementation:
user_details = LearnerInfoSerializer ( user )
course_details = CourseInfoSerializer ( course_overview )
statement = LearnerCourseCompletionStatement ( user , course_overview , user_details . data , course_details . data , course_grade , )
EnterpriseXAPIClient ( lrs_configuration ) . save_statement ( statement )


signature:
def export ( self ) :
    """Return the exported and transformed content metadata as a dictionary ."""

implementation:
content_metadata_export = { }
content_metadata_items = self . enterprise_api . get_content_metadata ( self . enterprise_customer )
LOGGER . info ( 'Retrieved content metadata for enterprise [%s]' , self . enterprise_customer . name )
for item in content_metadata_items :
    transformed = self . _transform_item ( item )
    LOGGER . info ( 'Exporting content metadata item with plugin configuration [%s]: [%s]' , self . enterprise_configuration , json . dumps ( transformed , indent = 4 ) , )
    content_metadata_item_export = ContentMetadataItemExport ( item , transformed )
    content_metadata_export [ content_metadata_item_export . content_id ] = content_metadata_item_export
return OrderedDict ( sorted ( content_metadata_export . items ( ) ) )


signature:
def _transform_item ( self , content_metadata_item ) :
    """Transform the provided content metadata item to the schema expected by the integrated channel ."""

implementation:
content_metadata_type = content_metadata_item [ 'content_type' ]
transformed_item = { }
for integrated_channel_schema_key , edx_data_schema_key in self . DATA_TRANSFORM_MAPPING . items ( ) : # Look for transformer functions defined on subclasses .
# Favor content type - specific functions .
    transformer = ( getattr ( self , 'transform_{content_type}_{edx_data_schema_key}' . format ( content_type = content_metadata_type , edx_data_schema_key = edx_data_schema_key ) , None ) or getattr ( self , 'transform_{edx_data_schema_key}' . format ( edx_data_schema_key = edx_data_schema_key ) , None ) )
    if transformer :
        transformed_item [ integrated_channel_schema_key ] = transformer ( content_metadata_item )
    else : # The concrete subclass does not define an override for the given field ,
    # so just use the data key to index the content metadata item dictionary .
        try :
            transformed_item [ integrated_channel_schema_key ] = content_metadata_item [ edx_data_schema_key ]
        except KeyError : # There may be a problem with the DATA _ TRANSFORM _ MAPPING on
        # the concrete subclass or the concrete subclass does not implement
        # the appropriate field tranformer function .
            LOGGER . exception ( 'Failed to transform content metadata item field [%s] for [%s]: [%s]' , edx_data_schema_key , self . enterprise_customer . name , content_metadata_item , )
return transformed_item


signature:
def get_consent_record ( self , request ) :
    """Get the consent record relevant to the request at hand ."""

implementation:
username , course_id , program_uuid , enterprise_customer_uuid = self . get_required_query_params ( request )
return get_data_sharing_consent ( username , enterprise_customer_uuid , course_id = course_id , program_uuid = program_uuid )


signature:
def get_required_query_params ( self , request ) :
    """Gets ` ` username ` ` , ` ` course _ id ` ` , and ` ` enterprise _ customer _ uuid ` ` ,
 which are the relevant query parameters for this API endpoint .
 : param request : The request to this endpoint .
 : return : The ` ` username ` ` , ` ` course _ id ` ` , and ` ` enterprise _ customer _ uuid ` ` from the request ."""

implementation:
username = get_request_value ( request , self . REQUIRED_PARAM_USERNAME , '' )
course_id = get_request_value ( request , self . REQUIRED_PARAM_COURSE_ID , '' )
program_uuid = get_request_value ( request , self . REQUIRED_PARAM_PROGRAM_UUID , '' )
enterprise_customer_uuid = get_request_value ( request , self . REQUIRED_PARAM_ENTERPRISE_CUSTOMER )
if not ( username and ( course_id or program_uuid ) and enterprise_customer_uuid ) :
    raise ConsentAPIRequestError ( self . get_missing_params_message ( [ ( "'username'" , bool ( username ) ) , ( "'enterprise_customer_uuid'" , bool ( enterprise_customer_uuid ) ) , ( "one of 'course_id' or 'program_uuid'" , bool ( course_id or program_uuid ) ) , ] ) )
return username , course_id , program_uuid , enterprise_customer_uuid


signature:
def get_no_record_response ( self , request ) :
    """Get an HTTPResponse that can be used when there ' s no related EnterpriseCustomer ."""

implementation:
username , course_id , program_uuid , enterprise_customer_uuid = self . get_required_query_params ( request )
data = { self . REQUIRED_PARAM_USERNAME : username , self . REQUIRED_PARAM_ENTERPRISE_CUSTOMER : enterprise_customer_uuid , self . CONSENT_EXISTS : False , self . CONSENT_GRANTED : False , self . CONSENT_REQUIRED : False , }
if course_id :
    data [ self . REQUIRED_PARAM_COURSE_ID ] = course_id
if program_uuid :
    data [ self . REQUIRED_PARAM_PROGRAM_UUID ] = program_uuid
return Response ( data , status = HTTP_200_OK )


signature:
def get ( self , request ) :
    """GET / consent / api / v1 / data _ sharing _ consent ? username = bob & course _ id = id & enterprise _ customer _ uuid = uuid
 * username *
 The edX username from whom to get consent .
 * course _ id *
 The course for which consent is granted .
 * enterprise _ customer _ uuid *
 The UUID of the enterprise customer that requires consent ."""

implementation:
try :
    consent_record = self . get_consent_record ( request )
    if consent_record is None :
        return self . get_no_record_response ( request )
except ConsentAPIRequestError as invalid_request :
    return Response ( { 'error' : str ( invalid_request ) } , status = HTTP_400_BAD_REQUEST )
return Response ( consent_record . serialize ( ) , status = HTTP_200_OK )


signature:
def post ( self , request ) :
    """POST / consent / api / v1 / data _ sharing _ consent
 Requires a JSON object of the following format :
 > > > " username " : " bob " ,
 > > > " course _ id " : " course - v1 : edX + DemoX + Demo _ Course " ,
 > > > " enterprise _ customer _ uuid " : " enterprise - uuid - goes - right - here "
 Keys :
 * username *
 The edX username from whom to get consent .
 * course _ id *
 The course for which consent is granted .
 * enterprise _ customer _ uuid *
 The UUID of the enterprise customer that requires consent ."""

implementation:
try :
    consent_record = self . get_consent_record ( request )
    if consent_record is None :
        return self . get_no_record_response ( request )
    if consent_record . consent_required ( ) : # If and only if the given EnterpriseCustomer requires data sharing consent
    # for the given course , then , since we ' ve received a POST request , set the
    # consent state for the EC / user / course combo .
        consent_record . granted = True
        # Models don ' t have return values when saving , but ProxyDataSharingConsent
        # objects do - they should return either a model instance , or another instance
        # of ProxyDataSharingConsent if representing a multi - course consent record .
        consent_record = consent_record . save ( ) or consent_record
except ConsentAPIRequestError as invalid_request :
    return Response ( { 'error' : str ( invalid_request ) } , status = HTTP_400_BAD_REQUEST )
return Response ( consent_record . serialize ( ) )


signature:
def delete ( self , request ) :
    """DELETE / consent / api / v1 / data _ sharing _ consent
 Requires a JSON object of the following format :
 > > > " username " : " bob " ,
 > > > " course _ id " : " course - v1 : edX + DemoX + Demo _ Course " ,
 > > > " enterprise _ customer _ uuid " : " enterprise - uuid - goes - right - here "
 Keys :
 * username *
 The edX username from whom to get consent .
 * course _ id *
 The course for which consent is granted .
 * enterprise _ customer _ uuid *
 The UUID of the enterprise customer that requires consent ."""

implementation:
try :
    consent_record = self . get_consent_record ( request )
    if consent_record is None :
        return self . get_no_record_response ( request )
    # We ' re fine with proactively refusing consent , even when there ' s no actual
    # requirement for consent yet .
    consent_record . granted = False
    # Models don ' t have return values when saving , but ProxyDataSharingConsent
    # objects do - they should return either a model instance , or another instance
    # of ProxyDataSharingConsent if representing a multi - course consent record .
    consent_record = consent_record . save ( ) or consent_record
except ConsentAPIRequestError as invalid_request :
    return Response ( { 'error' : str ( invalid_request ) } , status = HTTP_400_BAD_REQUEST )
return Response ( consent_record . serialize ( ) )


signature:
def ready ( self ) :
    """Perform other one - time initialization steps ."""

implementation:
from enterprise . signals import handle_user_post_save
from django . db . models . signals import pre_migrate , post_save
post_save . connect ( handle_user_post_save , sender = self . auth_user_model , dispatch_uid = USER_POST_SAVE_DISPATCH_UID )
pre_migrate . connect ( self . _disconnect_user_post_save_for_migrations )


signature:
def _disconnect_user_post_save_for_migrations ( self , sender , ** kwargs ) : # pylint : disable = unused - argument
    """Handle pre _ migrate signal - disconnect User post _ save handler ."""

implementation:
from django . db . models . signals import post_save
post_save . disconnect ( sender = self . auth_user_model , dispatch_uid = USER_POST_SAVE_DISPATCH_UID )


signature:
def get_actor ( self , username , email ) :
    """Get actor for the statement ."""

implementation: return Agent ( name = username , mbox = 'mailto:{email}' . format ( email = email ) , )


signature:
def get_object ( self , name , description ) :
    """Get object for the statement ."""

implementation: return Activity ( id = X_API_ACTIVITY_COURSE , definition = ActivityDefinition ( name = LanguageMap ( { 'en-US' : ( name or '' ) . encode ( "ascii" , "ignore" ) . decode ( 'ascii' ) } ) , description = LanguageMap ( { 'en-US' : ( description or '' ) . encode ( "ascii" , "ignore" ) . decode ( 'ascii' ) } ) , ) , )


signature:
def parse_csv ( file_stream , expected_columns = None ) :
    """Parse csv file and return a stream of dictionaries representing each row .
 First line of CSV file must contain column headers .
 Arguments :
 file _ stream : input file
 expected _ columns ( set [ unicode ] ) : columns that are expected to be present
 Yields :
 dict : CSV line parsed into a dictionary ."""

implementation:
reader = unicodecsv . DictReader ( file_stream , encoding = "utf-8" )
if expected_columns and set ( expected_columns ) - set ( reader . fieldnames ) :
    raise ValidationError ( ValidationMessages . MISSING_EXPECTED_COLUMNS . format ( expected_columns = ", " . join ( expected_columns ) , actual_columns = ", " . join ( reader . fieldnames ) ) )
# " yield from reader " would be nicer , but we ' re on python2.7 yet .
for row in reader :
    yield row


signature:
def validate_email_to_link ( email , raw_email = None , message_template = None , ignore_existing = False ) :
    """Validate email to be linked to Enterprise Customer .
 Performs two checks :
 * Checks that email is valid
 * Checks that it is not already linked to any Enterprise Customer
 Arguments :
 email ( str ) : user email to link
 raw _ email ( str ) : raw value as it was passed by user - used in error message .
 message _ template ( str ) : Validation error template string .
 ignore _ existing ( bool ) : If True to skip the check for an existing Enterprise Customer
 Raises :
 ValidationError : if email is invalid or already linked to Enterprise Customer .
 Returns :
 bool : Whether or not there is an existing record with the same email address ."""

implementation:
raw_email = raw_email if raw_email is not None else email
message_template = message_template if message_template is not None else ValidationMessages . INVALID_EMAIL
try :
    validate_email ( email )
except ValidationError :
    raise ValidationError ( message_template . format ( argument = raw_email ) )
existing_record = EnterpriseCustomerUser . objects . get_link_by_email ( email )
if existing_record and not ignore_existing :
    raise ValidationError ( ValidationMessages . USER_ALREADY_REGISTERED . format ( email = email , ec_name = existing_record . enterprise_customer . name ) )
return existing_record or False


signature:
def get_course_runs_from_program ( program ) :
    """Return course runs from program data .
 Arguments :
 program ( dict ) : Program data from Course Catalog API
 Returns :
 set : course runs in given program"""

implementation:
course_runs = set ( )
for course in program . get ( "courses" , [ ] ) :
    for run in course . get ( "course_runs" , [ ] ) :
        if "key" in run and run [ "key" ] :
            course_runs . add ( run [ "key" ] )
return course_runs


signature:
def get_earliest_start_date_from_program ( program ) :
    """Get the earliest date that one of the courses in the program was available .
 For the sake of emails to new learners , we treat this as the program start date .
 Arguemnts :
 program ( dict ) : Program data from Course Catalog API
 returns :
 datetime . datetime : The date and time at which the first course started"""

implementation:
start_dates = [ ]
for course in program . get ( 'courses' , [ ] ) :
    for run in course . get ( 'course_runs' , [ ] ) :
        if run . get ( 'start' ) :
            start_dates . append ( parse_lms_api_datetime ( run [ 'start' ] ) )
if not start_dates :
    return None
return min ( start_dates )


signature:
def paginated_list ( object_list , page , page_size = 25 ) :
    """Returns paginated list .
 Arguments :
 object _ list ( QuerySet ) : A list of records to be paginated .
 page ( int ) : Current page number .
 page _ size ( int ) : Number of records displayed in each paginated set .
 show _ all ( bool ) : Whether to show all records .
 Adopted from django / contrib / admin / templatetags / admin _ list . py
 https : / / github . com / django / django / blob / 1.11.1 / django / contrib / admin / templatetags / admin _ list . py # L50"""

implementation:
paginator = CustomPaginator ( object_list , page_size )
try :
    object_list = paginator . page ( page )
except PageNotAnInteger :
    object_list = paginator . page ( 1 )
except EmptyPage :
    object_list = paginator . page ( paginator . num_pages )
page_range = [ ]
page_num = object_list . number
# If there are 10 or fewer pages , display links to every page .
# Otherwise , do some fancy
if paginator . num_pages <= 10 :
    page_range = range ( paginator . num_pages )
else : # Insert " smart " pagination links , so that there are always ON _ ENDS
# links at either end of the list of pages , and there are always
# ON _ EACH _ SIDE links at either end of the " current page " link .
    if page_num > ( PAGES_ON_EACH_SIDE + PAGES_ON_ENDS + 1 ) :
        page_range . extend ( range ( 1 , PAGES_ON_ENDS + 1 ) )
        page_range . append ( DOT )
        page_range . extend ( range ( page_num - PAGES_ON_EACH_SIDE , page_num + 1 ) )
    else :
        page_range . extend ( range ( 1 , page_num + 1 ) )
    if page_num < ( paginator . num_pages - PAGES_ON_EACH_SIDE - PAGES_ON_ENDS ) :
        page_range . extend ( range ( page_num + 1 , page_num + PAGES_ON_EACH_SIDE + 1 ) )
        page_range . append ( DOT )
        page_range . extend ( range ( paginator . num_pages + 1 - PAGES_ON_ENDS , paginator . num_pages + 1 ) )
    else :
        page_range . extend ( range ( page_num + 1 , paginator . num_pages + 1 ) )
    # Override page range to implement custom smart links .
    object_list . paginator . page_range = page_range
return object_list


signature:
def clean_email_or_username ( self ) :
    """Clean email form field
 Returns :
 str : the cleaned value , converted to an email address ( or an empty string )"""

implementation:
email_or_username = self . cleaned_data [ self . Fields . EMAIL_OR_USERNAME ] . strip ( )
if not email_or_username : # The field is blank ; we just return the existing blank value .
    return email_or_username
email = email_or_username__to__email ( email_or_username )
bulk_entry = len ( split_usernames_and_emails ( email ) ) > 1
if bulk_entry :
    for email in split_usernames_and_emails ( email ) :
        validate_email_to_link ( email , None , ValidationMessages . INVALID_EMAIL_OR_USERNAME , ignore_existing = True )
    email = email_or_username
else :
    validate_email_to_link ( email , email_or_username , ValidationMessages . INVALID_EMAIL_OR_USERNAME , ignore_existing = True )
return email


signature:
def clean_course ( self ) :
    """Verify course ID and retrieve course details ."""

implementation:
course_id = self . cleaned_data [ self . Fields . COURSE ] . strip ( )
if not course_id :
    return None
try :
    client = EnrollmentApiClient ( )
    return client . get_course_details ( course_id )
except ( HttpClientError , HttpServerError ) :
    raise ValidationError ( ValidationMessages . INVALID_COURSE_ID . format ( course_id = course_id ) )


signature:
def clean_program ( self ) :
    """Clean program .
 Try obtaining program treating form value as program UUID or title .
 Returns :
 dict : Program information if program found"""

implementation:
program_id = self . cleaned_data [ self . Fields . PROGRAM ] . strip ( )
if not program_id :
    return None
try :
    client = CourseCatalogApiClient ( self . _user , self . _enterprise_customer . site )
    program = client . get_program_by_uuid ( program_id ) or client . get_program_by_title ( program_id )
except MultipleProgramMatchError as exc :
    raise ValidationError ( ValidationMessages . MULTIPLE_PROGRAM_MATCH . format ( program_count = exc . programs_matched ) )
except ( HttpClientError , HttpServerError ) :
    raise ValidationError ( ValidationMessages . INVALID_PROGRAM_ID . format ( program_id = program_id ) )
if not program :
    raise ValidationError ( ValidationMessages . INVALID_PROGRAM_ID . format ( program_id = program_id ) )
if program [ 'status' ] != ProgramStatuses . ACTIVE :
    raise ValidationError ( ValidationMessages . PROGRAM_IS_INACTIVE . format ( program_id = program_id , status = program [ 'status' ] ) )
return program


signature:
def clean_notify ( self ) :
    """Clean the notify _ on _ enrollment field ."""

implementation: return self . cleaned_data . get ( self . Fields . NOTIFY , self . NotificationTypes . DEFAULT )


signature:
def clean ( self ) :
    """Clean fields that depend on each other .
 In this case , the form can be used to link single user or bulk link multiple users . These are mutually
 exclusive modes , so this method checks that only one field is passed ."""

implementation:
cleaned_data = super ( ManageLearnersForm , self ) . clean ( )
# Here we take values from ` data ` ( and not ` cleaned _ data ` ) as we need raw values - field clean methods
# might " invalidate " the value and set it to None , while all we care here is if it was provided at all or not
email_or_username = self . data . get ( self . Fields . EMAIL_OR_USERNAME , None )
bulk_upload_csv = self . files . get ( self . Fields . BULK_UPLOAD , None )
if not email_or_username and not bulk_upload_csv :
    raise ValidationError ( ValidationMessages . NO_FIELDS_SPECIFIED )
if email_or_username and bulk_upload_csv :
    raise ValidationError ( ValidationMessages . BOTH_FIELDS_SPECIFIED )
if email_or_username :
    mode = self . Modes . MODE_SINGULAR
else :
    mode = self . Modes . MODE_BULK
cleaned_data [ self . Fields . MODE ] = mode
cleaned_data [ self . Fields . NOTIFY ] = self . clean_notify ( )
self . _validate_course ( )
self . _validate_program ( )
if self . data . get ( self . Fields . PROGRAM , None ) and self . data . get ( self . Fields . COURSE , None ) :
    raise ValidationError ( ValidationMessages . COURSE_AND_PROGRAM_ERROR )
return cleaned_data


signature:
def _validate_course ( self ) :
    """Verify that the selected mode is valid for the given course ."""

implementation:
# Verify that the selected mode is valid for the given course .
course_details = self . cleaned_data . get ( self . Fields . COURSE )
if course_details :
    course_mode = self . cleaned_data . get ( self . Fields . COURSE_MODE )
    if not course_mode :
        raise ValidationError ( ValidationMessages . COURSE_WITHOUT_COURSE_MODE )
    valid_course_modes = course_details [ "course_modes" ]
    if all ( course_mode != mode [ "slug" ] for mode in valid_course_modes ) :
        error = ValidationError ( ValidationMessages . COURSE_MODE_INVALID_FOR_COURSE . format ( course_mode = course_mode , course_id = course_details [ "course_id" ] , ) )
        raise ValidationError ( { self . Fields . COURSE_MODE : error } )


signature:
def _validate_program ( self ) :
    """Verify that selected mode is available for program and all courses in the program"""

implementation:
program = self . cleaned_data . get ( self . Fields . PROGRAM )
if not program :
    return
course_runs = get_course_runs_from_program ( program )
try :
    client = CourseCatalogApiClient ( self . _user , self . _enterprise_customer . site )
    available_modes = client . get_common_course_modes ( course_runs )
    course_mode = self . cleaned_data . get ( self . Fields . COURSE_MODE )
except ( HttpClientError , HttpServerError ) :
    raise ValidationError ( ValidationMessages . FAILED_TO_OBTAIN_COURSE_MODES . format ( program_title = program . get ( "title" ) ) )
if not course_mode :
    raise ValidationError ( ValidationMessages . COURSE_WITHOUT_COURSE_MODE )
if course_mode not in available_modes :
    raise ValidationError ( ValidationMessages . COURSE_MODE_NOT_AVAILABLE . format ( mode = course_mode , program_title = program . get ( "title" ) , modes = ", " . join ( available_modes ) ) )


signature:
def get_catalog_options ( self ) :
    """Retrieve a list of catalog ID and name pairs .
 Once retrieved , these name pairs can be used directly as a value
 for the ` choices ` argument to a ChoiceField ."""

implementation:
# TODO : We will remove the discovery service catalog implementation
# once we have fully migrated customer ' s to EnterpriseCustomerCatalogs .
# For now , this code will prevent an admin from creating a new
# EnterpriseCustomer with a discovery service catalog . They will have to first
# save the EnterpriseCustomer admin form and then edit the EnterpriseCustomer
# to add a discovery service catalog .
if hasattr ( self . instance , 'site' ) :
    catalog_api = CourseCatalogApiClient ( self . user , self . instance . site )
else :
    catalog_api = CourseCatalogApiClient ( self . user )
catalogs = catalog_api . get_all_catalogs ( )
# order catalogs by name .
catalogs = sorted ( catalogs , key = lambda catalog : catalog . get ( 'name' , '' ) . lower ( ) )
return BLANK_CHOICE_DASH + [ ( catalog [ 'id' ] , catalog [ 'name' ] , ) for catalog in catalogs ]


signature:
def clean ( self ) :
    """Clean form fields prior to database entry .
 In this case , the major cleaning operation is substituting a None value for a blank
 value in the Catalog field ."""

implementation:
cleaned_data = super ( EnterpriseCustomerAdminForm , self ) . clean ( )
if 'catalog' in cleaned_data and not cleaned_data [ 'catalog' ] :
    cleaned_data [ 'catalog' ] = None
return cleaned_data


signature:
def clean ( self ) :
    """Final validations of model fields .
 1 . Validate that selected site for enterprise customer matches with the selected identity provider ' s site ."""

implementation:
super ( EnterpriseCustomerIdentityProviderAdminForm , self ) . clean ( )
provider_id = self . cleaned_data . get ( 'provider_id' , None )
enterprise_customer = self . cleaned_data . get ( 'enterprise_customer' , None )
if provider_id is None or enterprise_customer is None : # field validation for either provider _ id or enterprise _ customer has already raised
# a validation error .
    return
identity_provider = utils . get_identity_provider ( provider_id )
if not identity_provider : # This should not happen , as identity providers displayed in drop down are fetched dynamically .
    message = _ ( "The specified Identity Provider does not exist. For more " "information, contact a system administrator." , )
    # Log message for debugging
    logger . exception ( message )
    raise ValidationError ( message )
if identity_provider and identity_provider . site != enterprise_customer . site :
    raise ValidationError ( _ ( "The site for the selected identity provider " "({identity_provider_site}) does not match the site for " "this enterprise customer ({enterprise_customer_site}). " "To correct this problem, select a site that has a domain " "of '{identity_provider_site}', or update the identity " "provider to '{enterprise_customer_site}'." ) . format ( enterprise_customer_site = enterprise_customer . site , identity_provider_site = identity_provider . site , ) , )


signature:
def clean ( self ) :
    """Override of clean method to perform additional validation"""

implementation:
cleaned_data = super ( EnterpriseCustomerReportingConfigAdminForm , self ) . clean ( )
report_customer = cleaned_data . get ( 'enterprise_customer' )
# Check that any selected catalogs are tied to the selected enterprise .
invalid_catalogs = [ '{} ({})' . format ( catalog . title , catalog . uuid ) for catalog in cleaned_data . get ( 'enterprise_customer_catalogs' ) if catalog . enterprise_customer != report_customer ]
if invalid_catalogs :
    message = _ ( 'These catalogs for reporting do not match enterprise' 'customer {enterprise_customer}: {invalid_catalogs}' , ) . format ( enterprise_customer = report_customer , invalid_catalogs = invalid_catalogs , )
    self . add_error ( 'enterprise_customer_catalogs' , message )


signature:
def clean_channel_worker_username ( self ) :
    """Clean enterprise channel worker user form field
 Returns :
 str : the cleaned value of channel user username for transmitting courses metadata ."""

implementation:
channel_worker_username = self . cleaned_data [ 'channel_worker_username' ] . strip ( )
try :
    User . objects . get ( username = channel_worker_username )
except User . DoesNotExist :
    raise ValidationError ( ValidationMessages . INVALID_CHANNEL_WORKER . format ( channel_worker_username = channel_worker_username ) )
return channel_worker_username


signature:
def verify_edx_resources ( ) :
    """Ensure that all necessary resources to render the view are present ."""

implementation:
required_methods = { 'ProgramDataExtender' : ProgramDataExtender , }
for method in required_methods :
    if required_methods [ method ] is None :
        raise NotConnectedToOpenEdX ( _ ( "The following method from the Open edX platform is necessary for this view but isn't available." ) + "\nUnavailable: {method}" . format ( method = method ) )


signature:
def get_global_context ( request , enterprise_customer ) :
    """Get the set of variables that are needed by default across views ."""

implementation:
platform_name = get_configuration_value ( "PLATFORM_NAME" , settings . PLATFORM_NAME )
# pylint : disable = no - member
return { 'enterprise_customer' : enterprise_customer , 'LMS_SEGMENT_KEY' : settings . LMS_SEGMENT_KEY , 'LANGUAGE_CODE' : get_language_from_request ( request ) , 'tagline' : get_configuration_value ( "ENTERPRISE_TAGLINE" , settings . ENTERPRISE_TAGLINE ) , 'platform_description' : get_configuration_value ( "PLATFORM_DESCRIPTION" , settings . PLATFORM_DESCRIPTION , ) , 'LMS_ROOT_URL' : settings . LMS_ROOT_URL , 'platform_name' : platform_name , 'header_logo_alt_text' : _ ( '{platform_name} home page' ) . format ( platform_name = platform_name ) , 'welcome_text' : constants . WELCOME_TEXT . format ( platform_name = platform_name ) , 'enterprise_welcome_text' : constants . ENTERPRISE_WELCOME_TEXT . format ( enterprise_customer_name = enterprise_customer . name , platform_name = platform_name , strong_start = '' , strong_end = '' , line_break = '
' , privacy_policy_link_start = "" . format ( pp_url = get_configuration_value ( 'PRIVACY' , 'https://www.edx.org/edx-privacy-policy' , type = 'url' ) , ) , privacy_policy_link_end = "" , ) , }


signature:
def get_price_text ( price , request ) :
    """Return the localized converted price as string ( ex . ' $ 150 USD ' ) .
 If the local _ currency switch is enabled and the users location has been determined this will convert the
 given price based on conversion rate from the Catalog service and return a localized string"""

implementation:
if waffle . switch_is_active ( 'local_currency' ) and get_localized_price_text :
    return get_localized_price_text ( price , request )
return format_price ( price )


signature:
def render_page_with_error_code_message ( request , context_data , error_code , log_message ) :
    """Return a 404 page with specified error _ code after logging error and adding message to django messages ."""

implementation:
LOGGER . error ( log_message )
messages . add_generic_error_message_with_code ( request , error_code )
return render ( request , ENTERPRISE_GENERAL_ERROR_PAGE , context = context_data , status = 404 , )


signature:
def course_or_program_exist ( self , course_id , program_uuid ) :
    """Return whether the input course or program exist ."""

implementation:
course_exists = course_id and CourseApiClient ( ) . get_course_details ( course_id )
program_exists = program_uuid and CourseCatalogApiServiceClient ( ) . program_exists ( program_uuid )
return course_exists or program_exists


signature:
def get_default_context ( self , enterprise_customer , platform_name ) :
    """Get the set of variables that will populate the template by default ."""

implementation:
context_data = { 'page_title' : _ ( 'Data sharing consent required' ) , 'consent_message_header' : _ ( 'Consent to share your data' ) , 'requested_permissions_header' : _ ( 'Per the {start_link}Data Sharing Policy{end_link}, ' '{bold_start}{enterprise_customer_name}{bold_end} would like to know about:' ) . format ( enterprise_customer_name = enterprise_customer . name , bold_start = '' , bold_end = '' , start_link = '' , end_link = '' , ) , 'agreement_text' : _ ( 'I agree to allow {platform_name} to share data about my enrollment, completion and performance in all ' '{platform_name} courses and programs where my enrollment is sponsored by {enterprise_customer_name}.' ) . format ( enterprise_customer_name = enterprise_customer . name , platform_name = platform_name , ) , 'continue_text' : _ ( 'Yes, continue' ) , 'abort_text' : _ ( 'No, take me back.' ) , 'policy_dropdown_header' : _ ( 'Data Sharing Policy' ) , 'sharable_items_header' : _ ( 'Enrollment, completion, and performance data that may be shared with {enterprise_customer_name} ' '(or its designee) for these courses and programs are limited to the following:' ) . format ( enterprise_customer_name = enterprise_customer . name ) , 'sharable_items' : [ _ ( 'My email address for my {platform_name} account, ' 'and the date when I created my {platform_name} account' ) . format ( platform_name = platform_name ) , _ ( 'My {platform_name} ID, and if I log in via single sign-on, ' 'my {enterprise_customer_name} SSO user-ID' ) . format ( platform_name = platform_name , enterprise_customer_name = enterprise_customer . name , ) , _ ( 'My {platform_name} username' ) . format ( platform_name = platform_name ) , _ ( 'My country or region of residence' ) , _ ( 'What courses and/or programs I\'ve enrolled in or unenrolled from, what track I ' 'enrolled in (audit or verified) and the date when I enrolled in each course or program' ) , _ ( 'Information about each course or program I\'ve enrolled in, ' 'including its duration and level of effort required' ) , _ ( 'Whether I completed specific parts of each course or program (for example, whether ' 'I watched a given video or completed a given homework assignment)' ) , _ ( 'My overall percentage completion of each course or program on a periodic basis, ' 'including the total time spent in each course or program and the date when I last ' 'logged in to each course or program' ) , _ ( 'My performance in each course or program' ) , _ ( 'My final grade in each course or program, and the date when I completed each course or program' ) , _ ( 'Whether I received a certificate in each course or program' ) , ] , 'sharable_items_footer' : _ ( 'My permission applies only to data from courses or programs that are sponsored by ' '{enterprise_customer_name}, and not to data from any {platform_name} courses or programs that ' 'I take on my own. I understand that I may withdraw my permission only by fully unenrolling ' 'from any courses or programs that are sponsored by {enterprise_customer_name}.' ) . format ( enterprise_customer_name = enterprise_customer . name , platform_name = platform_name , ) , 'sharable_items_note_header' : _ ( 'Please note' ) , 'sharable_items_notes' : [ _ ( 'If you decline to consent, that fact may be shared with {enterprise_customer_name}.' ) . format ( enterprise_customer_name = enterprise_customer . name ) , ] , 'confirmation_modal_header' : _ ( 'Are you aware...' ) , 'confirmation_modal_affirm_decline_text' : _ ( 'I decline' ) , 'confirmation_modal_abort_decline_text' : _ ( 'View the data sharing policy' ) , 'policy_link_template' : _ ( 'View the {start_link}data sharing policy{end_link}.' ) . format ( start_link = '' , end_link = '' , ) , 'policy_return_link_text' : _ ( 'Return to Top' ) , }
return context_data


signature:
def get_context_from_db ( self , consent_page , platform_name , item , context ) :
    """Make set of variables ( populated from db ) that will be used in data sharing consent page ."""

implementation:
enterprise_customer = consent_page . enterprise_customer
course_title = context . get ( 'course_title' , None )
course_start_date = context . get ( 'course_start_date' , None )
context_data = { 'text_override_available' : True , 'page_title' : consent_page . page_title , 'left_sidebar_text' : consent_page . left_sidebar_text . format ( enterprise_customer_name = enterprise_customer . name , platform_name = platform_name , item = item , course_title = course_title , course_start_date = course_start_date , ) , 'top_paragraph' : consent_page . top_paragraph . format ( enterprise_customer_name = enterprise_customer . name , platform_name = platform_name , item = item , course_title = course_title , course_start_date = course_start_date , ) , 'agreement_text' : consent_page . agreement_text . format ( enterprise_customer_name = enterprise_customer . name , platform_name = platform_name , item = item , course_title = course_title , course_start_date = course_start_date , ) , 'continue_text' : consent_page . continue_text , 'abort_text' : consent_page . abort_text , 'policy_dropdown_header' : consent_page . policy_dropdown_header , 'policy_paragraph' : consent_page . policy_paragraph . format ( enterprise_customer_name = enterprise_customer . name , platform_name = platform_name , item = item , course_title = course_title , course_start_date = course_start_date , ) , 'confirmation_modal_header' : consent_page . confirmation_modal_header . format ( enterprise_customer_name = enterprise_customer . name , platform_name = platform_name , item = item , course_title = course_title , course_start_date = course_start_date , ) , 'confirmation_alert_prompt' : consent_page . confirmation_modal_text . format ( enterprise_customer_name = enterprise_customer . name , platform_name = platform_name , item = item , course_title = course_title , course_start_date = course_start_date , ) , 'confirmation_modal_affirm_decline_text' : consent_page . modal_affirm_decline_text , 'confirmation_modal_abort_decline_text' : consent_page . modal_abort_decline_text , }
return context_data


signature:
def get_course_or_program_context ( self , enterprise_customer , course_id = None , program_uuid = None ) :
    """Return a dict having course or program specific keys for data sharing consent page ."""

implementation:
context_data = { }
if course_id :
    context_data . update ( { 'course_id' : course_id , 'course_specific' : True } )
    if not self . preview_mode :
        try :
            catalog_api_client = CourseCatalogApiServiceClient ( enterprise_customer . site )
        except ImproperlyConfigured :
            raise Http404
        course_run_details = catalog_api_client . get_course_run ( course_id )
        course_start_date = ''
        if course_run_details [ 'start' ] :
            course_start_date = parse ( course_run_details [ 'start' ] ) . strftime ( '%B %d, %Y' )
        context_data . update ( { 'course_title' : course_run_details [ 'title' ] , 'course_start_date' : course_start_date , } )
    else :
        context_data . update ( { 'course_title' : 'Demo Course' , 'course_start_date' : datetime . datetime . now ( ) . strftime ( '%B %d, %Y' ) , } )
else :
    context_data . update ( { 'program_uuid' : program_uuid , 'program_specific' : True , } )
return context_data


signature:
def get ( self , request ) :
    """Render a form to collect user input about data sharing consent ."""

implementation:
enterprise_customer_uuid = request . GET . get ( 'enterprise_customer_uuid' )
success_url = request . GET . get ( 'next' )
failure_url = request . GET . get ( 'failure_url' )
course_id = request . GET . get ( 'course_id' , '' )
program_uuid = request . GET . get ( 'program_uuid' , '' )
self . preview_mode = bool ( request . GET . get ( 'preview_mode' , False ) )
# Get enterprise _ customer to start in case we need to render a custom 404 page
# Then go through other business logic to determine ( and potentially overwrite ) the enterprise customer
enterprise_customer = get_enterprise_customer_or_404 ( enterprise_customer_uuid )
context_data = get_global_context ( request , enterprise_customer )
if not self . preview_mode :
    if not self . course_or_program_exist ( course_id , program_uuid ) :
        error_code = 'ENTGDS000'
        log_message = ( 'Neither the course with course_id: {course_id} ' 'or program with {program_uuid} exist for ' 'enterprise customer {enterprise_customer_uuid}' 'Error code {error_code} presented to user {userid}' . format ( course_id = course_id , program_uuid = program_uuid , error_code = error_code , userid = request . user . id , enterprise_customer_uuid = enterprise_customer_uuid , ) )
        return render_page_with_error_code_message ( request , context_data , error_code , log_message )
    try :
        consent_record = get_data_sharing_consent ( request . user . username , enterprise_customer_uuid , program_uuid = program_uuid , course_id = course_id )
    except NotConnectedToOpenEdX as error :
        error_code = 'ENTGDS001'
        log_message = ( 'The was a problem with getting the consent record of user {userid} with ' 'uuid {enterprise_customer_uuid}. get_data_sharing_consent threw ' 'the following NotConnectedToOpenEdX error: {error}' 'for course_id {course_id}.' 'Error code {error_code} presented to user' . format ( userid = request . user . id , enterprise_customer_uuid = enterprise_customer_uuid , error = error , error_code = error_code , course_id = course_id , ) )
        return render_page_with_error_code_message ( request , context_data , error_code , log_message )
    try :
        consent_required = consent_record . consent_required ( )
    except AttributeError :
        consent_required = None
    if consent_record is None or not consent_required :
        error_code = 'ENTGDS002'
        log_message = ( 'The was a problem with the consent record of user {userid} with ' 'enterprise_customer_uuid {enterprise_customer_uuid}. consent_record has a value ' 'of {consent_record} and consent_record.consent_required() a ' 'value of {consent_required} for course_id {course_id}. ' 'Error code {error_code} presented to user' . format ( userid = request . user . id , enterprise_customer_uuid = enterprise_customer_uuid , consent_record = consent_record , consent_required = consent_required , error_code = error_code , course_id = course_id , ) )
        return render_page_with_error_code_message ( request , context_data , error_code , log_message )
    else :
        enterprise_customer = consent_record . enterprise_customer
elif not request . user . is_staff :
    raise PermissionDenied ( )
# Retrieve context data again now that enterprise _ customer logic has been run
context_data = get_global_context ( request , enterprise_customer )
if not ( enterprise_customer_uuid and success_url and failure_url ) :
    error_code = 'ENTGDS003'
    log_message = ( 'Error: one or more of the following values was falsy: ' 'enterprise_customer_uuid: {enterprise_customer_uuid}, ' 'success_url: {success_url}, ' 'failure_url: {failure_url} for course id {course_id}' 'The following error code was reported to user {userid}: {error_code}' . format ( userid = request . user . id , enterprise_customer_uuid = enterprise_customer_uuid , success_url = success_url , failure_url = failure_url , error_code = error_code , course_id = course_id , ) )
    return render_page_with_error_code_message ( request , context_data , error_code , log_message )
try :
    updated_context_dict = self . get_course_or_program_context ( enterprise_customer , course_id = course_id , program_uuid = program_uuid )
    context_data . update ( updated_context_dict )
except Http404 :
    error_code = 'ENTGDS004'
    log_message = ( 'CourseCatalogApiServiceClient is improperly configured. ' 'Returned error code {error_code} to user {userid} ' 'and enterprise_customer {enterprise_customer} ' 'for course_id {course_id}' . format ( error_code = error_code , userid = request . user . id , enterprise_customer = enterprise_customer . uuid , course_id = course_id , ) )
    return render_page_with_error_code_message ( request , context_data , error_code , log_message )
item = 'course' if course_id else 'program'
# Translators : bold _ start and bold _ end are HTML tags for specifying enterprise name in bold text .
context_data . update ( { 'consent_request_prompt' : _ ( 'To access this {item}, you must first consent to share your learning achievements ' 'with {bold_start}{enterprise_customer_name}{bold_end}.' ) . format ( enterprise_customer_name = enterprise_customer . name , bold_start = '' , bold_end = '' , item = item , ) , 'confirmation_alert_prompt' : _ ( 'In order to start this {item} and use your discount, {bold_start}you must{bold_end} consent ' 'to share your {item} data with {enterprise_customer_name}.' ) . format ( enterprise_customer_name = enterprise_customer . name , bold_start = '' , bold_end = '' , item = item , ) , 'redirect_url' : success_url , 'failure_url' : failure_url , 'defer_creation' : request . GET . get ( 'defer_creation' ) is not None , 'requested_permissions' : [ _ ( 'your enrollment in this {item}' ) . format ( item = item ) , _ ( 'your learning progress' ) , _ ( 'course completion' ) , ] , 'policy_link_template' : '' , } )
platform_name = context_data [ 'platform_name' ]
published_only = False if self . preview_mode else True
enterprise_consent_page = enterprise_customer . get_data_sharing_consent_text_overrides ( published_only = published_only )
if enterprise_consent_page :
    context_data . update ( self . get_context_from_db ( enterprise_consent_page , platform_name , item , context_data ) )
else :
    context_data . update ( self . get_default_context ( enterprise_customer , platform_name ) )
return render ( request , 'enterprise/grant_data_sharing_permissions.html' , context = context_data )


signature:
def post ( self , request ) :
    """Process the above form ."""

implementation:
enterprise_uuid = request . POST . get ( 'enterprise_customer_uuid' )
success_url = request . POST . get ( 'redirect_url' )
failure_url = request . POST . get ( 'failure_url' )
course_id = request . POST . get ( 'course_id' , '' )
program_uuid = request . POST . get ( 'program_uuid' , '' )
enterprise_customer = get_enterprise_customer_or_404 ( enterprise_uuid )
context_data = get_global_context ( request , enterprise_customer )
if not ( enterprise_uuid and success_url and failure_url ) :
    error_code = 'ENTGDS005'
    log_message = ( 'Error: one or more of the following values was falsy: ' 'enterprise_uuid: {enterprise_uuid}, ' 'success_url: {success_url}, ' 'failure_url: {failure_url} for course_id {course_id}. ' 'The following error code was reported to the user {userid}: {error_code}' . format ( userid = request . user . id , enterprise_uuid = enterprise_uuid , success_url = success_url , failure_url = failure_url , error_code = error_code , course_id = course_id , ) )
    return render_page_with_error_code_message ( request , context_data , error_code , log_message )
if not self . course_or_program_exist ( course_id , program_uuid ) :
    error_code = 'ENTGDS006'
    log_message = ( 'Neither the course with course_id: {course_id} ' 'or program with {program_uuid} exist for ' 'enterprise customer {enterprise_uuid}' 'Error code {error_code} presented to user {userid}' . format ( course_id = course_id , program_uuid = program_uuid , error_code = error_code , userid = request . user . id , enterprise_uuid = enterprise_uuid , ) )
    return render_page_with_error_code_message ( request , context_data , error_code , log_message )
consent_record = get_data_sharing_consent ( request . user . username , enterprise_uuid , program_uuid = program_uuid , course_id = course_id )
if consent_record is None :
    error_code = 'ENTGDS007'
    log_message = ( 'The was a problem with the consent record of user {userid} with ' 'enterprise_uuid {enterprise_uuid}. consent_record has a value ' 'of {consent_record} and a ' 'value for course_id {course_id}. ' 'Error code {error_code} presented to user' . format ( userid = request . user . id , enterprise_uuid = enterprise_uuid , consent_record = consent_record , error_code = error_code , course_id = course_id , ) )
    return render_page_with_error_code_message ( request , context_data , error_code , log_message )
defer_creation = request . POST . get ( 'defer_creation' )
consent_provided = bool ( request . POST . get ( 'data_sharing_consent' , False ) )
if defer_creation is None and consent_record . consent_required ( ) :
    if course_id :
        enterprise_customer_user , __ = EnterpriseCustomerUser . objects . get_or_create ( enterprise_customer = consent_record . enterprise_customer , user_id = request . user . id )
        enterprise_customer_user . update_session ( request )
        __ , created = EnterpriseCourseEnrollment . objects . get_or_create ( enterprise_customer_user = enterprise_customer_user , course_id = course_id , )
        if created :
            track_enrollment ( 'data-consent-page-enrollment' , request . user . id , course_id , request . path )
    consent_record . granted = consent_provided
    consent_record . save ( )
return redirect ( success_url if consent_provided else failure_url )


signature:
def get ( self , request , enterprise_uuid , course_id ) :
    """Handle the enrollment of enterprise learner in the provided course .
 Based on ` enterprise _ uuid ` in URL , the view will decide which
 enterprise customer ' s course enrollment record should be created .
 Depending on the value of query parameter ` course _ mode ` then learner
 will be either redirected to LMS dashboard for audit modes or
 redirected to ecommerce basket flow for payment of premium modes ."""

implementation:
enrollment_course_mode = request . GET . get ( 'course_mode' )
enterprise_catalog_uuid = request . GET . get ( 'catalog' )
# Redirect the learner to LMS dashboard in case no course mode is
# provided as query parameter ` course _ mode `
if not enrollment_course_mode :
    return redirect ( LMS_DASHBOARD_URL )
enrollment_api_client = EnrollmentApiClient ( )
course_modes = enrollment_api_client . get_course_modes ( course_id )
# Verify that the request user belongs to the enterprise against the
# provided ` enterprise _ uuid ` .
enterprise_customer = get_enterprise_customer_or_404 ( enterprise_uuid )
enterprise_customer_user = get_enterprise_customer_user ( request . user . id , enterprise_customer . uuid )
if not course_modes :
    context_data = get_global_context ( request , enterprise_customer )
    error_code = 'ENTHCE000'
    log_message = ( 'No course_modes for course_id {course_id} for enterprise_catalog_uuid ' '{enterprise_catalog_uuid}.' 'The following error was presented to ' 'user {userid}: {error_code}' . format ( userid = request . user . id , enterprise_catalog_uuid = enterprise_catalog_uuid , course_id = course_id , error_code = error_code ) )
    return render_page_with_error_code_message ( request , context_data , error_code , log_message )
selected_course_mode = None
for course_mode in course_modes :
    if course_mode [ 'slug' ] == enrollment_course_mode :
        selected_course_mode = course_mode
        break
if not selected_course_mode :
    return redirect ( LMS_DASHBOARD_URL )
# Create the Enterprise backend database records for this course
# enrollment
__ , created = EnterpriseCourseEnrollment . objects . get_or_create ( enterprise_customer_user = enterprise_customer_user , course_id = course_id , )
if created :
    track_enrollment ( 'course-landing-page-enrollment' , request . user . id , course_id , request . get_full_path ( ) )
DataSharingConsent . objects . update_or_create ( username = enterprise_customer_user . username , course_id = course_id , enterprise_customer = enterprise_customer_user . enterprise_customer , defaults = { 'granted' : True } , )
audit_modes = getattr ( settings , 'ENTERPRISE_COURSE_ENROLLMENT_AUDIT_MODES' , [ 'audit' , 'honor' ] )
if selected_course_mode [ 'slug' ] in audit_modes : # In case of Audit course modes enroll the learner directly through
# enrollment API client and redirect the learner to dashboard .
    enrollment_api_client . enroll_user_in_course ( request . user . username , course_id , selected_course_mode [ 'slug' ] )
    return redirect ( LMS_COURSEWARE_URL . format ( course_id = course_id ) )
# redirect the enterprise learner to the ecommerce flow in LMS
# Note : LMS start flow automatically detects the paid mode
premium_flow = LMS_START_PREMIUM_COURSE_FLOW_URL . format ( course_id = course_id )
if enterprise_catalog_uuid :
    premium_flow += '?catalog={catalog_uuid}' . format ( catalog_uuid = enterprise_catalog_uuid )
return redirect ( premium_flow )


signature:
def set_final_prices ( self , modes , request ) :
    """Set the final discounted price on each premium mode ."""

implementation:
result = [ ]
for mode in modes :
    if mode [ 'premium' ] :
        mode [ 'final_price' ] = EcommerceApiClient ( request . user ) . get_course_final_price ( mode = mode , enterprise_catalog_uuid = request . GET . get ( 'catalog' ) if request . method == 'GET' else None , )
    result . append ( mode )
return result


signature:
def get_available_course_modes ( self , request , course_run_id , enterprise_catalog ) :
    """Return the available course modes for the course run .
 The provided EnterpriseCustomerCatalog is used to filter and order the
 course modes returned using the EnterpriseCustomerCatalog ' s
 field " enabled _ course _ modes " ."""

implementation:
modes = EnrollmentApiClient ( ) . get_course_modes ( course_run_id )
if not modes :
    LOGGER . warning ( 'Unable to get course modes for course run id {course_run_id}.' . format ( course_run_id = course_run_id ) )
    messages . add_generic_info_message_for_error ( request )
if enterprise_catalog : # filter and order course modes according to the enterprise catalog
    modes = [ mode for mode in modes if mode [ 'slug' ] in enterprise_catalog . enabled_course_modes ]
    modes . sort ( key = lambda course_mode : enterprise_catalog . enabled_course_modes . index ( course_mode [ 'slug' ] ) )
    if not modes :
        LOGGER . info ( 'No matching course modes found for course run {course_run_id} in ' 'EnterpriseCustomerCatalog [{enterprise_catalog_uuid}]' . format ( course_run_id = course_run_id , enterprise_catalog_uuid = enterprise_catalog , ) )
        messages . add_generic_info_message_for_error ( request )
return modes


signature:
def get_base_details ( self , request , enterprise_uuid , course_run_id ) :
    """Retrieve fundamental details used by both POST and GET versions of this view .
 Specifically , take an EnterpriseCustomer UUID and a course run ID , and transform those
 into an actual EnterpriseCustomer , a set of details about the course , and a list
 of the available course modes for that course run ."""

implementation:
enterprise_customer = get_enterprise_customer_or_404 ( enterprise_uuid )
# If the catalog query parameter was provided , we need to scope
# this request to the specified EnterpriseCustomerCatalog .
enterprise_catalog_uuid = request . GET . get ( 'catalog' )
enterprise_catalog = None
if enterprise_catalog_uuid :
    try :
        enterprise_catalog_uuid = UUID ( enterprise_catalog_uuid )
        enterprise_catalog = enterprise_customer . enterprise_customer_catalogs . get ( uuid = enterprise_catalog_uuid )
    except ( ValueError , EnterpriseCustomerCatalog . DoesNotExist ) :
        LOGGER . warning ( 'EnterpriseCustomerCatalog [{enterprise_catalog_uuid}] does not exist' . format ( enterprise_catalog_uuid = enterprise_catalog_uuid , ) )
        messages . add_generic_info_message_for_error ( request )
course = None
course_run = None
course_modes = [ ]
if enterprise_catalog :
    course , course_run = enterprise_catalog . get_course_and_course_run ( course_run_id )
else :
    try :
        course , course_run = CourseCatalogApiServiceClient ( enterprise_customer . site ) . get_course_and_course_run ( course_run_id )
    except ImproperlyConfigured :
        LOGGER . warning ( 'CourseCatalogApiServiceClient is improperly configured.' )
        messages . add_generic_info_message_for_error ( request )
        return enterprise_customer , course , course_run , course_modes
if not course or not course_run :
    course_id = course [ 'key' ] if course else "Not Found"
    course_title = course [ 'title' ] if course else "Not Found"
    course_run_title = course_run [ 'title' ] if course_run else "Not Found"
    enterprise_catalog_title = enterprise_catalog . title if enterprise_catalog else "Not Found"
    # The specified course either does not exist in the specified
    # EnterpriseCustomerCatalog , or does not exist at all in the
    # discovery service .
    LOGGER . warning ( 'Failed to fetch details for course "{course_title}" [{course_id}] ' 'or course run "{course_run_title}" [{course_run_id}] ' 'for enterprise "{enterprise_name}" [{enterprise_uuid}] ' 'with catalog "{enterprise_catalog_title}" [{enterprise_catalog_uuid}]' . format ( course_title = course_title , course_id = course_id , course_run_title = course_run_title , course_run_id = course_run_id , enterprise_name = enterprise_customer . name , enterprise_uuid = enterprise_customer . uuid , enterprise_catalog_title = enterprise_catalog_title , enterprise_catalog_uuid = enterprise_catalog_uuid , ) )
    messages . add_generic_info_message_for_error ( request )
    return enterprise_customer , course , course_run , course_modes
if enterprise_catalog_uuid and not enterprise_catalog : # A catalog query parameter was given , but the specified
# EnterpriseCustomerCatalog does not exist , so just return and
# display the generic error message .
    return enterprise_customer , course , course_run , course_modes
modes = self . get_available_course_modes ( request , course_run_id , enterprise_catalog )
audit_modes = getattr ( settings , 'ENTERPRISE_COURSE_ENROLLMENT_AUDIT_MODES' , [ 'audit' , 'honor' ] )
for mode in modes :
    if mode [ 'min_price' ] :
        price_text = get_price_text ( mode [ 'min_price' ] , request )
    else :
        price_text = _ ( 'FREE' )
    if mode [ 'slug' ] in audit_modes :
        description = _ ( 'Not eligible for a certificate.' )
    else :
        description = _ ( 'Earn a verified certificate!' )
    course_modes . append ( { 'mode' : mode [ 'slug' ] , 'min_price' : mode [ 'min_price' ] , 'sku' : mode [ 'sku' ] , 'title' : mode [ 'name' ] , 'original_price' : price_text , 'final_price' : price_text , 'description' : description , 'premium' : mode [ 'slug' ] not in audit_modes } )
return enterprise_customer , course , course_run , course_modes


signature:
def get_enterprise_course_enrollment_page ( self , request , enterprise_customer , course , course_run , course_modes , enterprise_course_enrollment , data_sharing_consent ) :
    """Render enterprise - specific course track selection page ."""

implementation:
context_data = get_global_context ( request , enterprise_customer )
enterprise_catalog_uuid = request . GET . get ( 'catalog' ) if request . method == 'GET' else None
html_template_for_rendering = ENTERPRISE_GENERAL_ERROR_PAGE
if course and course_run :
    course_enrollable = True
    course_start_date = ''
    course_in_future = False
    organization_name = ''
    organization_logo = ''
    expected_learning_items = course [ 'expected_learning_items' ]
    # Parse organization name and logo .
    if course [ 'owners' ] : # The owners key contains the organizations associated with the course .
    # We pick the first one in the list here to meet UX requirements .
        organization = course [ 'owners' ] [ 0 ]
        organization_name = organization [ 'name' ]
        organization_logo = organization [ 'logo_image_url' ]
    course_title = course_run [ 'title' ]
    course_short_description = course_run [ 'short_description' ] or ''
    course_full_description = clean_html_for_template_rendering ( course_run [ 'full_description' ] or '' )
    course_pacing = self . PACING_FORMAT . get ( course_run [ 'pacing_type' ] , '' )
    if course_run [ 'start' ] :
        course_start_date = parse ( course_run [ 'start' ] ) . strftime ( '%B %d, %Y' )
        now = datetime . datetime . now ( pytz . UTC )
        course_in_future = parse ( course_run [ 'start' ] ) > now
    course_level_type = course_run . get ( 'level_type' , '' )
    staff = course_run [ 'staff' ]
    # Format the course effort string using the min / max effort fields for the course run .
    course_effort = ungettext_min_max ( '{} hour per week' , '{} hours per week' , '{}-{} hours per week' , course_run [ 'min_effort' ] or None , course_run [ 'max_effort' ] or None , ) or ''
    # Parse course run image .
    course_run_image = course_run [ 'image' ] or { }
    course_image_uri = course_run_image . get ( 'src' , '' )
    # Retrieve the enterprise - discounted price from ecommerce .
    course_modes = self . set_final_prices ( course_modes , request )
    premium_modes = [ mode for mode in course_modes if mode [ 'premium' ] ]
    # Filter audit course modes .
    course_modes = filter_audit_course_modes ( enterprise_customer , course_modes )
    # Allows automatic assignment to a cohort upon enrollment .
    cohort = request . GET . get ( 'cohort' )
    # Add a message to the message display queue if the learner
    # has gone through the data sharing consent flow and declined
    # to give data sharing consent .
    if enterprise_course_enrollment and not data_sharing_consent . granted :
        messages . add_consent_declined_message ( request , enterprise_customer , course_run . get ( 'title' , '' ) )
    if not is_course_run_enrollable ( course_run ) :
        messages . add_unenrollable_item_message ( request , 'course' )
        course_enrollable = False
    context_data . update ( { 'course_enrollable' : course_enrollable , 'course_title' : course_title , 'course_short_description' : course_short_description , 'course_pacing' : course_pacing , 'course_start_date' : course_start_date , 'course_in_future' : course_in_future , 'course_image_uri' : course_image_uri , 'course_modes' : course_modes , 'course_effort' : course_effort , 'course_full_description' : course_full_description , 'cohort' : cohort , 'organization_logo' : organization_logo , 'organization_name' : organization_name , 'course_level_type' : course_level_type , 'premium_modes' : premium_modes , 'expected_learning_items' : expected_learning_items , 'catalog' : enterprise_catalog_uuid , 'staff' : staff , 'discount_text' : _ ( 'Discount provided by {strong_start}{enterprise_customer_name}{strong_end}' ) . format ( enterprise_customer_name = enterprise_customer . name , strong_start = '<strong>' , strong_end = '</strong>' , ) , 'hide_course_original_price' : enterprise_customer . hide_course_original_price } )
    html_template_for_rendering = 'enterprise/enterprise_course_enrollment_page.html'
context_data . update ( { 'page_title' : _ ( 'Confirm your course' ) , 'confirmation_text' : _ ( 'Confirm your course' ) , 'starts_at_text' : _ ( 'Starts' ) , 'view_course_details_text' : _ ( 'View Course Details' ) , 'select_mode_text' : _ ( 'Please select one:' ) , 'price_text' : _ ( 'Price' ) , 'continue_link_text' : _ ( 'Continue' ) , 'level_text' : _ ( 'Level' ) , 'effort_text' : _ ( 'Effort' ) , 'close_modal_button_text' : _ ( 'Close' ) , 'expected_learning_items_text' : _ ( "What you'll learn" ) , 'course_full_description_text' : _ ( 'About This Course' ) , 'staff_text' : _ ( 'Course Staff' ) , } )
return render ( request , html_template_for_rendering , context = context_data )


signature:
def post ( self , request , enterprise_uuid , course_id ) :
    """Process a submitted track selection form for the enterprise ."""

implementation:
enterprise_customer , course , course_run , course_modes = self . get_base_details ( request , enterprise_uuid , course_id )
# Create a link between the user and the enterprise customer if it does not already exist .
enterprise_customer_user , __ = EnterpriseCustomerUser . objects . get_or_create ( enterprise_customer = enterprise_customer , user_id = request . user . id )
enterprise_customer_user . update_session ( request )
data_sharing_consent = DataSharingConsent . objects . proxied_get ( username = enterprise_customer_user . username , course_id = course_id , enterprise_customer = enterprise_customer )
try :
    enterprise_course_enrollment = EnterpriseCourseEnrollment . objects . get ( enterprise_customer_user__enterprise_customer = enterprise_customer , enterprise_customer_user__user_id = request . user . id , course_id = course_id )
except EnterpriseCourseEnrollment . DoesNotExist :
    enterprise_course_enrollment = None
enterprise_catalog_uuid = request . POST . get ( 'catalog' )
selected_course_mode_name = request . POST . get ( 'course_mode' )
cohort_name = request . POST . get ( 'cohort' )
selected_course_mode = None
for course_mode in course_modes :
    if course_mode [ 'mode' ] == selected_course_mode_name :
        selected_course_mode = course_mode
        break
if not selected_course_mode :
    return self . get_enterprise_course_enrollment_page ( request , enterprise_customer , course , course_run , course_modes , enterprise_course_enrollment , data_sharing_consent )
user_consent_needed = get_data_sharing_consent ( enterprise_customer_user . username , enterprise_customer . uuid , course_id = course_id ) . consent_required ( )
if not selected_course_mode . get ( 'premium' ) and not user_consent_needed : # For the audit course modes ( audit , honor ) , where DSC is not
# required , enroll the learner directly through enrollment API
# client and redirect the learner to LMS courseware page .
    if not enterprise_course_enrollment : # Create the Enterprise backend database records for this course enrollment .
        enterprise_course_enrollment = EnterpriseCourseEnrollment . objects . create ( enterprise_customer_user = enterprise_customer_user , course_id = course_id , )
        track_enrollment ( 'course-landing-page-enrollment' , request . user . id , course_id , request . get_full_path ( ) )
    client = EnrollmentApiClient ( )
    client . enroll_user_in_course ( request . user . username , course_id , selected_course_mode_name , cohort = cohort_name )
    return redirect ( LMS_COURSEWARE_URL . format ( course_id = course_id ) )
if user_consent_needed : # For the audit course modes ( audit , honor ) or for the premium
# course modes ( Verified , Prof Ed ) where DSC is required , redirect
# the learner to course specific DSC with enterprise UUID from
# there the learner will be directed to the ecommerce flow after
# providing DSC .
    query_string_params = { 'course_mode' : selected_course_mode_name , }
    if enterprise_catalog_uuid :
        query_string_params . update ( { 'catalog' : enterprise_catalog_uuid } )
    next_url = '{handle_consent_enrollment_url}?{query_string}' . format ( handle_consent_enrollment_url = reverse ( 'enterprise_handle_consent_enrollment' , args = [ enterprise_customer . uuid , course_id ] ) , query_string = urlencode ( query_string_params ) )
    failure_url = reverse ( 'enterprise_course_run_enrollment_page' , args = [ enterprise_customer . uuid , course_id ] )
    if request . META [ 'QUERY_STRING' ] : # Preserve all querystring parameters in the request to build
    # failure url , so that learner views the same enterprise course
    # enrollment page ( after redirect ) as for the first time .
    # Since this is a POST view so use ` request . META ` to get
    # querystring instead of ` request . GET ` .
    # https : / / docs . djangoproject . com / en / 1.11 / ref / request - response / # django . http . HttpRequest . META
        failure_url = '{course_enrollment_url}?{query_string}' . format ( course_enrollment_url = reverse ( 'enterprise_course_run_enrollment_page' , args = [ enterprise_customer . uuid , course_id ] ) , query_string = request . META [ 'QUERY_STRING' ] )
    return redirect ( '{grant_data_sharing_url}?{params}' . format ( grant_data_sharing_url = reverse ( 'grant_data_sharing_permissions' ) , params = urlencode ( { 'next' : next_url , 'failure_url' : failure_url , 'enterprise_customer_uuid' : enterprise_customer . uuid , 'course_id' : course_id , } ) ) )
# For the premium course modes ( Verified , Prof Ed ) where DSC is
# not required , redirect the enterprise learner to the ecommerce
# flow in LMS .
# Note : LMS start flow automatically detects the paid mode
premium_flow = LMS_START_PREMIUM_COURSE_FLOW_URL . format ( course_id = course_id )
if enterprise_catalog_uuid :
    premium_flow += '?catalog={catalog_uuid}' . format ( catalog_uuid = enterprise_catalog_uuid )
return redirect ( premium_flow )


signature:
def get ( self , request , enterprise_uuid , course_id ) :
    """Show course track selection page for the enterprise .
 Based on ` enterprise _ uuid ` in URL , the view will decide which
 enterprise customer ' s course enrollment page is to use .
 Unauthenticated learners will be redirected to enterprise - linked SSO .
 A 404 will be raised if any of the following conditions are met :
 * No enterprise customer uuid kwarg ` enterprise _ uuid ` in request .
 * No enterprise customer found against the enterprise customer
 uuid ` enterprise _ uuid ` in the request kwargs .
 * No course is found in database against the provided ` course _ id ` ."""

implementation:
# Check to see if access to the course run is restricted for this user .
embargo_url = EmbargoApiClient . redirect_if_blocked ( [ course_id ] , request . user , get_ip ( request ) , request . path )
if embargo_url :
    return redirect ( embargo_url )
enterprise_customer , course , course_run , modes = self . get_base_details ( request , enterprise_uuid , course_id )
enterprise_customer_user = get_enterprise_customer_user ( request . user . id , enterprise_uuid )
data_sharing_consent = DataSharingConsent . objects . proxied_get ( username = enterprise_customer_user . username , course_id = course_id , enterprise_customer = enterprise_customer )
enrollment_client = EnrollmentApiClient ( )
enrolled_course = enrollment_client . get_course_enrollment ( request . user . username , course_id )
try :
    enterprise_course_enrollment = EnterpriseCourseEnrollment . objects . get ( enterprise_customer_user__enterprise_customer = enterprise_customer , enterprise_customer_user__user_id = request . user . id , course_id = course_id )
except EnterpriseCourseEnrollment . DoesNotExist :
    enterprise_course_enrollment = None
if enrolled_course and enterprise_course_enrollment : # The user is already enrolled in the course through the Enterprise Customer , so redirect to the course
# info page .
    return redirect ( LMS_COURSEWARE_URL . format ( course_id = course_id ) )
return self . get_enterprise_course_enrollment_page ( request , enterprise_customer , course , course_run , modes , enterprise_course_enrollment , data_sharing_consent , )


signature:
def extend_course ( course , enterprise_customer , request ) :
    """Extend a course with more details needed for the program landing page .
 In particular , we add the following :
 * ` course _ image _ uri `
 * ` course _ title `
 * ` course _ level _ type `
 * ` course _ short _ description `
 * ` course _ full _ description `
 * ` course _ effort `
 * ` expected _ learning _ items `
 * ` staff `"""

implementation:
course_run_id = course [ 'course_runs' ] [ 0 ] [ 'key' ]
try :
    catalog_api_client = CourseCatalogApiServiceClient ( enterprise_customer . site )
except ImproperlyConfigured :
    error_code = 'ENTPEV000'
    LOGGER . error ( 'CourseCatalogApiServiceClient is improperly configured. ' 'Returned error code {error_code} to user {userid} ' 'and enterprise_customer {enterprise_customer} ' 'for course_run_id {course_run_id}' . format ( error_code = error_code , userid = request . user . id , enterprise_customer = enterprise_customer . uuid , course_run_id = course_run_id , ) )
    messages . add_generic_error_message_with_code ( request , error_code )
    return ( { } , error_code )
course_details , course_run_details = catalog_api_client . get_course_and_course_run ( course_run_id )
if not course_details or not course_run_details :
    error_code = 'ENTPEV001'
    LOGGER . error ( 'User {userid} of enterprise customer {enterprise_customer} encountered an error.' 'No course_details or course_run_details found for ' 'course_run_id {course_run_id}. ' 'The following error code reported to the user: {error_code}' . format ( userid = request . user . id , enterprise_customer = enterprise_customer . uuid , course_run_id = course_run_id , error_code = error_code , ) )
    messages . add_generic_error_message_with_code ( request , error_code )
    return ( { } , error_code )
weeks_to_complete = course_run_details [ 'weeks_to_complete' ]
course_run_image = course_run_details [ 'image' ] or { }
course . update ( { 'course_image_uri' : course_run_image . get ( 'src' , '' ) , 'course_title' : course_run_details [ 'title' ] , 'course_level_type' : course_run_details . get ( 'level_type' , '' ) , 'course_short_description' : course_run_details [ 'short_description' ] or '' , 'course_full_description' : clean_html_for_template_rendering ( course_run_details [ 'full_description' ] or '' ) , 'expected_learning_items' : course_details . get ( 'expected_learning_items' , [ ] ) , 'staff' : course_run_details . get ( 'staff' , [ ] ) , 'course_effort' : ungettext_min_max ( '{} hour per week' , '{} hours per week' , '{}-{} hours per week' , course_run_details [ 'min_effort' ] or None , course_run_details [ 'max_effort' ] or None , ) or '' , 'weeks_to_complete' : ungettext ( '{} week' , '{} weeks' , weeks_to_complete ) . format ( weeks_to_complete ) if weeks_to_complete else '' , } )
return course , None


signature:
def get_program_details ( self , request , program_uuid , enterprise_customer ) :
    """Retrieve fundamental details used by both POST and GET versions of this view .
 Specifically :
 * Take the program UUID and get specific details about the program .
 * Determine whether the learner is enrolled in the program .
 * Determine whether the learner is certificate eligible for the program ."""

implementation:
try :
    course_catalog_api_client = CourseCatalogApiServiceClient ( enterprise_customer . site )
except ImproperlyConfigured :
    error_code = 'ENTPEV002'
    LOGGER . error ( 'CourseCatalogApiServiceClient is improperly configured. ' 'Returned error code {error_code} to user {userid} ' 'and enterprise_customer {enterprise_customer} ' 'for program {program_uuid}' . format ( error_code = error_code , userid = request . user . id , enterprise_customer = enterprise_customer . uuid , program_uuid = program_uuid , ) )
    messages . add_generic_error_message_with_code ( request , error_code )
    return ( { } , error_code )
program_details = course_catalog_api_client . get_program_by_uuid ( program_uuid )
if program_details is None :
    error_code = 'ENTPEV003'
    LOGGER . error ( 'User {userid} of enterprise customer {enterprise_customer} encountered an error. ' 'program_details is None for program_uuid {program_uuid}. ' 'Returned error code {error_code} to user' . format ( userid = request . user . id , enterprise_customer = enterprise_customer . uuid , program_uuid = program_uuid , error_code = error_code , ) )
    messages . add_generic_error_message_with_code ( request , error_code )
    return ( { } , error_code )
program_type = course_catalog_api_client . get_program_type_by_slug ( slugify ( program_details [ 'type' ] ) )
if program_type is None :
    error_code = 'ENTPEV004'
    LOGGER . error ( 'User {userid} of enterprise customer {enterprise_customer} encountered an error. ' 'program_type is None for program_details of program_uuid {program_uuid}. ' 'Returned error code {error_code} to user' . format ( userid = request . user . id , enterprise_customer = enterprise_customer . uuid , program_uuid = program_uuid , error_code = error_code , ) )
    messages . add_generic_error_message_with_code ( request , error_code )
    return ( { } , error_code )
# Extend our program details with context we ' ll need for display or for deciding redirects .
program_details = ProgramDataExtender ( program_details , request . user ) . extend ( )
# TODO : Upstream this additional context to the platform ' s ` ProgramDataExtender ` so we can avoid this here .
program_details [ 'enrolled_in_program' ] = False
enrollment_count = 0
for extended_course in program_details [ 'courses' ] : # We need to extend our course data further for modals and other displays .
    extended_data , error_code = ProgramEnrollmentView . extend_course ( extended_course , enterprise_customer , request )
    if error_code :
        return ( { } , error_code )
    extended_course . update ( extended_data )
    # We ' re enrolled in the program if we have certificate - eligible enrollment in even 1 of its courses .
    extended_course_run = extended_course [ 'course_runs' ] [ 0 ]
    if extended_course_run [ 'is_enrolled' ] and extended_course_run [ 'upgrade_url' ] is None :
        program_details [ 'enrolled_in_program' ] = True
        enrollment_count += 1
# We ' re certificate eligible for the program if we have certificate - eligible enrollment in all of its courses .
program_details [ 'certificate_eligible_for_program' ] = ( enrollment_count == len ( program_details [ 'courses' ] ) )
program_details [ 'type_details' ] = program_type
return program_details , None