signature:
def train_model ( best_processed_path , weight_path = '../weight/model_weight.h5' , verbose = 2 ) :
    """Given path to processed BEST dataset ,
 train CNN model for words beginning alongside with
 character label encoder and character type label encoder
 Input
 best _ processed _ path : str , path to processed BEST dataset
 weight _ path : str , path to weight path file
 verbose : int , verbost option for training Keras model
 Output
 model : keras model , keras model for tokenize prediction"""

implementation:
x_train_char , x_train_type , y_train = prepare_feature ( best_processed_path , option = 'train' )
x_test_char , x_test_type , y_test = prepare_feature ( best_processed_path , option = 'test' )
validation_set = False
if os . path . isdir ( os . path . join ( best_processed_path , 'val' ) ) :
    validation_set = True
    x_val_char , x_val_type , y_val = prepare_feature ( best_processed_path , option = 'val' )
if not os . path . isdir ( os . path . dirname ( weight_path ) ) :
    os . makedirs ( os . path . dirname ( weight_path ) )
# make directory if weight does not exist
callbacks_list = [ ReduceLROnPlateau ( ) , ModelCheckpoint ( weight_path , save_best_only = True , save_weights_only = True , monitor = 'val_loss' , mode = 'min' , verbose = 1 ) ]
# train model
model = get_convo_nn2 ( )
train_params = [ ( 10 , 256 ) , ( 3 , 512 ) , ( 3 , 2048 ) , ( 3 , 4096 ) , ( 3 , 8192 ) ]
for ( epochs , batch_size ) in train_params :
    print ( "train with {} epochs and {} batch size" . format ( epochs , batch_size ) )
    if validation_set :
        model . fit ( [ x_train_char , x_train_type ] , y_train , epochs = epochs , batch_size = batch_size , verbose = verbose , callbacks = callbacks_list , validation_data = ( [ x_val_char , x_val_type ] , y_val ) )
    else :
        model . fit ( [ x_train_char , x_train_type ] , y_train , epochs = epochs , batch_size = batch_size , verbose = verbose , callbacks = callbacks_list )
return model


signature:
def evaluate ( best_processed_path , model ) :
    """Evaluate model on splitted 10 percent testing set"""

implementation:
x_test_char , x_test_type , y_test = prepare_feature ( best_processed_path , option = 'test' )
y_predict = model . predict ( [ x_test_char , x_test_type ] )
y_predict = ( y_predict . ravel ( ) > 0.5 ) . astype ( int )
f1score = f1_score ( y_test , y_predict )
precision = precision_score ( y_test , y_predict )
recall = recall_score ( y_test , y_predict )
return f1score , precision , recall


signature:
def tokenize ( text , custom_dict = None ) :
    """Tokenize given Thai text string
 Input
 text : str , Thai text string
 custom _ dict : str ( or list ) , path to customized dictionary file
 It allows the function not to tokenize given dictionary wrongly .
 The file should contain custom words separated by line .
 Alternatively , you can provide list of custom words too .
 Output
 tokens : list , list of tokenized words
 Example
 > > deepcut . tokenize ( ' ตัดคําได้ดีมาก ' )
 > > [ ' ตัดคํา ' , ' ได้ ' , ' ดี ' , ' มาก ' ]"""

implementation:
global TOKENIZER
if not TOKENIZER :
    TOKENIZER = DeepcutTokenizer ( )
return TOKENIZER . tokenize ( text , custom_dict = custom_dict )


signature:
def _document_frequency ( X ) :
    """Count the number of non - zero values for each feature in sparse X ."""

implementation:
if sp . isspmatrix_csr ( X ) :
    return np . bincount ( X . indices , minlength = X . shape [ 1 ] )
return np . diff ( sp . csc_matrix ( X , copy = False ) . indptr )


signature:
def _check_stop_list ( stop ) :
    """Check stop words list
 ref : https : / / github . com / scikit - learn / scikit - learn / blob / master / sklearn / feature _ extraction / text . py # L87 - L95"""

implementation:
if stop == "thai" :
    return THAI_STOP_WORDS
elif isinstance ( stop , six . string_types ) :
    raise ValueError ( "not a built-in stop list: %s" % stop )
elif stop is None :
    return None
# assume it ' s a collection
return frozenset ( stop )


signature:
def _word_ngrams ( self , tokens ) :
    """Turn tokens into a tokens of n - grams
 ref : https : / / github . com / scikit - learn / scikit - learn / blob / ef5cb84a / sklearn / feature _ extraction / text . py # L124 - L153"""

implementation:
# handle stop words
if self . stop_words is not None :
    tokens = [ w for w in tokens if w not in self . stop_words ]
# handle token n - grams
min_n , max_n = self . ngram_range
if max_n != 1 :
    original_tokens = tokens
    if min_n == 1 : # no need to do any slicing for unigrams
    # just iterate through the original tokens
        tokens = list ( original_tokens )
        min_n += 1
    else :
        tokens = [ ]
    n_original_tokens = len ( original_tokens )
    # bind method outside of loop to reduce overhead
    tokens_append = tokens . append
    space_join = " " . join
    for n in range ( min_n , min ( max_n + 1 , n_original_tokens + 1 ) ) :
        for i in range ( n_original_tokens - n + 1 ) :
            tokens_append ( space_join ( original_tokens [ i : i + n ] ) )
return tokens


signature:
def fit_tranform ( self , raw_documents ) :
    """Transform given list of raw _ documents to document - term matrix in
 sparse CSR format ( see scipy )"""

implementation:
X = self . transform ( raw_documents , new_document = True )
return X


signature:
def create_feature_array ( text , n_pad = 21 ) :
    """Create feature array of character and surrounding characters"""

implementation:
n = len ( text )
n_pad_2 = int ( ( n_pad - 1 ) / 2 )
text_pad = [ ' ' ] * n_pad_2 + [ t for t in text ] + [ ' ' ] * n_pad_2
x_char , x_type = [ ] , [ ]
for i in range ( n_pad_2 , n_pad_2 + n ) :
    char_list = text_pad [ i + 1 : i + n_pad_2 + 1 ] + list ( reversed ( text_pad [ i - n_pad_2 : i ] ) ) + [ text_pad [ i ] ]
    char_map = [ CHARS_MAP . get ( c , 80 ) for c in char_list ]
    char_type = [ CHAR_TYPES_MAP . get ( CHAR_TYPE_FLATTEN . get ( c , 'o' ) , 4 ) for c in char_list ]
    x_char . append ( char_map )
    x_type . append ( char_type )
x_char = np . array ( x_char ) . astype ( float )
x_type = np . array ( x_type ) . astype ( float )
return x_char , x_type


signature:
def create_n_gram_df ( df , n_pad ) :
    """Given input dataframe , create feature dataframe of shifted characters"""

implementation:
n_pad_2 = int ( ( n_pad - 1 ) / 2 )
for i in range ( n_pad_2 ) :
    df [ 'char-{}' . format ( i + 1 ) ] = df [ 'char' ] . shift ( i + 1 )
    df [ 'type-{}' . format ( i + 1 ) ] = df [ 'type' ] . shift ( i + 1 )
    df [ 'char{}' . format ( i + 1 ) ] = df [ 'char' ] . shift ( - i - 1 )
    df [ 'type{}' . format ( i + 1 ) ] = df [ 'type' ] . shift ( - i - 1 )
return df [ n_pad_2 : - n_pad_2 ]


signature:
def _fetch_course_enrollment_data ( self , enterprise_customer_uuid ) :
    """Return enterprise customer UUID / user _ id / course _ run _ id triples which represent CourseEnrollment records
 which do not have a matching EnterpriseCourseEnrollment record .
 The query used below looks for CourseEnrollment records that are associated with enterprise
 learners where the enrollment data is after the creation of the link between the learner
 and the enterprise . It also excludes learners with edx . org email addresses in order to
 filter out test users ."""

implementation:
query = '''
            SELECT
                au.id as user_id,
                ecu.enterprise_customer_id as enterprise_customer_uuid,
                sce.course_id as course_run_id
            FROM student_courseenrollment sce
            JOIN auth_user au
                ON au.id = sce.user_id
            JOIN enterprise_enterprisecustomeruser ecu
                ON ecu.user_id = au.id
            LEFT JOIN enterprise_enterprisecourseenrollment ece
                ON ece.enterprise_customer_user_id = ecu.id
                AND ece.course_id = sce.course_id
            WHERE
                ece.id IS NULL
                AND ecu.created <= sce.created
                AND au.email NOT LIKE '%@edx.org'
                {enterprise_customer_filter}
            ORDER BY sce.created;
        '''
with connection . cursor ( ) as cursor :
    if enterprise_customer_uuid :
        cursor . execute ( query . format ( enterprise_customer_filter = 'AND ecu.enterprise_customer_id = %s' ) , [ enterprise_customer_uuid ] )
    else :
        cursor . execute ( query . format ( enterprise_customer_filter = '' ) )
    return self . _dictfetchall ( cursor )


signature:
def _dictfetchall ( self , cursor ) :
    """Return all rows from a cursor as a dict ."""

implementation:
columns = [ col [ 0 ] for col in cursor . description ]
return [ dict ( zip ( columns , row ) ) for row in cursor . fetchall ( ) ]


signature:
def parse_lms_api_datetime ( datetime_string , datetime_format = LMS_API_DATETIME_FORMAT ) :
    """Parse a received datetime into a timezone - aware , Python datetime object .
 Arguments :
 datetime _ string : A string to be parsed .
 datetime _ format : A datetime format string to be used for parsing"""

implementation:
if isinstance ( datetime_string , datetime . datetime ) :
    date_time = datetime_string
else :
    try :
        date_time = datetime . datetime . strptime ( datetime_string , datetime_format )
    except ValueError :
        date_time = datetime . datetime . strptime ( datetime_string , LMS_API_DATETIME_FORMAT_WITHOUT_TIMEZONE )
# If the datetime format didn ' t include a timezone , then set to UTC .
# Note that if we ' re using the default LMS _ API _ DATETIME _ FORMAT , it ends in ' Z ' ,
# which denotes UTC for ISO - 8661.
if date_time . tzinfo is None :
    date_time = date_time . replace ( tzinfo = timezone . utc )
return date_time


signature:
def connect ( self ) :
    """Connect to the REST API , authenticating with a JWT for the current user ."""

implementation:
if JwtBuilder is None :
    raise NotConnectedToOpenEdX ( "This package must be installed in an OpenEdX environment." )
now = int ( time ( ) )
jwt = JwtBuilder . create_jwt_for_user ( self . user )
self . client = EdxRestApiClient ( self . API_BASE_URL , append_slash = self . APPEND_SLASH , jwt = jwt , )
self . expires_at = now + self . expires_in


signature:
def redirect_if_blocked ( course_run_ids , user = None , ip_address = None , url = None ) :
    """Return redirect to embargo error page if the given user is blocked ."""

implementation:
for course_run_id in course_run_ids :
    redirect_url = embargo_api . redirect_if_blocked ( CourseKey . from_string ( course_run_id ) , user = user , ip_address = ip_address , url = url )
    if redirect_url :
        return redirect_url


signature:
def get_course_details ( self , course_id ) :
    """Query the Enrollment API for the course details of the given course _ id .
 Args :
 course _ id ( str ) : The string value of the course ' s unique identifier
 Returns :
 dict : A dictionary containing details about the course , in an enrollment context ( allowed modes , etc . )"""

implementation:
try :
    return self . client . course ( course_id ) . get ( )
except ( SlumberBaseException , ConnectionError , Timeout ) as exc :
    LOGGER . exception ( 'Failed to retrieve course enrollment details for course [%s] due to: [%s]' , course_id , str ( exc ) )
    return { }


signature:
def get_course_modes ( self , course_id ) :
    """Query the Enrollment API for the specific course modes that are available for the given course _ id .
 Arguments :
 course _ id ( str ) : The string value of the course ' s unique identifier
 Returns :
 list : A list of course mode dictionaries ."""

implementation:
details = self . get_course_details ( course_id )
modes = details . get ( 'course_modes' , [ ] )
return self . _sort_course_modes ( [ mode for mode in modes if mode [ 'slug' ] not in EXCLUDED_COURSE_MODES ] )


signature:
def has_course_mode ( self , course_run_id , mode ) :
    """Query the Enrollment API to see whether a course run has a given course mode available .
 Arguments :
 course _ run _ id ( str ) : The string value of the course run ' s unique identifier
 Returns :
 bool : Whether the course run has the given mode avaialble for enrollment ."""

implementation:
course_modes = self . get_course_modes ( course_run_id )
return any ( course_mode for course_mode in course_modes if course_mode [ 'slug' ] == mode )


signature:
def enroll_user_in_course ( self , username , course_id , mode , cohort = None ) :
    """Call the enrollment API to enroll the user in the course specified by course _ id .
 Args :
 username ( str ) : The username by which the user goes on the OpenEdX platform
 course _ id ( str ) : The string value of the course ' s unique identifier
 mode ( str ) : The enrollment mode which should be used for the enrollment
 cohort ( str ) : Add the user to this named cohort
 Returns :
 dict : A dictionary containing details of the enrollment , including course details , mode , username , etc ."""

implementation: return self . client . enrollment . post ( { 'user' : username , 'course_details' : { 'course_id' : course_id } , 'mode' : mode , 'cohort' : cohort , } )


signature:
def unenroll_user_from_course ( self , username , course_id ) :
    """Call the enrollment API to unenroll the user in the course specified by course _ id .
 Args :
 username ( str ) : The username by which the user goes on the OpenEdx platform
 course _ id ( str ) : The string value of the course ' s unique identifier
 Returns :
 bool : Whether the unenrollment succeeded"""

implementation:
enrollment = self . get_course_enrollment ( username , course_id )
if enrollment and enrollment [ 'is_active' ] :
    response = self . client . enrollment . post ( { 'user' : username , 'course_details' : { 'course_id' : course_id } , 'is_active' : False , 'mode' : enrollment [ 'mode' ] } )
    return not response [ 'is_active' ]
return False


signature:
def get_course_enrollment ( self , username , course_id ) :
    """Query the enrollment API to get information about a single course enrollment .
 Args :
 username ( str ) : The username by which the user goes on the OpenEdX platform
 course _ id ( str ) : The string value of the course ' s unique identifier
 Returns :
 dict : A dictionary containing details of the enrollment , including course details , mode , username , etc ."""

implementation:
endpoint = getattr ( self . client . enrollment , '{username},{course_id}' . format ( username = username , course_id = course_id ) )
try :
    result = endpoint . get ( )
except HttpNotFoundError : # This enrollment data endpoint returns a 404 if either the username or course _ id specified isn ' t valid
    LOGGER . error ( 'Course enrollment details not found for invalid username or course; username=[%s], course=[%s]' , username , course_id )
    return None
# This enrollment data endpoint returns an empty string if the username and course _ id is valid , but there ' s
# no matching enrollment found
if not result :
    LOGGER . info ( 'Failed to find course enrollment details for user [%s] and course [%s]' , username , course_id )
    return None
return result


signature:
def is_enrolled ( self , username , course_run_id ) :
    """Query the enrollment API and determine if a learner is enrolled in a course run .
 Args :
 username ( str ) : The username by which the user goes on the OpenEdX platform
 course _ run _ id ( str ) : The string value of the course ' s unique identifier
 Returns :
 bool : Indicating whether the user is enrolled in the course run . Returns False under any errors ."""

implementation:
enrollment = self . get_course_enrollment ( username , course_run_id )
return enrollment is not None and enrollment . get ( 'is_active' , False )


signature:
def _get_results ( self , identity_provider , param_name , param_value , result_field_name ) :
    """Calls the third party auth api endpoint to get the mapping between usernames and remote ids ."""

implementation:
try :
    kwargs = { param_name : param_value }
    returned = self . client . providers ( identity_provider ) . users . get ( ** kwargs )
    results = returned . get ( 'results' , [ ] )
except HttpNotFoundError :
    LOGGER . error ( 'username not found for third party provider={provider}, {querystring_param}={id}' . format ( provider = identity_provider , querystring_param = param_name , id = param_value ) )
    results = [ ]
for row in results :
    if row . get ( param_name ) == param_value :
        return row . get ( result_field_name )
return None


signature:
def get_course_grade ( self , course_id , username ) :
    """Retrieve the grade for the given username for the given course _ id .
 Args :
 * ` ` course _ id ` ` ( str ) : The string value of the course ' s unique identifier
 * ` ` username ` ` ( str ) : The username ID identifying the user for which to retrieve the grade .
 Raises :
 HttpNotFoundError if no grade found for the given user + course .
 Returns :
 a dict containing :
 * ` ` username ` ` : A string representation of a user ' s username passed in the request .
 * ` ` course _ key ` ` : A string representation of a Course ID .
 * ` ` passed ` ` : Boolean representing whether the course has been passed according the course ' s grading policy .
 * ` ` percent ` ` : A float representing the overall grade for the course
 * ` ` letter _ grade ` ` : A letter grade as defined in grading _ policy ( e . g . ' A ' ' B ' ' C ' for 6.002x ) or None"""

implementation:
results = self . client . courses ( course_id ) . get ( username = username )
for row in results :
    if row . get ( 'username' ) == username :
        return row
raise HttpNotFoundError ( 'No grade record found for course={}, username={}' . format ( course_id , username ) )


signature:
def get_course_certificate ( self , course_id , username ) :
    """Retrieve the certificate for the given username for the given course _ id .
 Args :
 * ` ` course _ id ` ` ( str ) : The string value of the course ' s unique identifier
 * ` ` username ` ` ( str ) : The username ID identifying the user for which to retrieve the certificate
 Raises :
 HttpNotFoundError if no certificate found for the given user + course .
 Returns :
 a dict containing :
 * ` ` username ` ` : A string representation of an user ' s username passed in the request .
 * ` ` course _ id ` ` : A string representation of a Course ID .
 * ` ` certificate _ type ` ` : A string representation of the certificate type .
 * ` ` created _ date ` : Datetime the certificate was created ( tz - aware ) .
 * ` ` status ` ` : A string representation of the certificate status .
 * ` ` is _ passing ` ` : True if the certificate has a passing status , False if not .
 * ` ` download _ url ` ` : A string representation of the certificate url .
 * ` ` grade ` ` : A string representation of a float for the user ' s course grade ."""

implementation: return self . client . certificates ( username ) . courses ( course_id ) . get ( )


signature:
def course_discovery_api_client ( user , catalog_url ) :
    """Return a Course Discovery API client setup with authentication for the specified user ."""

implementation:
if JwtBuilder is None :
    raise NotConnectedToOpenEdX ( _ ( "To get a Catalog API client, this package must be " "installed in an Open edX environment." ) )
jwt = JwtBuilder . create_jwt_for_user ( user )
return EdxRestApiClient ( catalog_url , jwt = jwt )


signature:
def traverse_pagination ( response , endpoint , content_filter_query , query_params ) :
    """Traverse a paginated API response and extracts and concatenates " results " returned by API .
 Arguments :
 response ( dict ) : API response object .
 endpoint ( Slumber . Resource ) : API endpoint object .
 content _ filter _ query ( dict ) : query parameters used to filter catalog results .
 query _ params ( dict ) : query parameters used to paginate results .
 Returns :
 list : all the results returned by the API ."""

implementation:
results = response . get ( 'results' , [ ] )
page = 1
while response . get ( 'next' ) :
    page += 1
    response = endpoint ( ) . post ( content_filter_query , ** dict ( query_params , page = page ) )
    results += response . get ( 'results' , [ ] )
return results


signature:
def get_catalog_results ( self , content_filter_query , query_params = None , traverse_pagination = False ) :
    """Return results from the discovery service ' s search / all endpoint .
 Arguments :
 content _ filter _ query ( dict ) : query parameters used to filter catalog results .
 query _ params ( dict ) : query parameters used to paginate results .
 traverse _ pagination ( bool ) : True to return all results , False to return the paginated response .
 Defaults to False .
 Returns :
 dict : Paginated response or all the records ."""

implementation:
query_params = query_params or { }
try :
    endpoint = getattr ( self . client , self . SEARCH_ALL_ENDPOINT )
    response = endpoint ( ) . post ( data = content_filter_query , ** query_params )
    if traverse_pagination :
        response [ 'results' ] = self . traverse_pagination ( response , endpoint , content_filter_query , query_params )
        response [ 'next' ] = response [ 'previous' ] = None
except Exception as ex : # pylint : disable = broad - except
    LOGGER . exception ( 'Attempted to call course-discovery search/all/ endpoint with the following parameters: ' 'content_filter_query: %s, query_params: %s, traverse_pagination: %s. ' 'Failed to retrieve data from the catalog API. content -- [%s]' , content_filter_query , query_params , traverse_pagination , getattr ( ex , 'content' , '' ) )
    # We need to bubble up failures when we encounter them instead of masking them !
    raise ex
return response


signature:
def get_catalog ( self , catalog_id ) :
    """Return specified course catalog .
 Returns :
 dict : catalog details if it is available for the user ."""

implementation: return self . _load_data ( self . CATALOGS_ENDPOINT , default = [ ] , resource_id = catalog_id )


signature:
def get_paginated_catalog_courses ( self , catalog_id , querystring = None ) :
    """Return paginated response for all catalog courses .
 Returns :
 dict : API response with links to next and previous pages ."""

implementation: return self . _load_data ( self . CATALOGS_COURSES_ENDPOINT . format ( catalog_id ) , default = [ ] , querystring = querystring , traverse_pagination = False , many = False , )


signature:
def get_paginated_catalogs ( self , querystring = None ) :
    """Return a paginated list of course catalogs , including name and ID .
 Returns :
 dict : Paginated response containing catalogs available for the user ."""

implementation: return self . _load_data ( self . CATALOGS_ENDPOINT , default = [ ] , querystring = querystring , traverse_pagination = False , many = False )


signature:
def get_catalog_courses ( self , catalog_id ) :
    """Return the courses included in a single course catalog by ID .
 Args :
 catalog _ id ( int ) : The catalog ID we want to retrieve .
 Returns :
 list : Courses of the catalog in question"""

implementation: return self . _load_data ( self . CATALOGS_COURSES_ENDPOINT . format ( catalog_id ) , default = [ ] )


signature:
def get_course_and_course_run ( self , course_run_id ) :
    """Return the course and course run metadata for the given course run ID .
 Arguments :
 course _ run _ id ( str ) : The course run ID .
 Returns :
 tuple : The course metadata and the course run metadata ."""

implementation:
# Parse the course ID from the course run ID .
course_id = parse_course_key ( course_run_id )
# Retrieve the course metadata from the catalog service .
course = self . get_course_details ( course_id )
course_run = None
if course : # Find the specified course run .
    course_run = None
    course_runs = [ course_run for course_run in course [ 'course_runs' ] if course_run [ 'key' ] == course_run_id ]
    if course_runs :
        course_run = course_runs [ 0 ]
return course , course_run


signature:
def get_course_details ( self , course_id ) :
    """Return the details of a single course by id - not a course run id .
 Args :
 course _ id ( str ) : The unique id for the course in question .
 Returns :
 dict : Details of the course in question ."""

implementation: return self . _load_data ( self . COURSES_ENDPOINT , resource_id = course_id , many = False )


signature:
def get_program_by_title ( self , program_title ) :
    """Return single program by name , or None if not found .
 Arguments :
 program _ title ( string ) : Program title as seen by students and in Course Catalog Admin
 Returns :
 dict : Program data provided by Course Catalog API"""

implementation:
all_programs = self . _load_data ( self . PROGRAMS_ENDPOINT , default = [ ] )
matching_programs = [ program for program in all_programs if program . get ( 'title' ) == program_title ]
if len ( matching_programs ) > 1 :
    raise MultipleProgramMatchError ( len ( matching_programs ) )
elif len ( matching_programs ) == 1 :
    return matching_programs [ 0 ]
else :
    return None


signature:
def get_program_by_uuid ( self , program_uuid ) :
    """Return single program by UUID , or None if not found .
 Arguments :
 program _ uuid ( string ) : Program UUID in string form
 Returns :
 dict : Program data provided by Course Catalog API"""

implementation: return self . _load_data ( self . PROGRAMS_ENDPOINT , resource_id = program_uuid , default = None )


signature:
def get_program_course_keys ( self , program_uuid ) :
    """Get a list of the course IDs ( not course run IDs ) contained in the program .
 Arguments :
 program _ uuid ( str ) : Program UUID in string form
 Returns :
 list ( str ) : List of course keys in string form that are included in the program"""

implementation:
program_details = self . get_program_by_uuid ( program_uuid )
if not program_details :
    return [ ]
return [ course [ 'key' ] for course in program_details . get ( 'courses' , [ ] ) ]


signature:
def get_program_type_by_slug ( self , slug ) :
    """Get a program type by its slug .
 Arguments :
 slug ( str ) : The slug to identify the program type .
 Returns :
 dict : A program type object ."""

implementation: return self . _load_data ( self . PROGRAM_TYPES_ENDPOINT , resource_id = slug , default = None , )


signature:
def get_common_course_modes ( self , course_run_ids ) :
    """Find common course modes for a set of course runs .
 This function essentially returns an intersection of types of seats available
 for each course run .
 Arguments :
 course _ run _ ids ( Iterable [ str ] ) : Target Course run IDs .
 Returns :
 set : course modes found in all given course runs
 Examples :
 # run1 has prof and audit , run 2 has the same
 get _ common _ course _ modes ( [ ' course - v1 : run1 ' , ' course - v1 : run2 ' ] )
 { ' prof ' , ' audit ' }
 # run1 has prof and audit , run 2 has only prof
 get _ common _ course _ modes ( [ ' course - v1 : run1 ' , ' course - v1 : run2 ' ] )
 { ' prof ' }
 # run1 has prof and audit , run 2 honor
 get _ common _ course _ modes ( [ ' course - v1 : run1 ' , ' course - v1 : run2 ' ] )
 # run1 has nothing , run2 has prof
 get _ common _ course _ modes ( [ ' course - v1 : run1 ' , ' course - v1 : run2 ' ] )
 # run1 has prof and audit , run 2 prof , run3 has audit
 get _ common _ course _ modes ( [ ' course - v1 : run1 ' , ' course - v1 : run2 ' , ' course - v1 : run3 ' ] )
 # run1 has nothing , run 2 prof , run3 has prof
 get _ common _ course _ modes ( [ ' course - v1 : run1 ' , ' course - v1 : run2 ' , ' course - v1 : run3 ' ] )"""

implementation:
available_course_modes = None
for course_run_id in course_run_ids :
    course_run = self . get_course_run ( course_run_id ) or { }
    course_run_modes = { seat . get ( 'type' ) for seat in course_run . get ( 'seats' , [ ] ) }
    if available_course_modes is None :
        available_course_modes = course_run_modes
    else :
        available_course_modes &= course_run_modes
    if not available_course_modes :
        return available_course_modes
return available_course_modes


signature:
def is_course_in_catalog ( self , catalog_id , course_id ) :
    """Determine if the given course or course run ID is contained in the catalog with the given ID .
 Args :
 catalog _ id ( int ) : The ID of the catalog
 course _ id ( str ) : The ID of the course or course run
 Returns :
 bool : Whether the course or course run is contained in the given catalog"""

implementation:
try : # Determine if we have a course run ID , rather than a plain course ID
    course_run_id = str ( CourseKey . from_string ( course_id ) )
except InvalidKeyError :
    course_run_id = None
endpoint = self . client . catalogs ( catalog_id ) . contains
if course_run_id :
    resp = endpoint . get ( course_run_id = course_run_id )
else :
    resp = endpoint . get ( course_id = course_id )
return resp . get ( 'courses' , { } ) . get ( course_id , False )


signature:
def _load_data ( self , resource , default = DEFAULT_VALUE_SAFEGUARD , ** kwargs ) :
    """Load data from API client .
 Arguments :
 resource ( string ) : type of resource to load
 default ( any ) : value to return if API query returned empty result . Sensible values : [ ] , { } , None etc .
 Returns :
 dict : Deserialized response from Course Catalog API"""

implementation:
default_val = default if default != self . DEFAULT_VALUE_SAFEGUARD else { }
try :
    return get_edx_api_data ( api_config = CatalogIntegration . current ( ) , resource = resource , api = self . client , ** kwargs ) or default_val
except ( SlumberBaseException , ConnectionError , Timeout ) as exc :
    LOGGER . exception ( 'Failed to load data from resource [%s] with kwargs [%s] due to: [%s]' , resource , kwargs , str ( exc ) )
    return default_val


signature:
def get_content_metadata ( self , enterprise_customer ) :
    """Return all content metadata contained in the catalogs associated with the EnterpriseCustomer .
 Arguments :
 enterprise _ customer ( EnterpriseCustomer ) : The EnterpriseCustomer to return content metadata for .
 Returns :
 list : List of dicts containing content metadata ."""

implementation:
content_metadata = OrderedDict ( )
# TODO : This if block can be removed when we get rid of discovery service - based catalogs .
if enterprise_customer . catalog :
    response = self . _load_data ( self . ENTERPRISE_CUSTOMER_ENDPOINT , detail_resource = 'courses' , resource_id = str ( enterprise_customer . uuid ) , traverse_pagination = True , )
    for course in response [ 'results' ] :
        for course_run in course [ 'course_runs' ] :
            course_run [ 'content_type' ] = 'courserun'
            # Make this look like a search endpoint result .
            content_metadata [ course_run [ 'key' ] ] = course_run
for enterprise_customer_catalog in enterprise_customer . enterprise_customer_catalogs . all ( ) :
    response = self . _load_data ( self . ENTERPRISE_CUSTOMER_CATALOGS_ENDPOINT , resource_id = str ( enterprise_customer_catalog . uuid ) , traverse_pagination = True , querystring = { 'page_size' : 1000 } , )
    for item in response [ 'results' ] :
        content_id = utils . get_content_metadata_item_id ( item )
        content_metadata [ content_id ] = item
return content_metadata . values ( )


signature:
def _load_data ( self , resource , detail_resource = None , resource_id = None , querystring = None , traverse_pagination = False , default = DEFAULT_VALUE_SAFEGUARD , ) :
    """Loads a response from a call to one of the Enterprise endpoints .
 : param resource : The endpoint resource name .
 : param detail _ resource : The sub - resource to append to the path .
 : param resource _ id : The resource ID for the specific detail to get from the endpoint .
 : param querystring : Optional query string parameters .
 : param traverse _ pagination : Whether to traverse pagination or return paginated response .
 : param default : The default value to return in case of no response content .
 : return : Data returned by the API ."""

implementation:
default_val = default if default != self . DEFAULT_VALUE_SAFEGUARD else { }
querystring = querystring if querystring else { }
cache_key = utils . get_cache_key ( resource = resource , querystring = querystring , traverse_pagination = traverse_pagination , resource_id = resource_id )
response = cache . get ( cache_key )
if not response : # Response is not cached , so make a call .
    endpoint = getattr ( self . client , resource ) ( resource_id )
    endpoint = getattr ( endpoint , detail_resource ) if detail_resource else endpoint
    response = endpoint . get ( ** querystring )
    if traverse_pagination :
        results = utils . traverse_pagination ( response , endpoint )
        response = { 'count' : len ( results ) , 'next' : 'None' , 'previous' : 'None' , 'results' : results , }
    if response : # Now that we ' ve got a response , cache it .
        cache . set ( cache_key , response , settings . ENTERPRISE_API_CACHE_TIMEOUT )
return response or default_val


signature:
def transmit ( self , payload , ** kwargs ) :
    """Transmit content metadata items to the integrated channel ."""

implementation:
items_to_create , items_to_update , items_to_delete , transmission_map = self . _partition_items ( payload )
self . _transmit_delete ( items_to_delete )
self . _transmit_create ( items_to_create )
self . _transmit_update ( items_to_update , transmission_map )


signature:
def _partition_items ( self , channel_metadata_item_map ) :
    """Return items that need to be created , updated , and deleted along with the
 current ContentMetadataItemTransmissions ."""

implementation:
items_to_create = { }
items_to_update = { }
items_to_delete = { }
transmission_map = { }
export_content_ids = channel_metadata_item_map . keys ( )
# Get the items that were previously transmitted to the integrated channel .
# If we are not transmitting something that was previously transmitted ,
# we need to delete it from the integrated channel .
for transmission in self . _get_transmissions ( ) :
    transmission_map [ transmission . content_id ] = transmission
    if transmission . content_id not in export_content_ids :
        items_to_delete [ transmission . content_id ] = transmission . channel_metadata
# Compare what is currently being transmitted to what was transmitted
# previously , identifying items that need to be created or updated .
for item in channel_metadata_item_map . values ( ) :
    content_id = item . content_id
    channel_metadata = item . channel_metadata
    transmitted_item = transmission_map . get ( content_id , None )
    if transmitted_item is not None :
        if diff ( channel_metadata , transmitted_item . channel_metadata ) :
            items_to_update [ content_id ] = channel_metadata
    else :
        items_to_create [ content_id ] = channel_metadata
LOGGER . info ( 'Preparing to transmit creation of [%s] content metadata items with plugin configuration [%s]: [%s]' , len ( items_to_create ) , self . enterprise_configuration , items_to_create . keys ( ) , )
LOGGER . info ( 'Preparing to transmit update of [%s] content metadata items with plugin configuration [%s]: [%s]' , len ( items_to_update ) , self . enterprise_configuration , items_to_update . keys ( ) , )
LOGGER . info ( 'Preparing to transmit deletion of [%s] content metadata items with plugin configuration [%s]: [%s]' , len ( items_to_delete ) , self . enterprise_configuration , items_to_delete . keys ( ) , )
return items_to_create , items_to_update , items_to_delete , transmission_map


signature:
def _serialize_items ( self , channel_metadata_items ) :
    """Serialize content metadata items for a create transmission to the integrated channel ."""

implementation: return json . dumps ( self . _prepare_items_for_transmission ( channel_metadata_items ) , sort_keys = True ) . encode ( 'utf-8' )


signature:
def _transmit_create ( self , channel_metadata_item_map ) :
    """Transmit content metadata creation to integrated channel ."""

implementation:
for chunk in chunks ( channel_metadata_item_map , self . enterprise_configuration . transmission_chunk_size ) :
    serialized_chunk = self . _serialize_items ( list ( chunk . values ( ) ) )
    try :
        self . client . create_content_metadata ( serialized_chunk )
    except ClientError as exc :
        LOGGER . error ( 'Failed to update [%s] content metadata items for integrated channel [%s] [%s]' , len ( chunk ) , self . enterprise_configuration . enterprise_customer . name , self . enterprise_configuration . channel_code , )
        LOGGER . error ( exc )
    else :
        self . _create_transmissions ( chunk )


signature:
def _transmit_update ( self , channel_metadata_item_map , transmission_map ) :
    """Transmit content metadata update to integrated channel ."""

implementation:
for chunk in chunks ( channel_metadata_item_map , self . enterprise_configuration . transmission_chunk_size ) :
    serialized_chunk = self . _serialize_items ( list ( chunk . values ( ) ) )
    try :
        self . client . update_content_metadata ( serialized_chunk )
    except ClientError as exc :
        LOGGER . error ( 'Failed to update [%s] content metadata items for integrated channel [%s] [%s]' , len ( chunk ) , self . enterprise_configuration . enterprise_customer . name , self . enterprise_configuration . channel_code , )
        LOGGER . error ( exc )
    else :
        self . _update_transmissions ( chunk , transmission_map )


signature:
def _transmit_delete ( self , channel_metadata_item_map ) :
    """Transmit content metadata deletion to integrated channel ."""

implementation:
for chunk in chunks ( channel_metadata_item_map , self . enterprise_configuration . transmission_chunk_size ) :
    serialized_chunk = self . _serialize_items ( list ( chunk . values ( ) ) )
    try :
        self . client . delete_content_metadata ( serialized_chunk )
    except ClientError as exc :
        LOGGER . error ( 'Failed to delete [%s] content metadata items for integrated channel [%s] [%s]' , len ( chunk ) , self . enterprise_configuration . enterprise_customer . name , self . enterprise_configuration . channel_code , )
        LOGGER . error ( exc )
    else :
        self . _delete_transmissions ( chunk . keys ( ) )


signature:
def _get_transmissions ( self ) :
    """Return the ContentMetadataItemTransmision models for previously
 transmitted content metadata items ."""

implementation:
# pylint : disable = invalid - name
ContentMetadataItemTransmission = apps . get_model ( 'integrated_channel' , 'ContentMetadataItemTransmission' )
return ContentMetadataItemTransmission . objects . filter ( enterprise_customer = self . enterprise_configuration . enterprise_customer , integrated_channel_code = self . enterprise_configuration . channel_code ( ) )


signature:
def _create_transmissions ( self , content_metadata_item_map ) :
    """Create ContentMetadataItemTransmision models for the given content metadata items ."""

implementation:
# pylint : disable = invalid - name
ContentMetadataItemTransmission = apps . get_model ( 'integrated_channel' , 'ContentMetadataItemTransmission' )
transmissions = [ ]
for content_id , channel_metadata in content_metadata_item_map . items ( ) :
    transmissions . append ( ContentMetadataItemTransmission ( enterprise_customer = self . enterprise_configuration . enterprise_customer , integrated_channel_code = self . enterprise_configuration . channel_code ( ) , content_id = content_id , channel_metadata = channel_metadata ) )
ContentMetadataItemTransmission . objects . bulk_create ( transmissions )


signature:
def _update_transmissions ( self , content_metadata_item_map , transmission_map ) :
    """Update ContentMetadataItemTransmision models for the given content metadata items ."""

implementation:
for content_id , channel_metadata in content_metadata_item_map . items ( ) :
    transmission = transmission_map [ content_id ]
    transmission . channel_metadata = channel_metadata
    transmission . save ( )


signature:
def _delete_transmissions ( self , content_metadata_item_ids ) :
    """Delete ContentMetadataItemTransmision models associated with the given content metadata items ."""

implementation:
# pylint : disable = invalid - name
ContentMetadataItemTransmission = apps . get_model ( 'integrated_channel' , 'ContentMetadataItemTransmission' )
ContentMetadataItemTransmission . objects . filter ( enterprise_customer = self . enterprise_configuration . enterprise_customer , integrated_channel_code = self . enterprise_configuration . channel_code ( ) , content_id__in = content_metadata_item_ids ) . delete ( )


signature:
def validate_username ( self , value ) :
    """Verify that the username has a matching user , and that the user has an associated EnterpriseCustomerUser ."""

implementation:
try :
    user = User . objects . get ( username = value )
except User . DoesNotExist :
    raise serializers . ValidationError ( "User does not exist" )
try :
    enterprise_customer_user = models . EnterpriseCustomerUser . objects . get ( user_id = user . pk )
except models . EnterpriseCustomerUser . DoesNotExist :
    raise serializers . ValidationError ( "User has no EnterpriseCustomerUser" )
self . enterprise_customer_user = enterprise_customer_user
return value


signature:
def save ( self ) : # pylint : disable = arguments - differ
    """Save the model with the found EnterpriseCustomerUser ."""

implementation:
course_id = self . validated_data [ 'course_id' ]
__ , created = models . EnterpriseCourseEnrollment . objects . get_or_create ( enterprise_customer_user = self . enterprise_customer_user , course_id = course_id , )
if created :
    track_enrollment ( 'rest-api-enrollment' , self . enterprise_customer_user . user_id , course_id )


signature:
def to_representation ( self , instance ) :
    """Serialize the EnterpriseCustomerCatalog object .
 Arguments :
 instance ( EnterpriseCustomerCatalog ) : The EnterpriseCustomerCatalog to serialize .
 Returns :
 dict : The EnterpriseCustomerCatalog converted to a dict ."""

implementation:
request = self . context [ 'request' ]
enterprise_customer = instance . enterprise_customer
representation = super ( EnterpriseCustomerCatalogDetailSerializer , self ) . to_representation ( instance )
# Retrieve the EnterpriseCustomerCatalog search results from the discovery service .
paginated_content = instance . get_paginated_content ( request . GET )
count = paginated_content [ 'count' ]
search_results = paginated_content [ 'results' ]
for item in search_results :
    content_type = item [ 'content_type' ]
    marketing_url = item . get ( 'marketing_url' )
    if marketing_url :
        item [ 'marketing_url' ] = utils . update_query_parameters ( marketing_url , utils . get_enterprise_utm_context ( enterprise_customer ) )
    # Add the Enterprise enrollment URL to each content item returned from the discovery service .
    if content_type == 'course' :
        item [ 'enrollment_url' ] = instance . get_course_enrollment_url ( item [ 'key' ] )
    if content_type == 'courserun' :
        item [ 'enrollment_url' ] = instance . get_course_run_enrollment_url ( item [ 'key' ] )
    if content_type == 'program' :
        item [ 'enrollment_url' ] = instance . get_program_enrollment_url ( item [ 'uuid' ] )
# Build pagination URLs
previous_url = None
next_url = None
page = int ( request . GET . get ( 'page' , '1' ) )
request_uri = request . build_absolute_uri ( )
if paginated_content [ 'previous' ] :
    previous_url = utils . update_query_parameters ( request_uri , { 'page' : page - 1 } )
if paginated_content [ 'next' ] :
    next_url = utils . update_query_parameters ( request_uri , { 'page' : page + 1 } )
representation [ 'count' ] = count
representation [ 'previous' ] = previous_url
representation [ 'next' ] = next_url
representation [ 'results' ] = search_results
return representation


signature:
def get_groups ( self , obj ) :
    """Return the enterprise related django groups that this user is a part of ."""

implementation:
if obj . user :
    return [ group . name for group in obj . user . groups . filter ( name__in = ENTERPRISE_PERMISSION_GROUPS ) ]
return [ ]


signature:
def validate_username ( self , value ) :
    """Verify that the username has a matching user ."""

implementation:
try :
    self . user = User . objects . get ( username = value )
except User . DoesNotExist :
    raise serializers . ValidationError ( "User does not exist" )
return value


signature:
def save ( self ) : # pylint : disable = arguments - differ
    """Save the EnterpriseCustomerUser ."""

implementation:
enterprise_customer = self . validated_data [ 'enterprise_customer' ]
ecu = models . EnterpriseCustomerUser ( user_id = self . user . pk , enterprise_customer = enterprise_customer , )
ecu . save ( )


signature:
def to_representation ( self , instance ) :
    """Return the updated course data dictionary .
 Arguments :
 instance ( dict ) : The course data .
 Returns :
 dict : The updated course data ."""

implementation:
updated_course = copy . deepcopy ( instance )
enterprise_customer_catalog = self . context [ 'enterprise_customer_catalog' ]
updated_course [ 'enrollment_url' ] = enterprise_customer_catalog . get_course_enrollment_url ( updated_course [ 'key' ] )
for course_run in updated_course [ 'course_runs' ] :
    course_run [ 'enrollment_url' ] = enterprise_customer_catalog . get_course_run_enrollment_url ( course_run [ 'key' ] )
return updated_course


signature:
def to_representation ( self , instance ) :
    """Return the updated course run data dictionary .
 Arguments :
 instance ( dict ) : The course run data .
 Returns :
 dict : The updated course run data ."""

implementation:
updated_course_run = copy . deepcopy ( instance )
enterprise_customer_catalog = self . context [ 'enterprise_customer_catalog' ]
updated_course_run [ 'enrollment_url' ] = enterprise_customer_catalog . get_course_run_enrollment_url ( updated_course_run [ 'key' ] )
return updated_course_run


signature:
def to_representation ( self , instance ) :
    """Return the updated program data dictionary .
 Arguments :
 instance ( dict ) : The program data .
 Returns :
 dict : The updated program data ."""

implementation:
updated_program = copy . deepcopy ( instance )
enterprise_customer_catalog = self . context [ 'enterprise_customer_catalog' ]
updated_program [ 'enrollment_url' ] = enterprise_customer_catalog . get_program_enrollment_url ( updated_program [ 'uuid' ] )
for course in updated_program [ 'courses' ] :
    course [ 'enrollment_url' ] = enterprise_customer_catalog . get_course_enrollment_url ( course [ 'key' ] )
    for course_run in course [ 'course_runs' ] :
        course_run [ 'enrollment_url' ] = enterprise_customer_catalog . get_course_run_enrollment_url ( course_run [ 'key' ] )
return updated_program


signature:
def to_internal_value ( self , data ) :
    """This implements the same relevant logic as ListSerializer except that if one or more items fail validation ,
 processing for other items that did not fail will continue ."""

implementation:
if not isinstance ( data , list ) :
    message = self . error_messages [ 'not_a_list' ] . format ( input_type = type ( data ) . __name__ )
    raise serializers . ValidationError ( { api_settings . NON_FIELD_ERRORS_KEY : [ message ] } )
ret = [ ]
for item in data :
    try :
        validated = self . child . run_validation ( item )
    except serializers . ValidationError as exc :
        ret . append ( exc . detail )
    else :
        ret . append ( validated )
return ret


signature:
def create ( self , validated_data ) :
    """This selectively calls the child create method based on whether or not validation failed for each payload ."""

implementation:
ret = [ ]
for attrs in validated_data :
    if 'non_field_errors' not in attrs and not any ( isinstance ( attrs [ field ] , list ) for field in attrs ) :
        ret . append ( self . child . create ( attrs ) )
    else :
        ret . append ( attrs )
return ret


signature:
def to_representation ( self , data ) :
    """This selectively calls to _ representation on each result that was processed by create ."""

implementation: return [ self . child . to_representation ( item ) if 'detail' in item else item for item in data ]


signature:
def create ( self , validated_data ) :
    """Perform the enrollment for existing enterprise customer users , or create the pending objects for new users ."""

implementation:
enterprise_customer = self . context . get ( 'enterprise_customer' )
lms_user = validated_data . get ( 'lms_user_id' )
tpa_user = validated_data . get ( 'tpa_user_id' )
user_email = validated_data . get ( 'user_email' )
course_run_id = validated_data . get ( 'course_run_id' )
course_mode = validated_data . get ( 'course_mode' )
cohort = validated_data . get ( 'cohort' )
email_students = validated_data . get ( 'email_students' )
is_active = validated_data . get ( 'is_active' )
enterprise_customer_user = lms_user or tpa_user or user_email
if isinstance ( enterprise_customer_user , models . EnterpriseCustomerUser ) :
    validated_data [ 'enterprise_customer_user' ] = enterprise_customer_user
    try :
        if is_active :
            enterprise_customer_user . enroll ( course_run_id , course_mode , cohort = cohort )
        else :
            enterprise_customer_user . unenroll ( course_run_id )
    except ( CourseEnrollmentDowngradeError , CourseEnrollmentPermissionError , HttpClientError ) as exc :
        validated_data [ 'detail' ] = str ( exc )
        return validated_data
    if is_active :
        track_enrollment ( 'enterprise-customer-enrollment-api' , enterprise_customer_user . user_id , course_run_id )
else :
    if is_active :
        enterprise_customer_user = enterprise_customer . enroll_user_pending_registration ( user_email , course_mode , course_run_id , cohort = cohort )
    else :
        enterprise_customer . clear_pending_registration ( user_email , course_run_id )
if email_students :
    enterprise_customer . notify_enrolled_learners ( self . context . get ( 'request_user' ) , course_run_id , [ enterprise_customer_user ] )
validated_data [ 'detail' ] = 'success'
return validated_data


signature:
def validate_lms_user_id ( self , value ) :
    """Validates the lms _ user _ id , if is given , to see if there is an existing EnterpriseCustomerUser for it ."""

implementation:
enterprise_customer = self . context . get ( 'enterprise_customer' )
try : # Ensure the given user is associated with the enterprise .
    return models . EnterpriseCustomerUser . objects . get ( user_id = value , enterprise_customer = enterprise_customer )
except models . EnterpriseCustomerUser . DoesNotExist :
    pass
return None


signature:
def validate_tpa_user_id ( self , value ) :
    """Validates the tpa _ user _ id , if is given , to see if there is an existing EnterpriseCustomerUser for it .
 It first uses the third party auth api to find the associated username to do the lookup ."""

implementation:
enterprise_customer = self . context . get ( 'enterprise_customer' )
try :
    tpa_client = ThirdPartyAuthApiClient ( )
    username = tpa_client . get_username_from_remote_id ( enterprise_customer . identity_provider , value )
    user = User . objects . get ( username = username )
    return models . EnterpriseCustomerUser . objects . get ( user_id = user . id , enterprise_customer = enterprise_customer )
except ( models . EnterpriseCustomerUser . DoesNotExist , User . DoesNotExist ) :
    pass
return None


signature:
def validate_user_email ( self , value ) :
    """Validates the user _ email , if given , to see if an existing EnterpriseCustomerUser exists for it .
 If it does not , it does not fail validation , unlike for the other field validation methods above ."""

implementation:
enterprise_customer = self . context . get ( 'enterprise_customer' )
try :
    user = User . objects . get ( email = value )
    return models . EnterpriseCustomerUser . objects . get ( user_id = user . id , enterprise_customer = enterprise_customer )
except ( models . EnterpriseCustomerUser . DoesNotExist , User . DoesNotExist ) :
    pass
return value


signature:
def validate_course_run_id ( self , value ) :
    """Validates that the course run id is part of the Enterprise Customer ' s catalog ."""

implementation:
enterprise_customer = self . context . get ( 'enterprise_customer' )
if not enterprise_customer . catalog_contains_course ( value ) :
    raise serializers . ValidationError ( 'The course run id {course_run_id} is not in the catalog ' 'for Enterprise Customer {enterprise_customer}' . format ( course_run_id = value , enterprise_customer = enterprise_customer . name , ) )
return value


signature:
def validate ( self , data ) : # pylint : disable = arguments - differ
    """Validate that at least one of the user identifier fields has been passed in ."""

implementation:
lms_user_id = data . get ( 'lms_user_id' )
tpa_user_id = data . get ( 'tpa_user_id' )
user_email = data . get ( 'user_email' )
if not lms_user_id and not tpa_user_id and not user_email :
    raise serializers . ValidationError ( 'At least one of the following fields must be specified and map to an EnterpriseCustomerUser: ' 'lms_user_id, tpa_user_id, user_email' )
return data


signature:
def get_paginated_response ( data , request ) :
    """Update pagination links in course catalog data and return DRF Response .
 Arguments :
 data ( dict ) : Dictionary containing catalog courses .
 request ( HttpRequest ) : Current request object .
 Returns :
 ( Response ) : DRF response object containing pagination links ."""

implementation:
url = urlparse ( request . build_absolute_uri ( ) ) . _replace ( query = None ) . geturl ( )
next_page = None
previous_page = None
if data [ 'next' ] :
    next_page = "{base_url}?{query_parameters}" . format ( base_url = url , query_parameters = urlparse ( data [ 'next' ] ) . query , )
    next_page = next_page . rstrip ( '?' )
if data [ 'previous' ] :
    previous_page = "{base_url}?{query_parameters}" . format ( base_url = url , query_parameters = urlparse ( data [ 'previous' ] or "" ) . query , )
    previous_page = previous_page . rstrip ( '?' )
return Response ( OrderedDict ( [ ( 'count' , data [ 'count' ] ) , ( 'next' , next_page ) , ( 'previous' , previous_page ) , ( 'results' , data [ 'results' ] ) ] ) )


signature:
def create_switch ( apps , schema_editor ) :
    """Create the ` role _ based _ access _ control ` switch if it does not already exist ."""

implementation:
Switch = apps . get_model ( 'waffle' , 'Switch' )
Switch . objects . update_or_create ( name = ENTERPRISE_ROLE_BASED_ACCESS_CONTROL_SWITCH , defaults = { 'active' : False } )


signature:
def delete_switch ( apps , schema_editor ) :
    """Delete the ` role _ based _ access _ control ` switch ."""

implementation:
Switch = apps . get_model ( 'waffle' , 'Switch' )
Switch . objects . filter ( name = ENTERPRISE_ROLE_BASED_ACCESS_CONTROL_SWITCH ) . delete ( )


signature:
def create_switch ( apps , schema_editor ) :
    """Create and activate the SAP _ USE _ ENTERPRISE _ ENROLLMENT _ PAGE switch if it does not already exist ."""

implementation:
Switch = apps . get_model ( 'waffle' , 'Switch' )
Switch . objects . get_or_create ( name = 'SAP_USE_ENTERPRISE_ENROLLMENT_PAGE' , defaults = { 'active' : False } )


signature:
def transmit ( self , payload , ** kwargs ) :
    """Send a completion status call to SAP SuccessFactors using the client .
 Args :
 payload : The learner completion data payload to send to SAP SuccessFactors"""

implementation:
kwargs [ 'app_label' ] = 'sap_success_factors'
kwargs [ 'model_name' ] = 'SapSuccessFactorsLearnerDataTransmissionAudit'
kwargs [ 'remote_user_id' ] = 'sapsf_user_id'
super ( SapSuccessFactorsLearnerTransmitter , self ) . transmit ( payload , ** kwargs )


signature:
def handle_transmission_error ( self , learner_data , request_exception ) :
    """Handle the case where the employee on SAPSF ' s side is marked as inactive ."""

implementation:
try :
    sys_msg = request_exception . response . content
except AttributeError :
    pass
else :
    if 'user account is inactive' in sys_msg :
        ecu = EnterpriseCustomerUser . objects . get ( enterprise_enrollments__id = learner_data . enterprise_course_enrollment_id )
        ecu . active = False
        ecu . save ( )
        LOGGER . warning ( 'User %s with ID %s and email %s is a former employee of %s ' 'and has been marked inactive in SAPSF. Now marking inactive internally.' , ecu . username , ecu . user_id , ecu . user_email , ecu . enterprise_customer )
        return
super ( SapSuccessFactorsLearnerTransmitter , self ) . handle_transmission_error ( learner_data , request_exception )


signature:
def allow_request ( self , request , view ) :
    """Modify throttling for service users .
 Updates throttling rate if the request is coming from the service user , and
 defaults to UserRateThrottle ' s configured setting otherwise .
 Updated throttling rate comes from ` DEFAULT _ THROTTLE _ RATES ` key in ` REST _ FRAMEWORK `
 setting . service user throttling is specified in ` DEFAULT _ THROTTLE _ RATES ` by ` service _ user ` key
 Example Setting :
 REST _ FRAMEWORK = {
 ' DEFAULT _ THROTTLE _ RATES ' : {
 ' service _ user ' : ' 50 / day '"""

implementation:
service_users = get_service_usernames ( )
# User service user throttling rates for service user .
if request . user . username in service_users :
    self . update_throttle_scope ( )
return super ( ServiceUserThrottle , self ) . allow_request ( request , view )


signature:
def update_throttle_scope ( self ) :
    """Update throttle scope so that service user throttle rates are applied ."""

implementation:
self . scope = SERVICE_USER_SCOPE
self . rate = self . get_rate ( )
self . num_requests , self . duration = self . parse_rate ( self . rate )


signature:
def get_course_final_price ( self , mode , currency = '$' , enterprise_catalog_uuid = None ) :
    """Get course mode ' s SKU discounted price after applying any entitlement available for this user .
 Returns :
 str : Discounted price of the course mode ."""

implementation:
try :
    price_details = self . client . baskets . calculate . get ( sku = [ mode [ 'sku' ] ] , username = self . user . username , catalog = enterprise_catalog_uuid , )
except ( SlumberBaseException , ConnectionError , Timeout ) as exc :
    LOGGER . exception ( 'Failed to get price details for sku %s due to: %s' , mode [ 'sku' ] , str ( exc ) )
    price_details = { }
price = price_details . get ( 'total_incl_tax' , mode [ 'min_price' ] )
if price != mode [ 'min_price' ] :
    return format_price ( price , currency )
return mode [ 'original_price' ]


signature:
def update_enterprise_courses ( self , enterprise_customer , course_container_key = 'results' , ** kwargs ) :
    """This method adds enterprise - specific metadata for each course .
 We are adding following field in all the courses .
 tpa _ hint : a string for identifying Identity Provider .
 enterprise _ id : the UUID of the enterprise
 * * kwargs : any additional data one would like to add on a per - use basis .
 Arguments :
 enterprise _ customer : The customer whose data will be used to fill the enterprise context .
 course _ container _ key : The key used to find the container for courses in the serializer ' s data dictionary ."""

implementation:
enterprise_context = { 'tpa_hint' : enterprise_customer and enterprise_customer . identity_provider , 'enterprise_id' : enterprise_customer and str ( enterprise_customer . uuid ) , }
enterprise_context . update ( ** kwargs )
courses = [ ]
for course in self . data [ course_container_key ] :
    courses . append ( self . update_course ( course , enterprise_customer , enterprise_context ) )
self . data [ course_container_key ] = courses


signature:
def update_course ( self , course , enterprise_customer , enterprise_context ) :
    """Update course metadata of the given course and return updated course .
 Arguments :
 course ( dict ) : Course Metadata returned by course catalog API
 enterprise _ customer ( EnterpriseCustomer ) : enterprise customer instance .
 enterprise _ context ( dict ) : Enterprise context to be added to course runs and URLs . .
 Returns :
 ( dict ) : Updated course metadata"""

implementation:
course [ 'course_runs' ] = self . update_course_runs ( course_runs = course . get ( 'course_runs' ) or [ ] , enterprise_customer = enterprise_customer , enterprise_context = enterprise_context , )
# Update marketing urls in course metadata to include enterprise related info ( i . e . our global context ) .
marketing_url = course . get ( 'marketing_url' )
if marketing_url :
    query_parameters = dict ( enterprise_context , ** utils . get_enterprise_utm_context ( enterprise_customer ) )
    course . update ( { 'marketing_url' : utils . update_query_parameters ( marketing_url , query_parameters ) } )
# Finally , add context to the course as a whole .
course . update ( enterprise_context )
return course


signature:
def update_course_runs ( self , course_runs , enterprise_customer , enterprise_context ) :
    """Update Marketing urls in course metadata and return updated course .
 Arguments :
 course _ runs ( list ) : List of course runs .
 enterprise _ customer ( EnterpriseCustomer ) : enterprise customer instance .
 enterprise _ context ( dict ) : The context to inject into URLs .
 Returns :
 ( dict ) : Dictionary containing updated course metadata ."""

implementation:
updated_course_runs = [ ]
for course_run in course_runs :
    track_selection_url = utils . get_course_track_selection_url ( course_run = course_run , query_parameters = dict ( enterprise_context , ** utils . get_enterprise_utm_context ( enterprise_customer ) ) , )
    enrollment_url = enterprise_customer . get_course_run_enrollment_url ( course_run . get ( 'key' ) )
    course_run . update ( { 'enrollment_url' : enrollment_url , 'track_selection_url' : track_selection_url , } )
    # Update marketing urls in course metadata to include enterprise related info .
    marketing_url = course_run . get ( 'marketing_url' )
    if marketing_url :
        query_parameters = dict ( enterprise_context , ** utils . get_enterprise_utm_context ( enterprise_customer ) )
        course_run . update ( { 'marketing_url' : utils . update_query_parameters ( marketing_url , query_parameters ) } )
    # Add updated course run to the list .
    updated_course_runs . append ( course_run )
return updated_course_runs


signature:
def export ( self ) :
    """Collect learner data for the ` ` EnterpriseCustomer ` ` where data sharing consent is granted .
 Yields a learner data object for each enrollment , containing :
 * ` ` enterprise _ enrollment ` ` : ` ` EnterpriseCourseEnrollment ` ` object .
 * ` ` completed _ date ` ` : datetime instance containing the course / enrollment completion date ; None if not complete .
 " Course completion " occurs for instructor - paced courses when course certificates are issued , and
 for self - paced courses , when the course end date is passed , or when the learner achieves a passing grade .
 * ` ` grade ` ` : string grade recorded for the learner in the course ."""

implementation:
# Fetch the consenting enrollment data , including the enterprise _ customer _ user .
# Order by the course _ id , to avoid fetching course API data more than we have to .
enrollment_queryset = EnterpriseCourseEnrollment . objects . select_related ( 'enterprise_customer_user' ) . filter ( enterprise_customer_user__enterprise_customer = self . enterprise_customer , enterprise_customer_user__active = True , ) . order_by ( 'course_id' )
# Fetch course details from the Course API , and cache between calls .
course_details = None
for enterprise_enrollment in enrollment_queryset :
    course_id = enterprise_enrollment . course_id
    # Fetch course details from Courses API
    # pylint : disable = unsubscriptable - object
    if course_details is None or course_details [ 'course_id' ] != course_id :
        if self . course_api is None :
            self . course_api = CourseApiClient ( )
        course_details = self . course_api . get_course_details ( course_id )
    if course_details is None : # Course not found , so we have nothing to report .
        LOGGER . error ( "No course run details found for enrollment [%d]: [%s]" , enterprise_enrollment . pk , course_id )
        continue
    consent = DataSharingConsent . objects . proxied_get ( username = enterprise_enrollment . enterprise_customer_user . username , course_id = enterprise_enrollment . course_id , enterprise_customer = enterprise_enrollment . enterprise_customer_user . enterprise_customer )
    if not consent . granted or enterprise_enrollment . audit_reporting_disabled :
        continue
    # For instructor - paced courses , let the certificate determine course completion
    if course_details . get ( 'pacing' ) == 'instructor' :
        completed_date , grade , is_passing = self . _collect_certificate_data ( enterprise_enrollment )
    # For self - paced courses , check the Grades API
    else :
        completed_date , grade , is_passing = self . _collect_grades_data ( enterprise_enrollment , course_details )
    records = self . get_learner_data_records ( enterprise_enrollment = enterprise_enrollment , completed_date = completed_date , grade = grade , is_passing = is_passing , )
    if records : # There are some cases where we won ' t receive a record from the above
    # method ; right now , that should only happen if we have an Enterprise - linked
    # user for the integrated channel , and transmission of that user ' s
    # data requires an upstream user identifier that we don ' t have ( due to a
    # failure of SSO or similar ) . In such a case , ` get _ learner _ data _ record `
    # would return None , and we ' d simply skip yielding it here .
        for record in records :
            yield record


signature:
def get_learner_data_records ( self , enterprise_enrollment , completed_date = None , grade = None , is_passing = False ) :
    """Generate a learner data transmission audit with fields properly filled in ."""

implementation:
# pylint : disable = invalid - name
LearnerDataTransmissionAudit = apps . get_model ( 'integrated_channel' , 'LearnerDataTransmissionAudit' )
completed_timestamp = None
course_completed = False
if completed_date is not None :
    completed_timestamp = parse_datetime_to_epoch_millis ( completed_date )
    course_completed = is_passing
return [ LearnerDataTransmissionAudit ( enterprise_course_enrollment_id = enterprise_enrollment . id , course_id = enterprise_enrollment . course_id , course_completed = course_completed , completed_timestamp = completed_timestamp , grade = grade , ) ]


signature:
def _collect_certificate_data ( self , enterprise_enrollment ) :
    """Collect the learner completion data from the course certificate .
 Used for Instructor - paced courses .
 If no certificate is found , then returns the completed _ date = None , grade = In Progress , on the idea that a
 certificate will eventually be generated .
 Args :
 enterprise _ enrollment ( EnterpriseCourseEnrollment ) : the enterprise enrollment record for which we need to
 collect completion / grade data
 Returns :
 completed _ date : Date the course was completed , this is None if course has not been completed .
 grade : Current grade in the course .
 is _ passing : Boolean indicating if the grade is a passing grade or not ."""

implementation:
if self . certificates_api is None :
    self . certificates_api = CertificatesApiClient ( self . user )
course_id = enterprise_enrollment . course_id
username = enterprise_enrollment . enterprise_customer_user . user . username
try :
    certificate = self . certificates_api . get_course_certificate ( course_id , username )
    completed_date = certificate . get ( 'created_date' )
    if completed_date :
        completed_date = parse_datetime ( completed_date )
    else :
        completed_date = timezone . now ( )
    # For consistency with _ collect _ grades _ data , we only care about Pass / Fail grades . This could change .
    is_passing = certificate . get ( 'is_passing' )
    grade = self . grade_passing if is_passing else self . grade_failing
except HttpNotFoundError :
    completed_date = None
    grade = self . grade_incomplete
    is_passing = False
return completed_date , grade , is_passing


signature:
def _collect_grades_data ( self , enterprise_enrollment , course_details ) :
    """Collect the learner completion data from the Grades API .
 Used for self - paced courses .
 Args :
 enterprise _ enrollment ( EnterpriseCourseEnrollment ) : the enterprise enrollment record for which we need to
 collect completion / grade data
 course _ details ( dict ) : the course details for the course in the enterprise enrollment record .
 Returns :
 completed _ date : Date the course was completed , this is None if course has not been completed .
 grade : Current grade in the course .
 is _ passing : Boolean indicating if the grade is a passing grade or not ."""

implementation:
if self . grades_api is None :
    self . grades_api = GradesApiClient ( self . user )
course_id = enterprise_enrollment . course_id
username = enterprise_enrollment . enterprise_customer_user . user . username
try :
    grades_data = self . grades_api . get_course_grade ( course_id , username )
except HttpNotFoundError as error : # Grade not found , so we have nothing to report .
    if hasattr ( error , 'content' ) :
        response_content = json . loads ( error . content )
        if response_content . get ( 'error_code' , '' ) == 'user_not_enrolled' : # This means the user has an enterprise enrollment record but is not enrolled in the course yet
            LOGGER . info ( "User [%s] not enrolled in course [%s], enterprise enrollment [%d]" , username , course_id , enterprise_enrollment . pk )
            return None , None , None
    LOGGER . error ( "No grades data found for [%d]: [%s], [%s]" , enterprise_enrollment . pk , course_id , username )
    return None , None , None
# Prepare to process the course end date and pass / fail grade
course_end_date = course_details . get ( 'end' )
if course_end_date is not None :
    course_end_date = parse_datetime ( course_end_date )
now = timezone . now ( )
is_passing = grades_data . get ( 'passed' )
# We can consider a course complete if :
# * the course ' s end date has passed
if course_end_date is not None and course_end_date < now :
    completed_date = course_end_date
    grade = self . grade_passing if is_passing else self . grade_failing
# * Or , the learner has a passing grade ( as of now )
elif is_passing :
    completed_date = now
    grade = self . grade_passing
# Otherwise , the course is still in progress
else :
    completed_date = None
    grade = self . grade_incomplete
return completed_date , grade , is_passing


signature:
def get_enterprise_user_id ( self , obj ) :
    """Get enterprise user id from user object .
 Arguments :
 obj ( User ) : Django User object
 Returns :
 ( int ) : Primary Key identifier for enterprise user object ."""

implementation:
# An enterprise learner can not belong to multiple enterprise customer at the same time
# but if such scenario occurs we will pick the first .
enterprise_learner = EnterpriseCustomerUser . objects . filter ( user_id = obj . id ) . first ( )
return enterprise_learner and enterprise_learner . id


signature:
def get_enterprise_sso_uid ( self , obj ) :
    """Get enterprise SSO UID .
 Arguments :
 obj ( User ) : Django User object
 Returns :
 ( str ) : string containing UUID for enterprise customer ' s Identity Provider ."""

implementation:
# An enterprise learner can not belong to multiple enterprise customer at the same time
# but if such scenario occurs we will pick the first .
enterprise_learner = EnterpriseCustomerUser . objects . filter ( user_id = obj . id ) . first ( )
return enterprise_learner and enterprise_learner . get_remote_id ( )


signature:
def get_course_duration ( self , obj ) :
    """Get course ' s duration as a timedelta .
 Arguments :
 obj ( CourseOverview ) : CourseOverview object
 Returns :
 ( timedelta ) : Duration of a course ."""

implementation:
duration = obj . end - obj . start if obj . start and obj . end else None
if duration :
    return strfdelta ( duration , '{W} weeks {D} days.' )
return ''


signature:
def transmit ( self , payload , ** kwargs ) :
    """Transmit content metadata items to the integrated channel ."""

implementation:
items_to_create , items_to_update , items_to_delete , transmission_map = self . _partition_items ( payload )
self . _prepare_items_for_delete ( items_to_delete )
prepared_items = { }
prepared_items . update ( items_to_create )
prepared_items . update ( items_to_update )
prepared_items . update ( items_to_delete )
skip_metadata_transmission = False
for chunk in chunks ( prepared_items , self . enterprise_configuration . transmission_chunk_size ) :
    chunked_items = list ( chunk . values ( ) )
    if skip_metadata_transmission : # Remove the failed items from the create / update / delete dictionaries ,
    # so ContentMetadataItemTransmission objects are not synchronized for
    # these items below .
        self . _remove_failed_items ( chunked_items , items_to_create , items_to_update , items_to_delete )
    else :
        try :
            self . client . update_content_metadata ( self . _serialize_items ( chunked_items ) )
        except ClientError as exc :
            LOGGER . error ( 'Failed to update [%s] content metadata items for integrated channel [%s] [%s]' , len ( chunked_items ) , self . enterprise_configuration . enterprise_customer . name , self . enterprise_configuration . channel_code , )
            LOGGER . error ( exc )
            # Remove the failed items from the create / update / delete dictionaries ,
            # so ContentMetadataItemTransmission objects are not synchronized for
            # these items below .
            self . _remove_failed_items ( chunked_items , items_to_create , items_to_update , items_to_delete )
            # SAP servers throttle incoming traffic , If a request fails than the subsequent would fail too ,
            # So , no need to keep trying and failing . We should stop here and retry later .
            skip_metadata_transmission = True
self . _create_transmissions ( items_to_create )
self . _update_transmissions ( items_to_update , transmission_map )
self . _delete_transmissions ( items_to_delete . keys ( ) )


signature:
def _remove_failed_items ( self , failed_items , items_to_create , items_to_update , items_to_delete ) :
    """Remove content metadata items from the ` items _ to _ create ` , ` items _ to _ update ` , ` items _ to _ delete ` dicts .
 Arguments :
 failed _ items ( list ) : Failed Items to be removed .
 items _ to _ create ( dict ) : dict containing the items created successfully .
 items _ to _ update ( dict ) : dict containing the items updated successfully .
 items _ to _ delete ( dict ) : dict containing the items deleted successfully ."""

implementation:
for item in failed_items :
    content_metadata_id = item [ 'courseID' ]
    items_to_create . pop ( content_metadata_id , None )
    items_to_update . pop ( content_metadata_id , None )
    items_to_delete . pop ( content_metadata_id , None )


signature:
def parse_arguments ( * args , ** options ) : # pylint : disable = unused - argument
    """Parse and validate arguments for send _ course _ enrollments command .
 Arguments :
 * args : Positional arguments passed to the command
 * * options : optional arguments passed to the command
 Returns :
 A tuple containing parsed values for
 1 . days ( int ) : Integer showing number of days to lookup enterprise enrollments ,
 course completion etc and send to xAPI LRS
 2 . enterprise _ customer _ uuid ( EnterpriseCustomer ) : Enterprise Customer if present then
 send xAPI statements just for this enterprise ."""

implementation:
days = options . get ( 'days' , 1 )
enterprise_customer_uuid = options . get ( 'enterprise_customer_uuid' )
enterprise_customer = None
if enterprise_customer_uuid :
    try : # pylint : disable = no - member
        enterprise_customer = EnterpriseCustomer . objects . get ( uuid = enterprise_customer_uuid )
    except EnterpriseCustomer . DoesNotExist :
        raise CommandError ( 'Enterprise customer with uuid "{enterprise_customer_uuid}" does not exist.' . format ( enterprise_customer_uuid = enterprise_customer_uuid ) )
return days , enterprise_customer


signature:
def handle ( self , * args , ** options ) :
    """Send xAPI statements ."""

implementation:
if not CourseEnrollment :
    raise NotConnectedToOpenEdX ( "This package must be installed in an OpenEdX environment." )
days , enterprise_customer = self . parse_arguments ( * args , ** options )
if enterprise_customer :
    try :
        lrs_configuration = XAPILRSConfiguration . objects . get ( active = True , enterprise_customer = enterprise_customer )
    except XAPILRSConfiguration . DoesNotExist :
        raise CommandError ( 'No xAPI Configuration found for "{enterprise_customer}"' . format ( enterprise_customer = enterprise_customer . name ) )
    # Send xAPI analytics data to the configured LRS
    self . send_xapi_statements ( lrs_configuration , days )
else :
    for lrs_configuration in XAPILRSConfiguration . objects . filter ( active = True ) :
        self . send_xapi_statements ( lrs_configuration , days )


signature:
def send_xapi_statements ( self , lrs_configuration , days ) :
    """Send xAPI analytics data of the enterprise learners to the given LRS .
 Arguments :
 lrs _ configuration ( XAPILRSConfiguration ) : Configuration object containing LRS configurations
 of the LRS where to send xAPI learner analytics .
 days ( int ) : Include course enrollment of this number of days ."""

implementation:
for course_enrollment in self . get_course_enrollments ( lrs_configuration . enterprise_customer , days ) :
    try :
        send_course_enrollment_statement ( lrs_configuration , course_enrollment )
    except ClientError :
        LOGGER . exception ( 'Client error while sending course enrollment to xAPI for' ' enterprise customer {enterprise_customer}.' . format ( enterprise_customer = lrs_configuration . enterprise_customer . name ) )


signature:
def get_course_enrollments ( self , enterprise_customer , days ) :
    """Get course enrollments for all the learners of given enterprise customer .
 Arguments :
 enterprise _ customer ( EnterpriseCustomer ) : Include Course enrollments for learners
 of this enterprise customer .
 days ( int ) : Include course enrollment of this number of days .
 Returns :
 ( list ) : A list of CourseEnrollment objects ."""

implementation: return CourseEnrollment . objects . filter ( created__gt = datetime . datetime . now ( ) - datetime . timedelta ( days = days ) ) . filter ( user_id__in = enterprise_customer . enterprise_customer_users . values_list ( 'user_id' , flat = True ) )


signature:
def course_modal ( context , course = None ) :
    """Django template tag that returns course information to display in a modal .
 You may pass in a particular course if you like . Otherwise , the modal will look for course context
 within the parent context .
 Usage :
 { % course _ modal % }
 { % course _ modal course % }"""

implementation:
if course :
    context . update ( { 'course_image_uri' : course . get ( 'course_image_uri' , '' ) , 'course_title' : course . get ( 'course_title' , '' ) , 'course_level_type' : course . get ( 'course_level_type' , '' ) , 'course_short_description' : course . get ( 'course_short_description' , '' ) , 'course_effort' : course . get ( 'course_effort' , '' ) , 'course_full_description' : course . get ( 'course_full_description' , '' ) , 'expected_learning_items' : course . get ( 'expected_learning_items' , [ ] ) , 'staff' : course . get ( 'staff' , [ ] ) , 'premium_modes' : course . get ( 'premium_modes' , [ ] ) , } )
return context


signature:
def link_to_modal ( link_text , index , autoescape = True ) : # pylint : disable = unused - argument
    """Django template filter that returns an anchor with attributes useful for course modal selection .
 General Usage :
 { { link _ text | link _ to _ modal : index } }
 Examples :
 { { course _ title | link _ to _ modal : forloop . counter0 } }
 { { course _ title | link _ to _ modal : 3 } }
 { { view _ details _ text | link _ to _ modal : 0 } }"""

implementation:
link = ( '<a' ' href="#!"' ' class="text-underline view-course-details-link"' ' id="view-course-details-link-{index}"' ' data-toggle="modal"' ' data-target="#course-details-modal-{index}"' '>{link_text}</a>' ) . format ( index = index , link_text = link_text , )
return mark_safe ( link )


signature:
def handle ( self , * args , ** options ) :
    """Unlink inactive EnterpriseCustomer ( s ) SAP learners ."""

implementation:
channels = self . get_integrated_channels ( options )
for channel in channels :
    channel_code = channel . channel_code ( )
    channel_pk = channel . pk
    if channel_code == 'SAP' : # Transmit the learner data to each integrated channel
        unlink_inactive_learners . delay ( channel_code , channel_pk )


signature:
def populate_data_sharing_consent ( apps , schema_editor ) :
    """Populates the ` ` DataSharingConsent ` ` model with the ` ` enterprise ` ` application ' s consent data .
 Consent data from the ` ` enterprise ` ` application come from the ` ` EnterpriseCourseEnrollment ` ` model ."""

implementation:
DataSharingConsent = apps . get_model ( 'consent' , 'DataSharingConsent' )
EnterpriseCourseEnrollment = apps . get_model ( 'enterprise' , 'EnterpriseCourseEnrollment' )
User = apps . get_model ( 'auth' , 'User' )
for enrollment in EnterpriseCourseEnrollment . objects . all ( ) :
    user = User . objects . get ( pk = enrollment . enterprise_customer_user . user_id )
    data_sharing_consent , __ = DataSharingConsent . objects . get_or_create ( username = user . username , enterprise_customer = enrollment . enterprise_customer_user . enterprise_customer , course_id = enrollment . course_id , )
    if enrollment . consent_granted is not None :
        data_sharing_consent . granted = enrollment . consent_granted
    else : # Check UDSCA instead .
        consent_state = enrollment . enterprise_customer_user . data_sharing_consent . first ( )
        if consent_state is not None :
            data_sharing_consent . granted = consent_state . state in [ 'enabled' , 'external' ]
        else :
            data_sharing_consent . granted = False
    data_sharing_consent . save ( )


signature:
def create_course_completion ( self , user_id , payload ) : # pylint : disable = unused - argument
    """Send a completion status payload to the Degreed Completion Status endpoint
 Args :
 user _ id : Unused .
 payload : JSON encoded object ( serialized from DegreedLearnerDataTransmissionAudit )
 containing completion status fields per Degreed documentation .
 Returns :
 A tuple containing the status code and the body of the response .
 Raises :
 HTTPError : if we received a failure response code from Degreed"""

implementation: return self . _post ( urljoin ( self . enterprise_configuration . degreed_base_url , self . global_degreed_config . completion_status_api_path ) , payload , self . COMPLETION_PROVIDER_SCOPE )


signature:
def delete_course_completion ( self , user_id , payload ) : # pylint : disable = unused - argument
    """Delete a completion status previously sent to the Degreed Completion Status endpoint
 Args :
 user _ id : Unused .
 payload : JSON encoded object ( serialized from DegreedLearnerDataTransmissionAudit )
 containing the required completion status fields for deletion per Degreed documentation .
 Returns :
 A tuple containing the status code and the body of the response .
 Raises :
 HTTPError : if we received a failure response code from Degreed"""

implementation: return self . _delete ( urljoin ( self . enterprise_configuration . degreed_base_url , self . global_degreed_config . completion_status_api_path ) , payload , self . COMPLETION_PROVIDER_SCOPE )


signature:
def _sync_content_metadata ( self , serialized_data , http_method ) :
    """Synchronize content metadata using the Degreed course content API .
 Args :
 serialized _ data : JSON - encoded object containing content metadata .
 http _ method : The HTTP method to use for the API request .
 Raises :
 ClientError : If Degreed API request fails ."""

implementation:
try :
    status_code , response_body = getattr ( self , '_' + http_method ) ( urljoin ( self . enterprise_configuration . degreed_base_url , self . global_degreed_config . course_api_path ) , serialized_data , self . CONTENT_PROVIDER_SCOPE )
except requests . exceptions . RequestException as exc :
    raise ClientError ( 'DegreedAPIClient request failed: {error} {message}' . format ( error = exc . __class__ . __name__ , message = str ( exc ) ) )
if status_code >= 400 :
    raise ClientError ( 'DegreedAPIClient request failed with status {status_code}: {message}' . format ( status_code = status_code , message = response_body ) )


signature:
def _post ( self , url , data , scope ) :
    """Make a POST request using the session object to a Degreed endpoint .
 Args :
 url ( str ) : The url to send a POST request to .
 data ( str ) : The json encoded payload to POST .
 scope ( str ) : Must be one of the scopes Degreed expects :
 - ` CONTENT _ PROVIDER _ SCOPE `
 - ` COMPLETION _ PROVIDER _ SCOPE `"""

implementation:
self . _create_session ( scope )
response = self . session . post ( url , data = data )
return response . status_code , response . text


signature:
def _delete ( self , url , data , scope ) :
    """Make a DELETE request using the session object to a Degreed endpoint .
 Args :
 url ( str ) : The url to send a DELETE request to .
 data ( str ) : The json encoded payload to DELETE .
 scope ( str ) : Must be one of the scopes Degreed expects :
 - ` CONTENT _ PROVIDER _ SCOPE `
 - ` COMPLETION _ PROVIDER _ SCOPE `"""

implementation:
self . _create_session ( scope )
response = self . session . delete ( url , data = data )
return response . status_code , response . text


signature:
def _create_session ( self , scope ) :
    """Instantiate a new session object for use in connecting with Degreed"""

implementation:
now = datetime . datetime . utcnow ( )
if self . session is None or self . expires_at is None or now >= self . expires_at : # Create a new session with a valid token
    if self . session :
        self . session . close ( )
    oauth_access_token , expires_at = self . _get_oauth_access_token ( self . enterprise_configuration . key , self . enterprise_configuration . secret , self . enterprise_configuration . degreed_user_id , self . enterprise_configuration . degreed_user_password , scope )
    session = requests . Session ( )
    session . timeout = self . SESSION_TIMEOUT
    session . headers [ 'Authorization' ] = 'Bearer {}' . format ( oauth_access_token )
    session . headers [ 'content-type' ] = 'application/json'
    self . session = session
    self . expires_at = expires_at


signature:
def _get_oauth_access_token ( self , client_id , client_secret , user_id , user_password , scope ) :
    """Retrieves OAuth 2.0 access token using the client credentials grant .
 Args :
 client _ id ( str ) : API client ID
 client _ secret ( str ) : API client secret
 user _ id ( str ) : Degreed company ID
 user _ password ( str ) : Degreed user password
 scope ( str ) : Must be one of the scopes Degreed expects :
 - ` CONTENT _ PROVIDER _ SCOPE `
 - ` COMPLETION _ PROVIDER _ SCOPE `
 Returns :
 tuple : Tuple containing access token string and expiration datetime .
 Raises :
 HTTPError : If we received a failure response code from Degreed .
 RequestException : If an unexpected response format was received that we could not parse ."""

implementation:
response = requests . post ( urljoin ( self . enterprise_configuration . degreed_base_url , self . global_degreed_config . oauth_api_path ) , data = { 'grant_type' : 'password' , 'username' : user_id , 'password' : user_password , 'scope' : scope , } , auth = ( client_id , client_secret ) , headers = { 'Content-Type' : 'application/x-www-form-urlencoded' } )
response . raise_for_status ( )
data = response . json ( )
try :
    expires_at = data [ 'expires_in' ] + int ( time . time ( ) )
    return data [ 'access_token' ] , datetime . datetime . utcfromtimestamp ( expires_at )
except KeyError :
    raise requests . RequestException ( response = response )


signature:
def ensure_data_exists ( self , request , data , error_message = None ) :
    """Ensure that the wrapped API client ' s response brings us valid data . If not , raise an error and log it ."""

implementation:
if not data :
    error_message = ( error_message or "Unable to fetch API response from endpoint '{}'." . format ( request . get_full_path ( ) ) )
    LOGGER . error ( error_message )
    raise NotFound ( error_message )


signature:
def contains_content_items ( self , request , pk , course_run_ids , program_uuids ) :
    """Return whether or not the specified content is available to the EnterpriseCustomer .
 Multiple course _ run _ ids and / or program _ uuids query parameters can be sent to this view to check
 for their existence in the EnterpriseCustomerCatalogs associated with this EnterpriseCustomer .
 At least one course run key or program UUID value must be included in the request ."""

implementation:
enterprise_customer = self . get_object ( )
# Maintain plus characters in course key .
course_run_ids = [ unquote ( quote_plus ( course_run_id ) ) for course_run_id in course_run_ids ]
contains_content_items = False
for catalog in enterprise_customer . enterprise_customer_catalogs . all ( ) :
    contains_course_runs = not course_run_ids or catalog . contains_courses ( course_run_ids )
    contains_program_uuids = not program_uuids or catalog . contains_programs ( program_uuids )
    if contains_course_runs and contains_program_uuids :
        contains_content_items = True
        break
return Response ( { 'contains_content_items' : contains_content_items } )


signature:
def courses ( self , request , pk = None ) : # pylint : disable = invalid - name , unused - argument
    """Retrieve the list of courses contained within the catalog linked to this enterprise .
 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:
enterprise_customer = self . get_object ( )
self . check_object_permissions ( request , enterprise_customer )
self . ensure_data_exists ( request , enterprise_customer . catalog , error_message = "No catalog is associated with Enterprise {enterprise_name} from endpoint '{path}'." . format ( enterprise_name = enterprise_customer . name , path = request . get_full_path ( ) ) )
# We have handled potential error cases and are now ready to call out to the Catalog API .
catalog_api = CourseCatalogApiClient ( request . user , enterprise_customer . site )
courses = catalog_api . get_paginated_catalog_courses ( enterprise_customer . catalog , request . GET )
# An empty response means that there was a problem fetching data from Catalog API , since
# a Catalog with no courses has a non empty response indicating that there are no courses .
self . ensure_data_exists ( request , courses , error_message = ( "Unable to fetch API response for catalog courses for " "Enterprise {enterprise_name} from endpoint '{path}'." . format ( enterprise_name = enterprise_customer . name , path = request . get_full_path ( ) ) ) )
serializer = serializers . EnterpriseCatalogCoursesReadOnlySerializer ( courses )
# Add enterprise related context for the courses .
serializer . update_enterprise_courses ( enterprise_customer , catalog_id = enterprise_customer . catalog )
return get_paginated_response ( serializer . data , request )


signature:
def course_enrollments ( self , request , pk ) :
    """Creates a course enrollment for an EnterpriseCustomerUser ."""

implementation:
enterprise_customer = self . get_object ( )
serializer = serializers . EnterpriseCustomerCourseEnrollmentsSerializer ( data = request . data , many = True , context = { 'enterprise_customer' : enterprise_customer , 'request_user' : request . user , } )
if serializer . is_valid ( ) :
    serializer . save ( )
    return Response ( serializer . data , status = HTTP_200_OK )
return Response ( serializer . errors , status = HTTP_400_BAD_REQUEST )


signature:
def with_access_to ( self , request , * args , ** kwargs ) : # pylint : disable = invalid - name , unused - argument
    """Returns the list of enterprise customers the user has a specified group permission access to ."""

implementation:
self . queryset = self . queryset . order_by ( 'name' )
enterprise_id = self . request . query_params . get ( 'enterprise_id' , None )
enterprise_slug = self . request . query_params . get ( 'enterprise_slug' , None )
enterprise_name = self . request . query_params . get ( 'search' , None )
if enterprise_id is not None :
    self . queryset = self . queryset . filter ( uuid = enterprise_id )
elif enterprise_slug is not None :
    self . queryset = self . queryset . filter ( slug = enterprise_slug )
elif enterprise_name is not None :
    self . queryset = self . queryset . filter ( name__icontains = enterprise_name )
return self . list ( request , * args , ** kwargs )


signature:
def entitlements ( self , request , pk = None ) : # pylint : disable = invalid - name , unused - argument
    """Retrieve the list of entitlements available to this learner .
 Only those entitlements are returned that satisfy enterprise customer ' s data sharing setting .
 Arguments :
 request ( HttpRequest ) : Reference to in - progress request instance .
 pk ( Int ) : Primary key value of the selected enterprise learner .
 Returns :
 ( HttpResponse ) : Response object containing a list of learner ' s entitlements ."""

implementation:
enterprise_customer_user = self . get_object ( )
instance = { "entitlements" : enterprise_customer_user . entitlements }
serializer = serializers . EnterpriseCustomerUserEntitlementSerializer ( instance , context = { 'request' : request } )
return Response ( serializer . data )


signature:
def contains_content_items ( self , request , pk , course_run_ids , program_uuids ) :
    """Return whether or not the EnterpriseCustomerCatalog contains the specified content .
 Multiple course _ run _ ids and / or program _ uuids query parameters can be sent to this view to check
 for their existence in the EnterpriseCustomerCatalog . At least one course run key
 or program UUID value must be included in the request ."""

implementation:
enterprise_customer_catalog = self . get_object ( )
# Maintain plus characters in course key .
course_run_ids = [ unquote ( quote_plus ( course_run_id ) ) for course_run_id in course_run_ids ]
contains_content_items = True
if course_run_ids :
    contains_content_items = enterprise_customer_catalog . contains_courses ( course_run_ids )
if program_uuids :
    contains_content_items = ( contains_content_items and enterprise_customer_catalog . contains_programs ( program_uuids ) )
return Response ( { 'contains_content_items' : contains_content_items } )


signature:
def course_detail ( self , request , pk , course_key ) : # pylint : disable = invalid - name , unused - argument
    """Return the metadata for the specified course .
 The course needs to be included in the specified EnterpriseCustomerCatalog
 in order for metadata to be returned from this endpoint ."""

implementation:
enterprise_customer_catalog = self . get_object ( )
course = enterprise_customer_catalog . get_course ( course_key )
if not course :
    raise Http404
context = self . get_serializer_context ( )
context [ 'enterprise_customer_catalog' ] = enterprise_customer_catalog
serializer = serializers . CourseDetailSerializer ( course , context = context )
return Response ( serializer . data )


signature:
def course_run_detail ( self , request , pk , course_id ) : # pylint : disable = invalid - name , unused - argument
    """Return the metadata for the specified course run .
 The course run needs to be included in the specified EnterpriseCustomerCatalog
 in order for metadata to be returned from this endpoint ."""

implementation:
enterprise_customer_catalog = self . get_object ( )
course_run = enterprise_customer_catalog . get_course_run ( course_id )
if not course_run :
    raise Http404
context = self . get_serializer_context ( )
context [ 'enterprise_customer_catalog' ] = enterprise_customer_catalog
serializer = serializers . CourseRunDetailSerializer ( course_run , context = context )
return Response ( serializer . data )


signature:
def program_detail ( self , request , pk , program_uuid ) : # pylint : disable = invalid - name , unused - argument
    """Return the metadata for the specified program .
 The program needs to be included in the specified EnterpriseCustomerCatalog
 in order for metadata to be returned from this endpoint ."""

implementation:
enterprise_customer_catalog = self . get_object ( )
program = enterprise_customer_catalog . get_program ( program_uuid )
if not program :
    raise Http404
context = self . get_serializer_context ( )
context [ 'enterprise_customer_catalog' ] = enterprise_customer_catalog
serializer = serializers . ProgramDetailSerializer ( program , context = context )
return Response ( serializer . data )


signature:
def list ( self , request ) :
    """DRF view to list all catalogs .
 Arguments :
 request ( HttpRequest ) : Current request
 Returns :
 ( Response ) : DRF response object containing course catalogs ."""

implementation:
catalog_api = CourseCatalogApiClient ( request . user )
catalogs = catalog_api . get_paginated_catalogs ( request . GET )
self . ensure_data_exists ( request , catalogs )
serializer = serializers . ResponsePaginationSerializer ( catalogs )
return get_paginated_response ( serializer . data , request )


signature:
def retrieve ( self , request , pk = None ) : # pylint : disable = invalid - name
    """DRF view to get catalog details .
 Arguments :
 request ( HttpRequest ) : Current request
 pk ( int ) : Course catalog identifier
 Returns :
 ( Response ) : DRF response object containing course catalogs ."""

implementation:
catalog_api = CourseCatalogApiClient ( request . user )
catalog = catalog_api . get_catalog ( pk )
self . ensure_data_exists ( request , catalog , error_message = ( "Unable to fetch API response for given catalog from endpoint '/catalog/{pk}/'. " "The resource you are looking for does not exist." . format ( pk = pk ) ) )
serializer = self . serializer_class ( catalog )
return Response ( serializer . data )