signature:
def get_enterprise_program_enrollment_page ( self , request , enterprise_customer , program_details ) :
    """Render Enterprise - specific program enrollment page ."""

implementation:
# Safely make the assumption that we can use the first authoring organization .
organizations = program_details [ 'authoring_organizations' ]
organization = organizations [ 0 ] if organizations else { }
platform_name = get_configuration_value ( 'PLATFORM_NAME' , settings . PLATFORM_NAME )
program_title = program_details [ 'title' ]
program_type_details = program_details [ 'type_details' ]
program_type = program_type_details [ 'name' ]
# Make any modifications for singular / plural - dependent text .
program_courses = program_details [ 'courses' ]
course_count = len ( program_courses )
course_count_text = ungettext ( '{count} Course' , '{count} Courses' , course_count , ) . format ( count = course_count )
effort_info_text = ungettext_min_max ( '{} hour per week, per course' , '{} hours per week, per course' , _ ( '{}-{} hours per week, per course' ) , program_details . get ( 'min_hours_effort_per_week' ) , program_details . get ( 'max_hours_effort_per_week' ) , )
length_info_text = ungettext_min_max ( '{} week per course' , '{} weeks per course' , _ ( '{}-{} weeks per course' ) , program_details . get ( 'weeks_to_complete_min' ) , program_details . get ( 'weeks_to_complete_max' ) , )
# Update some enrollment - related text requirements .
if program_details [ 'enrolled_in_program' ] :
    purchase_action = _ ( 'Purchase all unenrolled courses' )
    item = _ ( 'enrollment' )
else :
    purchase_action = _ ( 'Pursue the program' )
    item = _ ( 'program enrollment' )
# Add any DSC warning messages .
program_data_sharing_consent = get_data_sharing_consent ( request . user . username , enterprise_customer . uuid , program_uuid = program_details [ 'uuid' ] , )
if program_data_sharing_consent . exists and not program_data_sharing_consent . granted :
    messages . add_consent_declined_message ( request , enterprise_customer , program_title )
discount_data = program_details . get ( 'discount_data' , { } )
one_click_purchase_eligibility = program_details . get ( 'is_learner_eligible_for_one_click_purchase' , False )
# The following messages shouldn ' t both appear at the same time , and we prefer the eligibility message .
if not one_click_purchase_eligibility :
    messages . add_unenrollable_item_message ( request , 'program' )
elif discount_data . get ( 'total_incl_tax_excl_discounts' ) is None :
    messages . add_missing_price_information_message ( request , program_title )
context_data = get_global_context ( request , enterprise_customer )
context_data . update ( { 'enrolled_in_course_and_paid_text' : _ ( 'enrolled' ) , 'enrolled_in_course_and_unpaid_text' : _ ( 'already enrolled, must pay for certificate' ) , 'expected_learning_items_text' : _ ( "What you'll learn" ) , 'expected_learning_items_show_count' : 2 , 'corporate_endorsements_text' : _ ( 'Real Career Impact' ) , 'corporate_endorsements_show_count' : 1 , 'see_more_text' : _ ( 'See More' ) , 'see_less_text' : _ ( 'See Less' ) , 'confirm_button_text' : _ ( 'Confirm Program' ) , 'summary_header' : _ ( 'Program Summary' ) , 'price_text' : _ ( 'Price' ) , 'length_text' : _ ( 'Length' ) , 'effort_text' : _ ( 'Effort' ) , 'level_text' : _ ( 'Level' ) , 'course_full_description_text' : _ ( 'About This Course' ) , 'staff_text' : _ ( 'Course Staff' ) , 'close_modal_button_text' : _ ( 'Close' ) , 'program_not_eligible_for_one_click_purchase_text' : _ ( 'Program not eligible for one-click purchase.' ) , 'program_type_description_header' : _ ( 'What is an {platform_name} {program_type}?' ) . format ( platform_name = platform_name , program_type = program_type , ) , 'platform_description_header' : _ ( 'What is {platform_name}?' ) . format ( platform_name = platform_name ) , 'organization_name' : organization . get ( 'name' ) , 'organization_logo' : organization . get ( 'logo_image_url' ) , 'organization_text' : _ ( 'Presented by {organization}' ) . format ( organization = organization . get ( 'name' ) ) , 'page_title' : _ ( 'Confirm your {item}' ) . format ( item = item ) , 'program_type_logo' : program_type_details [ 'logo_image' ] . get ( 'medium' , { } ) . get ( 'url' , '' ) , 'program_type' : program_type , 'program_type_description' : get_program_type_description ( program_type ) , 'program_title' : program_title , 'program_subtitle' : program_details [ 'subtitle' ] , 'program_overview' : program_details [ 'overview' ] , 'program_price' : get_price_text ( discount_data . get ( 'total_incl_tax_excl_discounts' , 0 ) , request ) , 'program_discounted_price' : get_price_text ( discount_data . get ( 'total_incl_tax' , 0 ) , request ) , 'is_discounted' : discount_data . get ( 'is_discounted' , False ) , 'courses' : program_courses , 'item_bullet_points' : [ _ ( 'Credit- and Certificate-eligible' ) , _ ( 'Self-paced; courses can be taken in any order' ) , ] , 'purchase_text' : _ ( '{purchase_action} for' ) . format ( purchase_action = purchase_action ) , 'expected_learning_items' : program_details [ 'expected_learning_items' ] , 'corporate_endorsements' : program_details [ 'corporate_endorsements' ] , 'course_count_text' : course_count_text , 'length_info_text' : length_info_text , 'effort_info_text' : effort_info_text , 'is_learner_eligible_for_one_click_purchase' : one_click_purchase_eligibility , } )
return render ( request , 'enterprise/enterprise_program_enrollment_page.html' , context = context_data )


signature:
def get ( self , request , enterprise_uuid , program_uuid ) :
    """Show Program Landing page for the Enterprise ' s Program .
 Render the Enterprise ' s Program Enrollment page for a specific program .
 The Enterprise and Program are both selected by their respective UUIDs .
 Unauthenticated learners will be redirected to enterprise - linked SSO .
 A 404 will be raised if any of the following conditions are met :
 * No enterprise customer UUID query parameter ` ` enterprise _ uuid ` ` found in request .
 * No enterprise customer found against the enterprise customer
 uuid ` ` enterprise _ uuid ` ` in the request kwargs .
 * No Program can be found given ` ` program _ uuid ` ` either at all or associated with
 the Enterprise . ."""

implementation:
verify_edx_resources ( )
enterprise_customer = get_enterprise_customer_or_404 ( enterprise_uuid )
context_data = get_global_context ( request , enterprise_customer )
program_details , error_code = self . get_program_details ( request , program_uuid , enterprise_customer )
if error_code :
    return render ( request , ENTERPRISE_GENERAL_ERROR_PAGE , context = context_data , status = 404 , )
if program_details [ 'certificate_eligible_for_program' ] : # The user is already enrolled in the program , so redirect to the program ' s dashboard .
    return redirect ( LMS_PROGRAMS_DASHBOARD_URL . format ( uuid = program_uuid ) )
# Check to see if access to any of the course runs in the program are restricted for this user .
course_run_ids = [ ]
for course in program_details [ 'courses' ] :
    for course_run in course [ 'course_runs' ] :
        course_run_ids . append ( course_run [ 'key' ] )
embargo_url = EmbargoApiClient . redirect_if_blocked ( course_run_ids , request . user , get_ip ( request ) , request . path )
if embargo_url :
    return redirect ( embargo_url )
return self . get_enterprise_program_enrollment_page ( request , enterprise_customer , program_details )


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

implementation:
verify_edx_resources ( )
# Create a link between the user and the enterprise customer if it does not already exist .
enterprise_customer = get_enterprise_customer_or_404 ( enterprise_uuid )
with transaction . atomic ( ) :
    enterprise_customer_user , __ = EnterpriseCustomerUser . objects . get_or_create ( enterprise_customer = enterprise_customer , user_id = request . user . id )
    enterprise_customer_user . update_session ( request )
context_data = get_global_context ( request , enterprise_customer )
program_details , error_code = self . get_program_details ( request , program_uuid , enterprise_customer )
if error_code :
    return render ( request , ENTERPRISE_GENERAL_ERROR_PAGE , context = context_data , status = 404 , )
if program_details [ 'certificate_eligible_for_program' ] : # The user is already enrolled in the program , so redirect to the program ' s dashboard .
    return redirect ( LMS_PROGRAMS_DASHBOARD_URL . format ( uuid = program_uuid ) )
basket_page = '{basket_url}?{params}' . format ( basket_url = BASKET_URL , params = urlencode ( [ tuple ( [ 'sku' , sku ] ) for sku in program_details [ 'skus' ] ] + [ tuple ( [ 'bundle' , program_uuid ] ) ] ) )
if get_data_sharing_consent ( enterprise_customer_user . username , enterprise_customer . uuid , program_uuid = program_uuid , ) . consent_required ( ) :
    return redirect ( '{grant_data_sharing_url}?{params}' . format ( grant_data_sharing_url = reverse ( 'grant_data_sharing_permissions' ) , params = urlencode ( { 'next' : basket_page , 'failure_url' : reverse ( 'enterprise_program_enrollment_page' , args = [ enterprise_customer . uuid , program_uuid ] ) , 'enterprise_customer_uuid' : enterprise_customer . uuid , 'program_uuid' : program_uuid , } ) ) )
return redirect ( basket_page )


signature:
def get_path_variables ( ** kwargs ) :
    """Get the base variables for any view to route to .
 Currently gets :
 - ` enterprise _ uuid ` - the UUID of the enterprise customer .
 - ` course _ run _ id ` - the ID of the course , if applicable .
 - ` program _ uuid ` - the UUID of the program , if applicable ."""

implementation:
enterprise_customer_uuid = kwargs . get ( 'enterprise_uuid' , '' )
course_run_id = kwargs . get ( 'course_id' , '' )
course_key = kwargs . get ( 'course_key' , '' )
program_uuid = kwargs . get ( 'program_uuid' , '' )
return enterprise_customer_uuid , course_run_id , course_key , program_uuid


signature:
def get_course_run_id ( user , enterprise_customer , course_key ) :
    """User is requesting a course , we need to translate that into the current course run .
 : param user :
 : param enterprise _ customer :
 : param course _ key :
 : return : course _ run _ id"""

implementation:
try :
    course = CourseCatalogApiServiceClient ( enterprise_customer . site ) . get_course_details ( course_key )
except ImproperlyConfigured :
    raise Http404
users_all_enrolled_courses = EnrollmentApiClient ( ) . get_enrolled_courses ( user . username )
users_active_course_runs = get_active_course_runs ( course , users_all_enrolled_courses ) if users_all_enrolled_courses else [ ]
course_run = get_current_course_run ( course , users_active_course_runs )
if course_run :
    course_run_id = course_run [ 'key' ]
    return course_run_id
else :
    raise Http404


signature:
def eligible_for_direct_audit_enrollment ( self , request , enterprise_customer , resource_id , course_key = None ) :
    """Return whether a request is eligible for direct audit enrollment for a particular enterprise customer .
 ' resource _ id ' can be either course _ run _ id or program _ uuid .
 We check for the following criteria :
 - The ` audit ` query parameter .
 - The user ' s being routed to the course enrollment landing page .
 - The customer ' s catalog contains the course in question .
 - The audit track is an available mode for the course ."""

implementation:
course_identifier = course_key if course_key else resource_id
# Return it in one big statement to utilize short - circuiting behavior . Avoid the API call if possible .
return request . GET . get ( 'audit' ) and request . path == self . COURSE_ENROLLMENT_VIEW_URL . format ( enterprise_customer . uuid , course_identifier ) and enterprise_customer . catalog_contains_course ( resource_id ) and EnrollmentApiClient ( ) . has_course_mode ( resource_id , 'audit' )


signature:
def redirect ( self , request , * args , ** kwargs ) :
    """Redirects to the appropriate view depending on where the user came from ."""

implementation:
enterprise_customer_uuid , course_run_id , course_key , program_uuid = RouterView . get_path_variables ( ** kwargs )
resource_id = course_key or course_run_id or program_uuid
# Replace enterprise UUID and resource ID with ' { } ' , to easily match with a path in RouterView . VIEWS . Example :
# / enterprise / fake - uuid / course / course - v1 : cool + course + 2017 / enroll / - > / enterprise / { } / course / { } / enroll /
path = re . sub ( '{}|{}' . format ( enterprise_customer_uuid , re . escape ( resource_id ) ) , '{}' , request . path )
# Remove course _ key from kwargs if it exists because delegate views are not expecting it .
kwargs . pop ( 'course_key' , None )
return self . VIEWS [ path ] . as_view ( ) ( request , * args , ** kwargs )


signature:
def get ( self , request , * args , ** kwargs ) :
    """Run some custom GET logic for Enterprise workflows before routing the user through existing views .
 In particular , before routing to existing views :
 - If the requested resource is a course , find the current course run for that course ,
 and make that course run the requested resource instead .
 - Look to see whether a request is eligible for direct audit enrollment , and if so , directly enroll the user ."""

implementation:
enterprise_customer_uuid , course_run_id , course_key , program_uuid = RouterView . get_path_variables ( ** kwargs )
enterprise_customer = get_enterprise_customer_or_404 ( enterprise_customer_uuid )
if course_key :
    try :
        course_run_id = RouterView . get_course_run_id ( request . user , enterprise_customer , course_key )
    except Http404 :
        context_data = get_global_context ( request , enterprise_customer )
        error_code = 'ENTRV000'
        log_message = ( 'Could not find course run with id {course_run_id} ' 'for course key {course_key} and program_uuid {program_uuid} ' 'for enterprise_customer_uuid {enterprise_customer_uuid} ' 'Returned error code {error_code} to user {userid}' . format ( course_key = course_key , course_run_id = course_run_id , enterprise_customer_uuid = enterprise_customer_uuid , error_code = error_code , userid = request . user . id , program_uuid = program_uuid , ) )
        return render_page_with_error_code_message ( request , context_data , error_code , log_message )
    kwargs [ 'course_id' ] = course_run_id
# Ensure that the link is saved to the database prior to making some call in a downstream view
# which may need to know that the user belongs to an enterprise customer .
with transaction . atomic ( ) :
    enterprise_customer_user , __ = EnterpriseCustomerUser . objects . get_or_create ( enterprise_customer = enterprise_customer , user_id = request . user . id )
    enterprise_customer_user . update_session ( request )
# Directly enroll in audit mode if the request in question has full direct audit enrollment eligibility .
resource_id = course_run_id or program_uuid
if self . eligible_for_direct_audit_enrollment ( request , enterprise_customer , resource_id , course_key ) :
    try :
        enterprise_customer_user . enroll ( resource_id , 'audit' , cohort = request . GET . get ( 'cohort' , None ) )
        track_enrollment ( 'direct-audit-enrollment' , request . user . id , resource_id , request . get_full_path ( ) )
    except ( CourseEnrollmentDowngradeError , CourseEnrollmentPermissionError ) :
        pass
    # The courseware view logic will check for DSC requirements , and route to the DSC page if necessary .
    return redirect ( LMS_COURSEWARE_URL . format ( course_id = resource_id ) )
return self . redirect ( request , * args , ** kwargs )


signature:
def post ( self , request , * args , ** kwargs ) :
    """Run some custom POST logic for Enterprise workflows before routing the user through existing views ."""

implementation:
# pylint : disable = unused - variable
enterprise_customer_uuid , course_run_id , course_key , program_uuid = RouterView . get_path_variables ( ** kwargs )
enterprise_customer = get_enterprise_customer_or_404 ( enterprise_customer_uuid )
if course_key :
    context_data = get_global_context ( request , enterprise_customer )
    try :
        kwargs [ 'course_id' ] = RouterView . get_course_run_id ( request . user , enterprise_customer , course_key )
    except Http404 :
        error_code = 'ENTRV001'
        log_message = ( 'Could not find course run with id {course_run_id} ' 'for course key {course_key} and ' 'for enterprise_customer_uuid {enterprise_customer_uuid} ' 'and program {program_uuid}. ' 'Returned error code {error_code} to user {userid}' . format ( course_key = course_key , course_run_id = course_run_id , enterprise_customer_uuid = enterprise_customer_uuid , error_code = error_code , userid = request . user . id , program_uuid = program_uuid , ) )
        return render_page_with_error_code_message ( request , context_data , error_code , log_message )
return self . redirect ( request , * args , ** kwargs )


signature:
def transmit_content_metadata ( username , channel_code , channel_pk ) :
    """Task to send content metadata to each linked integrated channel .
 Arguments :
 username ( str ) : The username of the User to be used for making API requests to retrieve content metadata .
 channel _ code ( str ) : Capitalized identifier for the integrated channel .
 channel _ pk ( str ) : Primary key for identifying integrated channel ."""

implementation:
start = time . time ( )
api_user = User . objects . get ( username = username )
integrated_channel = INTEGRATED_CHANNEL_CHOICES [ channel_code ] . objects . get ( pk = channel_pk )
LOGGER . info ( 'Transmitting content metadata to integrated channel using configuration: [%s]' , integrated_channel )
try :
    integrated_channel . transmit_content_metadata ( api_user )
except Exception : # pylint : disable = broad - except
    LOGGER . exception ( 'Transmission of content metadata failed for user [%s] and for integrated ' 'channel with code [%s] and id [%s].' , username , channel_code , channel_pk )
duration = time . time ( ) - start
LOGGER . info ( 'Content metadata transmission task for integrated channel configuration [%s] took [%s] seconds' , integrated_channel , duration )


signature:
def transmit_learner_data ( username , channel_code , channel_pk ) :
    """Task to send learner data to each linked integrated channel .
 Arguments :
 username ( str ) : The username of the User to be used for making API requests for learner data .
 channel _ code ( str ) : Capitalized identifier for the integrated channel
 channel _ pk ( str ) : Primary key for identifying integrated channel"""

implementation:
start = time . time ( )
api_user = User . objects . get ( username = username )
integrated_channel = INTEGRATED_CHANNEL_CHOICES [ channel_code ] . objects . get ( pk = channel_pk )
LOGGER . info ( 'Processing learners for integrated channel using configuration: [%s]' , integrated_channel )
# Note : learner data transmission code paths don ' t raise any uncaught exception , so we don ' t need a broad
# try - except block here .
integrated_channel . transmit_learner_data ( api_user )
duration = time . time ( ) - start
LOGGER . info ( 'Learner data transmission task for integrated channel configuration [%s] took [%s] seconds' , integrated_channel , duration )


signature:
def unlink_inactive_learners ( channel_code , channel_pk ) :
    """Task to unlink inactive learners of provided integrated channel .
 Arguments :
 channel _ code ( str ) : Capitalized identifier for the integrated channel
 channel _ pk ( str ) : Primary key for identifying integrated channel"""

implementation:
start = time . time ( )
integrated_channel = INTEGRATED_CHANNEL_CHOICES [ channel_code ] . objects . get ( pk = channel_pk )
LOGGER . info ( 'Processing learners to unlink inactive users using configuration: [%s]' , integrated_channel )
# Note : learner data transmission code paths don ' t raise any uncaught exception , so we don ' t need a broad
# try - except block here .
integrated_channel . unlink_inactive_learners ( )
duration = time . time ( ) - start
LOGGER . info ( 'Unlink inactive learners task for integrated channel configuration [%s] took [%s] seconds' , integrated_channel , duration )


signature:
def handle_user_post_save ( sender , ** kwargs ) : # pylint : disable = unused - argument
    """Handle User model changes - checks if pending enterprise customer user record exists and upgrades it to actual link .
 If there are pending enrollments attached to the PendingEnterpriseCustomerUser , then this signal also takes the
 newly - created users and enrolls them in the relevant courses ."""

implementation:
created = kwargs . get ( "created" , False )
user_instance = kwargs . get ( "instance" , None )
if user_instance is None :
    return
# should never happen , but better safe than 500 error
try :
    pending_ecu = PendingEnterpriseCustomerUser . objects . get ( user_email = user_instance . email )
except PendingEnterpriseCustomerUser . DoesNotExist :
    return
# nothing to do in this case
if not created : # existing user changed his email to match one of pending link records - try linking him to EC
    try :
        existing_record = EnterpriseCustomerUser . objects . get ( user_id = user_instance . id )
        message_template = "User {user} have changed email to match pending Enterprise Customer link, " "but was already linked to Enterprise Customer {enterprise_customer} - " "deleting pending link record"
        logger . info ( message_template . format ( user = user_instance , enterprise_customer = existing_record . enterprise_customer ) )
        pending_ecu . delete ( )
        return
    except EnterpriseCustomerUser . DoesNotExist :
        pass
# everything ok - current user is not linked to other ECs
enterprise_customer_user = EnterpriseCustomerUser . objects . create ( enterprise_customer = pending_ecu . enterprise_customer , user_id = user_instance . id )
pending_enrollments = list ( pending_ecu . pendingenrollment_set . all ( ) )
if pending_enrollments :
    def _complete_user_enrollment ( ) : # pylint : disable = missing - docstring
        for enrollment in pending_enrollments : # EnterpriseCustomers may enroll users in courses before the users themselves
        # actually exist in the system ; in such a case , the enrollment for each such
        # course is finalized when the user registers with the OpenEdX platform .
            enterprise_customer_user . enroll ( enrollment . course_id , enrollment . course_mode , cohort = enrollment . cohort_name )
            track_enrollment ( 'pending-admin-enrollment' , user_instance . id , enrollment . course_id )
        pending_ecu . delete ( )
    transaction . on_commit ( _complete_user_enrollment )
else :
    pending_ecu . delete ( )


signature:
def default_content_filter ( sender , instance , ** kwargs ) : # pylint : disable = unused - argument
    """Set default value for ` EnterpriseCustomerCatalog . content _ filter ` if not already set ."""

implementation:
if kwargs [ 'created' ] and not instance . content_filter :
    instance . content_filter = get_default_catalog_content_filter ( )
    instance . save ( )


signature:
def assign_enterprise_learner_role ( sender , instance , ** kwargs ) : # pylint : disable = unused - argument
    """Assign an enterprise learner role to EnterpriseCustomerUser whenever a new record is created ."""

implementation:
if kwargs [ 'created' ] and instance . user :
    enterprise_learner_role , __ = SystemWideEnterpriseRole . objects . get_or_create ( name = ENTERPRISE_LEARNER_ROLE )
    SystemWideEnterpriseUserRoleAssignment . objects . get_or_create ( user = instance . user , role = enterprise_learner_role )


signature:
def delete_enterprise_learner_role_assignment ( sender , instance , ** kwargs ) : # pylint : disable = unused - argument
    """Delete the associated enterprise learner role assignment record when deleting an EnterpriseCustomerUser record ."""

implementation:
if instance . user :
    enterprise_learner_role , __ = SystemWideEnterpriseRole . objects . get_or_create ( name = ENTERPRISE_LEARNER_ROLE )
    try :
        SystemWideEnterpriseUserRoleAssignment . objects . get ( user = instance . user , role = enterprise_learner_role ) . delete ( )
    except SystemWideEnterpriseUserRoleAssignment . DoesNotExist : # Do nothing if no role assignment is present for the enterprise customer user .
        pass


signature:
def handle ( self , * args , ** options ) :
    """Transmit the courseware data for the EnterpriseCustomer ( s ) to the active integration channels ."""

implementation:
username = options [ 'catalog_user' ]
# Before we do a whole bunch of database queries , make sure that the user we were passed exists .
try :
    User . objects . get ( username = username )
except User . DoesNotExist :
    raise CommandError ( 'A user with the username {} was not found.' . format ( username ) )
channels = self . get_integrated_channels ( options )
for channel in channels :
    channel_code = channel . channel_code ( )
    channel_pk = channel . pk
    transmit_content_metadata . delay ( username , channel_code , channel_pk )


signature:
def create_roles ( apps , schema_editor ) :
    """Create the enterprise roles if they do not already exist ."""

implementation:
SystemWideEnterpriseRole = apps . get_model ( 'enterprise' , 'SystemWideEnterpriseRole' )
SystemWideEnterpriseRole . objects . update_or_create ( name = ENTERPRISE_ADMIN_ROLE )
SystemWideEnterpriseRole . objects . update_or_create ( name = ENTERPRISE_LEARNER_ROLE )


signature:
def delete_roles ( apps , schema_editor ) :
    """Delete the enterprise roles ."""

implementation:
SystemWideEnterpriseRole = apps . get_model ( 'enterprise' , 'SystemWideEnterpriseRole' )
SystemWideEnterpriseRole . objects . filter ( name__in = [ ENTERPRISE_ADMIN_ROLE , ENTERPRISE_LEARNER_ROLE ] ) . delete ( )


signature:
def _get_enterprise_admin_users_batch ( self , start , end ) :
    """Returns a batched queryset of User objects ."""

implementation:
LOGGER . info ( 'Fetching new batch of enterprise admin users from indexes: %s to %s' , start , end )
return User . objects . filter ( groups__name = ENTERPRISE_DATA_API_ACCESS_GROUP , is_staff = False ) [ start : end ]


signature:
def _get_enterprise_operator_users_batch ( self , start , end ) :
    """Returns a batched queryset of User objects ."""

implementation:
LOGGER . info ( 'Fetching new batch of enterprise operator users from indexes: %s to %s' , start , end )
return User . objects . filter ( groups__name = ENTERPRISE_DATA_API_ACCESS_GROUP , is_staff = True ) [ start : end ]


signature:
def _get_enterprise_customer_users_batch ( self , start , end ) :
    """Returns a batched queryset of EnterpriseCustomerUser objects ."""

implementation:
LOGGER . info ( 'Fetching new batch of enterprise customer users from indexes: %s to %s' , start , end )
return User . objects . filter ( pk__in = self . _get_enterprise_customer_user_ids ( ) ) [ start : end ]


signature:
def _get_enterprise_enrollment_api_admin_users_batch ( self , start , end ) : # pylint : disable = invalid - name
    """Returns a batched queryset of User objects ."""

implementation:
LOGGER . info ( 'Fetching new batch of enterprise enrollment admin users from indexes: %s to %s' , start , end )
return User . objects . filter ( groups__name = ENTERPRISE_ENROLLMENT_API_ACCESS_GROUP , is_staff = False ) [ start : end ]


signature:
def _get_enterprise_catalog_admin_users_batch ( self , start , end ) :
    """Returns a batched queryset of User objects ."""

implementation:
Application = apps . get_model ( OAUTH2_PROVIDER_APPLICATION_MODEL )
# pylint : disable = invalid - name
LOGGER . info ( 'Fetching new batch of enterprise catalog admin users from indexes: %s to %s' , start , end )
catalog_admin_user_ids = Application . objects . filter ( user_id__in = self . _get_enterprise_customer_user_ids ( ) ) . exclude ( name = EDX_ORG_NAME ) . values ( 'user_id' )
return User . objects . filter ( pk__in = catalog_admin_user_ids ) [ start : end ]


signature:
def _assign_enterprise_role_to_users ( self , _get_batch_method , options , is_feature_role = False ) :
    """Assigns enterprise role to users ."""

implementation:
role_name = options [ 'role' ]
batch_limit = options [ 'batch_limit' ]
batch_sleep = options [ 'batch_sleep' ]
batch_offset = options [ 'batch_offset' ]
current_batch_index = batch_offset
users_batch = _get_batch_method ( batch_offset , batch_offset + batch_limit )
role_class = SystemWideEnterpriseRole
role_assignment_class = SystemWideEnterpriseUserRoleAssignment
if is_feature_role :
    role_class = EnterpriseFeatureRole
    role_assignment_class = EnterpriseFeatureUserRoleAssignment
enterprise_role = role_class . objects . get ( name = role_name )
while users_batch . count ( ) > 0 :
    for index , user in enumerate ( users_batch ) :
        LOGGER . info ( 'Processing user with index %s and id %s' , current_batch_index + index , user . id )
        role_assignment_class . objects . get_or_create ( user = user , role = enterprise_role )
    sleep ( batch_sleep )
    current_batch_index += len ( users_batch )
    users_batch = _get_batch_method ( current_batch_index , current_batch_index + batch_limit )


signature:
def handle ( self , * args , ** options ) :
    """Entry point for managment command execution ."""

implementation:
LOGGER . info ( 'Starting assigning enterprise roles to users!' )
role = options [ 'role' ]
if role == ENTERPRISE_ADMIN_ROLE : # Assign admin role to non - staff users with enterprise data api access .
    self . _assign_enterprise_role_to_users ( self . _get_enterprise_admin_users_batch , options )
elif role == ENTERPRISE_OPERATOR_ROLE : # Assign operator role to staff users with enterprise data api access .
    self . _assign_enterprise_role_to_users ( self . _get_enterprise_operator_users_batch , options )
elif role == ENTERPRISE_LEARNER_ROLE : # Assign enterprise learner role to enterprise customer users .
    self . _assign_enterprise_role_to_users ( self . _get_enterprise_customer_users_batch , options )
elif role == ENTERPRISE_ENROLLMENT_API_ADMIN_ROLE : # Assign enterprise enrollment api admin to non - staff users with enterprise data api access .
    self . _assign_enterprise_role_to_users ( self . _get_enterprise_enrollment_api_admin_users_batch , options , True )
elif role == ENTERPRISE_CATALOG_ADMIN_ROLE : # Assign enterprise catalog admin role to users with having credentials in catalog .
    self . _assign_enterprise_role_to_users ( self . _get_enterprise_catalog_admin_users_batch , options , True )
else :
    raise CommandError ( 'Please provide a valid role name. Supported roles are {admin} and {learner}' . format ( admin = ENTERPRISE_ADMIN_ROLE , learner = ENTERPRISE_LEARNER_ROLE ) )
LOGGER . info ( 'Successfully finished assigning enterprise roles to users!' )


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

implementation:
kwargs [ 'app_label' ] = 'degreed'
kwargs [ 'model_name' ] = 'DegreedLearnerDataTransmissionAudit'
kwargs [ 'remote_user_id' ] = 'degreed_user_email'
super ( DegreedLearnerTransmitter , self ) . transmit ( payload , ** kwargs )


signature:
def get_enterprise_customer_for_running_pipeline ( request , pipeline ) : # pylint : disable = invalid - name
    """Get the EnterpriseCustomer associated with a running pipeline ."""

implementation:
sso_provider_id = request . GET . get ( 'tpa_hint' )
if pipeline :
    sso_provider_id = Registry . get_from_pipeline ( pipeline ) . provider_id
return get_enterprise_customer_for_sso ( sso_provider_id )


signature:
def handle_enterprise_logistration ( backend , user , ** kwargs ) :
    """Perform the linking of user in the process of logging to the Enterprise Customer .
 Args :
 backend : The class handling the SSO interaction ( SAML , OAuth , etc )
 user : The user object in the process of being logged in with
 * * kwargs : Any remaining pipeline variables"""

implementation:
request = backend . strategy . request
enterprise_customer = get_enterprise_customer_for_running_pipeline ( request , { 'backend' : backend . name , 'kwargs' : kwargs } )
if enterprise_customer is None : # This pipeline element is not being activated as a part of an Enterprise logistration
    return
# proceed with the creation of a link between the user and the enterprise customer , then exit .
enterprise_customer_user , _ = EnterpriseCustomerUser . objects . update_or_create ( enterprise_customer = enterprise_customer , user_id = user . id )
enterprise_customer_user . update_session ( request )


signature:
def get_user_from_social_auth ( tpa_provider , tpa_username ) :
    """Find the LMS user from the LMS model ` UserSocialAuth ` .
 Arguments :
 tpa _ provider ( third _ party _ auth . provider ) : third party auth provider object
 tpa _ username ( str ) : Username returned by the third party auth"""

implementation:
user_social_auth = UserSocialAuth . objects . select_related ( 'user' ) . filter ( user__username = tpa_username , provider = tpa_provider . backend_name ) . first ( )
return user_social_auth . user if user_social_auth else None


signature:
def get_oauth_access_token ( url_base , client_id , client_secret , company_id , user_id , user_type ) :
    """Retrieves OAuth 2.0 access token using the client credentials grant .
 Args :
 url _ base ( str ) : Oauth2 access token endpoint
 client _ id ( str ) : client ID
 client _ secret ( str ) : client secret
 company _ id ( str ) : SAP company ID
 user _ id ( str ) : SAP user ID
 user _ type ( str ) : type of SAP user ( admin or user )
 Returns :
 tuple : Tuple containing access token string and expiration datetime .
 Raises :
 HTTPError : If we received a failure response code from SAP SuccessFactors .
 RequestException : If an unexpected response format was received that we could not parse ."""

implementation:
SAPSuccessFactorsGlobalConfiguration = apps . get_model ( # pylint : disable = invalid - name
'sap_success_factors' , 'SAPSuccessFactorsGlobalConfiguration' )
global_sap_config = SAPSuccessFactorsGlobalConfiguration . current ( )
url = url_base + global_sap_config . oauth_api_path
response = requests . post ( url , json = { 'grant_type' : 'client_credentials' , 'scope' : { 'userId' : user_id , 'companyId' : company_id , 'userType' : user_type , 'resourceType' : 'learning_public_api' , } } , auth = ( client_id , client_secret ) , headers = { 'content-type' : 'application/json' } )
response . raise_for_status ( )
data = response . json ( )
try :
    return data [ 'access_token' ] , datetime . datetime . utcfromtimestamp ( data [ 'expires_in' ] + int ( time . time ( ) ) )
except KeyError :
    raise requests . RequestException ( response = response )


signature:
def _create_session ( self ) :
    """Instantiate a new session object for use in connecting with SAP SuccessFactors"""

implementation:
session = requests . Session ( )
session . timeout = self . SESSION_TIMEOUT
oauth_access_token , expires_at = SAPSuccessFactorsAPIClient . get_oauth_access_token ( self . enterprise_configuration . sapsf_base_url , self . enterprise_configuration . key , self . enterprise_configuration . secret , self . enterprise_configuration . sapsf_company_id , self . enterprise_configuration . sapsf_user_id , self . enterprise_configuration . user_type )
session . headers [ 'Authorization' ] = 'Bearer {}' . format ( oauth_access_token )
session . headers [ 'content-type' ] = 'application/json'
self . session = session
self . expires_at = expires_at


signature:
def create_course_completion ( self , user_id , payload ) :
    """Send a completion status payload to the SuccessFactors OCN Completion Status endpoint
 Args :
 user _ id ( str ) : The sap user id that the completion status is being sent for .
 payload ( str ) : JSON encoded object ( serialized from SapSuccessFactorsLearnerDataTransmissionAudit )
 containing completion status fields per SuccessFactors documentation .
 Returns :
 The body of the response from SAP SuccessFactors , if successful
 Raises :
 HTTPError : if we received a failure response code from SAP SuccessFactors"""

implementation:
url = self . enterprise_configuration . sapsf_base_url + self . global_sap_config . completion_status_api_path
return self . _call_post_with_user_override ( user_id , url , payload )


signature:
def _sync_content_metadata ( self , serialized_data ) :
    """Create / update / delete content metadata records using the SuccessFactors OCN Course Import API endpoint .
 Arguments :
 serialized _ data : Serialized JSON string representing a list of content metadata items .
 Raises :
 ClientError : If SuccessFactors API call fails ."""

implementation:
url = self . enterprise_configuration . sapsf_base_url + self . global_sap_config . course_api_path
try :
    status_code , response_body = self . _call_post_with_session ( url , serialized_data )
except requests . exceptions . RequestException as exc :
    raise ClientError ( 'SAPSuccessFactorsAPIClient request failed: {error} {message}' . format ( error = exc . __class__ . __name__ , message = str ( exc ) ) )
if status_code >= 400 :
    raise ClientError ( 'SAPSuccessFactorsAPIClient request failed with status {status_code}: {message}' . format ( status_code = status_code , message = response_body ) )


signature:
def _call_post_with_user_override ( self , sap_user_id , url , payload ) :
    """Make a post request with an auth token acquired for a specific user to a SuccessFactors endpoint .
 Args :
 sap _ user _ id ( str ) : The user to use to retrieve an auth token .
 url ( str ) : The url to post to .
 payload ( str ) : The json encoded payload to post ."""

implementation:
SAPSuccessFactorsEnterpriseCustomerConfiguration = apps . get_model ( # pylint : disable = invalid - name
'sap_success_factors' , 'SAPSuccessFactorsEnterpriseCustomerConfiguration' )
oauth_access_token , _ = SAPSuccessFactorsAPIClient . get_oauth_access_token ( self . enterprise_configuration . sapsf_base_url , self . enterprise_configuration . key , self . enterprise_configuration . secret , self . enterprise_configuration . sapsf_company_id , sap_user_id , SAPSuccessFactorsEnterpriseCustomerConfiguration . USER_TYPE_USER )
response = requests . post ( url , data = payload , headers = { 'Authorization' : 'Bearer {}' . format ( oauth_access_token ) , 'content-type' : 'application/json' } )
return response . status_code , response . text


signature:
def _call_post_with_session ( self , url , payload ) :
    """Make a post request using the session object to a SuccessFactors endpoint .
 Args :
 url ( str ) : The url to post to .
 payload ( str ) : The json encoded payload to post ."""

implementation:
now = datetime . datetime . utcnow ( )
if now >= self . expires_at : # Create a new session with a valid token
    self . session . close ( )
    self . _create_session ( )
response = self . session . post ( url , data = payload )
return response . status_code , response . text


signature:
def get_inactive_sap_learners ( self ) :
    """Make a GET request using the session object to a SuccessFactors endpoint for inactive learners .
 Example :
 sap _ search _ student _ url : " / learning / odatav4 / searchStudent / v1 / Students ?
 $ filter = criteria / isActive eq False & $ select = studentID "
 SAP API response : {
 u ' @ odata . metadataEtag ' : u ' W / " 17090d86-20fa - 49c8-8de0 - de1d308c8b55 " ' ,
 u ' value ' : [
 u ' studentID ' : u ' admint6 ' ,
 u ' studentID ' : u ' adminsap1 ' ,
 Returns : List of inactive learners
 u ' studentID ' : u ' admint6'
 u ' studentID ' : u ' adminsap1'"""

implementation:
now = datetime . datetime . utcnow ( )
if now >= self . expires_at : # Create a new session with a valid token
    self . session . close ( )
    self . _create_session ( )
sap_search_student_url = '{sapsf_base_url}/{search_students_path}?$filter={search_filter}' . format ( sapsf_base_url = self . enterprise_configuration . sapsf_base_url . rstrip ( '/' ) , search_students_path = self . global_sap_config . search_student_api_path . rstrip ( '/' ) , search_filter = 'criteria/isActive eq False&$select=studentID' , )
all_inactive_learners = self . _call_search_students_recursively ( sap_search_student_url , all_inactive_learners = [ ] , page_size = 500 , start_at = 0 )
return all_inactive_learners


signature:
def _call_search_students_recursively ( self , sap_search_student_url , all_inactive_learners , page_size , start_at ) :
    """Make recursive GET calls to traverse the paginated API response for search students ."""

implementation:
search_student_paginated_url = '{sap_search_student_url}&{pagination_criterion}' . format ( sap_search_student_url = sap_search_student_url , pagination_criterion = '$count=true&$top={page_size}&$skip={start_at}' . format ( page_size = page_size , start_at = start_at , ) , )
try :
    response = self . session . get ( search_student_paginated_url )
    sap_inactive_learners = response . json ( )
except ( ConnectionError , Timeout ) :
    LOGGER . warning ( 'Unable to fetch inactive learners from SAP searchStudent API with url ' '"{%s}".' , search_student_paginated_url , )
    return None
if 'error' in sap_inactive_learners :
    LOGGER . warning ( 'SAP searchStudent API for customer %s and base url %s returned response with ' 'error message "%s" and with error code "%s".' , self . enterprise_configuration . enterprise_customer . name , self . enterprise_configuration . sapsf_base_url , sap_inactive_learners [ 'error' ] . get ( 'message' ) , sap_inactive_learners [ 'error' ] . get ( 'code' ) , )
    return None
new_page_start_at = page_size + start_at
all_inactive_learners += sap_inactive_learners [ 'value' ]
if sap_inactive_learners [ '@odata.count' ] > new_page_start_at :
    return self . _call_search_students_recursively ( sap_search_student_url , all_inactive_learners , page_size = page_size , start_at = new_page_start_at , )
return all_inactive_learners


signature:
def filter_queryset ( self , request , queryset , view ) :
    """Filter only for the user ' s ID if non - staff ."""

implementation:
if not request . user . is_staff :
    filter_kwargs = { view . USER_ID_FILTER : request . user . id }
    queryset = queryset . filter ( ** filter_kwargs )
return queryset


signature:
def filter_queryset ( self , request , queryset , view ) :
    """Apply incoming filters only if user is staff . If not , only filter by user ' s ID ."""

implementation:
if request . user . is_staff :
    email = request . query_params . get ( 'email' , None )
    username = request . query_params . get ( 'username' , None )
    query_parameters = { }
    if email :
        query_parameters . update ( email = email )
    if username :
        query_parameters . update ( username = username )
    if query_parameters :
        users = User . objects . filter ( ** query_parameters ) . values_list ( 'id' , flat = True )
        queryset = queryset . filter ( user_id__in = users )
else :
    queryset = queryset . filter ( user_id = request . user . id )
return queryset


signature:
def transmit ( self , payload , ** kwargs ) :
    """Send a completion status call to the integrated channel using the client .
 Args :
 payload : The learner completion data payload to send to the integrated channel .
 kwargs : Contains integrated channel - specific information for customized transmission variables .
 - app _ label : The app label of the integrated channel for whom to store learner data records for .
 - model _ name : The name of the specific learner data record model to use .
 - remote _ user _ id : The remote ID field name of the learner on the audit model ."""

implementation:
IntegratedChannelLearnerDataTransmissionAudit = apps . get_model ( # pylint : disable = invalid - name
app_label = kwargs . get ( 'app_label' , 'integrated_channel' ) , model_name = kwargs . get ( 'model_name' , 'LearnerDataTransmissionAudit' ) , )
# Since we have started sending courses to integrated channels instead of course runs ,
# we need to attempt to send transmissions with course keys and course run ids in order to
# ensure that we account for whether courses or course runs exist in the integrated channel .
# The exporters have been changed to return multiple transmission records to attempt ,
# one by course key and one by course run id .
# If the transmission with the course key succeeds , the next one will get skipped .
# If it fails , the one with the course run id will be attempted and ( presumably ) succeed .
for learner_data in payload . export ( ) :
    serialized_payload = learner_data . serialize ( enterprise_configuration = self . enterprise_configuration )
    LOGGER . debug ( 'Attempting to transmit serialized payload: %s' , serialized_payload )
    enterprise_enrollment_id = learner_data . enterprise_course_enrollment_id
    if learner_data . completed_timestamp is None : # The user has not completed the course , so we shouldn ' t send a completion status call
        LOGGER . info ( 'Skipping in-progress enterprise enrollment {}' . format ( enterprise_enrollment_id ) )
        continue
    previous_transmissions = IntegratedChannelLearnerDataTransmissionAudit . objects . filter ( enterprise_course_enrollment_id = enterprise_enrollment_id , error_message = '' )
    if previous_transmissions . exists ( ) : # We ' ve already sent a completion status call for this enrollment
        LOGGER . info ( 'Skipping previously sent enterprise enrollment {}' . format ( enterprise_enrollment_id ) )
        continue
    try :
        code , body = self . client . create_course_completion ( getattr ( learner_data , kwargs . get ( 'remote_user_id' ) ) , serialized_payload )
        LOGGER . info ( 'Successfully sent completion status call for enterprise enrollment {}' . format ( enterprise_enrollment_id , ) )
    except RequestException as request_exception :
        code = 500
        body = str ( request_exception )
        self . handle_transmission_error ( learner_data , request_exception )
    learner_data . status = str ( code )
    learner_data . error_message = body if code >= 400 else ''
    learner_data . save ( )


signature:
def handle_transmission_error ( self , learner_data , request_exception ) :
    """Handle the case where the transmission fails ."""

implementation:
try :
    sys_msg = request_exception . response . content
except AttributeError :
    sys_msg = 'Not available'
LOGGER . error ( ( 'Failed to send completion status call for enterprise enrollment %s' 'with payload %s' '\nError message: %s' '\nSystem message: %s' ) , learner_data . enterprise_course_enrollment_id , learner_data , str ( request_exception ) , sys_msg )


signature:
def add_missing_price_information_message ( request , item ) :
    """Add a message to the Django messages store indicating that we failed to retrieve price information about an item .
 : param request : The current request .
 : param item : The item for which price information is missing . Example : a program title , or a course ."""

implementation: messages . warning ( request , _ ( '{strong_start}We could not gather price information for {em_start}{item}{em_end}.{strong_end} ' '{span_start}If you continue to have these issues, please contact ' '{link_start}{platform_name} support{link_end}.{span_end}' ) . format ( item = item , em_start = '' , em_end = '' , link_start = '' . format ( support_link = get_configuration_value ( 'ENTERPRISE_SUPPORT_URL' , settings . ENTERPRISE_SUPPORT_URL ) , ) , platform_name = get_configuration_value ( 'PLATFORM_NAME' , settings . PLATFORM_NAME ) , link_end = '' , span_start = '' , span_end = '' , strong_start = '' , strong_end = '' , ) )


signature:
def add_unenrollable_item_message ( request , item ) :
    """Add a message to the Django message store indicating that the item ( i . e . course run , program ) is unenrollable .
 : param request : The current request .
 : param item : The item that is unenrollable ( i . e . a course run ) ."""

implementation: messages . info ( request , _ ( '{strong_start}Something happened.{strong_end} ' '{span_start}This {item} is not currently open to new learners. Please start over and select a different ' '{item}.{span_end}' ) . format ( item = item , strong_start = '' , strong_end = '' , span_start = '' , span_end = '' , ) )


signature:
def add_generic_info_message_for_error ( request ) :
    """Add message to request indicating that there was an issue processing request .
 Arguments :
 request : The current request ."""

implementation: messages . info ( request , _ ( '{strong_start}Something happened.{strong_end} ' '{span_start}This course link is currently invalid. ' 'Please reach out to your Administrator for assistance to this course.{span_end}' ) . format ( span_start = '' , span_end = '' , strong_start = '' , strong_end = '' , ) )


signature:
def add_generic_error_message_with_code ( request , error_code ) :
    """Add message to request indicating that there was an issue processing request .
 Arguments :
 request : The current request .
 error _ code : A string error code to be used to point devs to the spot in
 the code where this error occurred ."""

implementation: messages . error ( request , _ ( '{strong_start}Something happened.{strong_end} ' '{span_start}Please reach out to your learning administrator with ' 'the following error code and they will be able to help you out.{span_end}' '{span_start}Error code: {error_code}{span_end}' ) . format ( error_code = error_code , strong_start = '' , strong_end = '' , span_start = '' , span_end = '' , ) )


signature:
def validate_image_extension ( value ) :
    """Validate that a particular image extension ."""

implementation:
config = get_app_config ( )
ext = os . path . splitext ( value . name ) [ 1 ]
if config and not ext . lower ( ) in config . valid_image_extensions :
    raise ValidationError ( _ ( "Unsupported file extension." ) )


signature:
def validate_image_size ( image ) :
    """Validate that a particular image size ."""

implementation:
config = get_app_config ( )
valid_max_image_size_in_bytes = config . valid_max_image_size * 1024
if config and not image . size <= valid_max_image_size_in_bytes :
    raise ValidationError ( _ ( "The logo image file size must be less than or equal to %s KB." ) % config . valid_max_image_size )


signature:
def get_enterprise_customer_from_catalog_id ( catalog_id ) :
    """Get the enterprise customer id given an enterprise customer catalog id ."""

implementation:
try :
    return str ( EnterpriseCustomerCatalog . objects . get ( pk = catalog_id ) . enterprise_customer . uuid )
except EnterpriseCustomerCatalog . DoesNotExist :
    return None


signature:
def on_init ( app ) : # pylint : disable = unused - argument
    """Run sphinx - apidoc after Sphinx initialization .
 Read the Docs won ' t run tox or custom shell commands , so we need this to
 avoid checking in the generated reStructuredText files ."""

implementation:
docs_path = os . path . abspath ( os . path . dirname ( __file__ ) )
root_path = os . path . abspath ( os . path . join ( docs_path , '..' ) )
apidoc_path = 'sphinx-apidoc'
if hasattr ( sys , 'real_prefix' ) : # Check to see if we are in a virtualenv
# If we are , assemble the path manually
    bin_path = os . path . abspath ( os . path . join ( sys . prefix , 'bin' ) )
    apidoc_path = os . path . join ( bin_path , apidoc_path )
check_call ( [ apidoc_path , '-o' , docs_path , os . path . join ( root_path , 'enterprise' ) , os . path . join ( root_path , 'enterprise/migrations' ) ] )


signature:
def setup ( app ) :
    """Sphinx extension : run sphinx - apidoc ."""

implementation:
event = 'builder-inited' if six . PY3 else b'builder-inited'
app . connect ( event , on_init )


signature:
def add_arguments ( self , parser ) :
    """Adds the optional arguments : ` ` - - enterprise _ customer ` ` , ` ` - - channel ` `"""

implementation:
parser . add_argument ( '--enterprise_customer' , dest = 'enterprise_customer' , default = None , metavar = 'ENTERPRISE_CUSTOMER_UUID' , help = _ ( 'Transmit data for only this EnterpriseCustomer. ' 'Omit this option to transmit to all EnterpriseCustomers with active integrated channels.' ) , )
parser . add_argument ( '--channel' , dest = 'channel' , default = '' , metavar = 'INTEGRATED_CHANNEL' , help = _ ( 'Transmit data to this IntegrateChannel. ' 'Omit this option to transmit to all configured, active integrated channels.' ) , choices = INTEGRATED_CHANNEL_CHOICES . keys ( ) , )


signature:
def get_integrated_channels ( self , options ) :
    """Generates a list of active integrated channels for active customers , filtered from the given options .
 Raises errors when invalid options are encountered .
 See ` ` add _ arguments ` ` for the accepted options ."""

implementation:
channel_classes = self . get_channel_classes ( options . get ( 'channel' ) )
filter_kwargs = { 'active' : True , 'enterprise_customer__active' : True , }
enterprise_customer = self . get_enterprise_customer ( options . get ( 'enterprise_customer' ) )
if enterprise_customer :
    filter_kwargs [ 'enterprise_customer' ] = enterprise_customer
for channel_class in channel_classes :
    for integrated_channel in channel_class . objects . filter ( ** filter_kwargs ) :
        yield integrated_channel


signature:
def get_enterprise_customer ( uuid ) :
    """Returns the enterprise customer requested for the given uuid , None if not .
 Raises CommandError if uuid is invalid ."""

implementation:
if uuid is None :
    return None
try :
    return EnterpriseCustomer . active_customers . get ( uuid = uuid )
except EnterpriseCustomer . DoesNotExist :
    raise CommandError ( _ ( 'Enterprise customer {uuid} not found, or not active' ) . format ( uuid = uuid ) )


signature:
def get_channel_classes ( channel_code ) :
    """Assemble a list of integrated channel classes to transmit to .
 If a valid channel type was provided , use it .
 Otherwise , use all the available channel types ."""

implementation:
if channel_code : # Channel code is case - insensitive
    channel_code = channel_code . upper ( )
    if channel_code not in INTEGRATED_CHANNEL_CHOICES :
        raise CommandError ( _ ( 'Invalid integrated channel: {channel}' ) . format ( channel = channel_code ) )
    channel_classes = [ INTEGRATED_CHANNEL_CHOICES [ channel_code ] ]
else :
    channel_classes = INTEGRATED_CHANNEL_CHOICES . values ( )
return channel_classes


signature:
def get_result ( self , course_grade ) :
    """Get result for the statement .
 Arguments :
 course _ grade ( CourseGrade ) : Course grade ."""

implementation: return Result ( score = Score ( scaled = course_grade . percent , raw = course_grade . percent * 100 , min = MIN_SCORE , max = MAX_SCORE , ) , success = course_grade . passed , completion = course_grade . passed )


signature:
def get_requirements ( requirements_file ) :
    """Get the contents of a file listing the requirements"""

implementation:
lines = open ( requirements_file ) . readlines ( )
dependencies = [ ]
dependency_links = [ ]
for line in lines :
    package = line . strip ( )
    if package . startswith ( '#' ) : # Skip pure comment lines
        continue
    if any ( package . startswith ( prefix ) for prefix in VCS_PREFIXES ) : # VCS reference for dev purposes , expect a trailing comment
    # with the normal requirement
        package_link , __ , package = package . rpartition ( '#' )
        # Remove - e < version _ control > string
        package_link = re . sub ( r'(.*)(?P<dependency_link>https?.*$)' , r'\g<dependency_link>' , package_link )
        package = re . sub ( r'(egg=)?(?P<package_name>.*)==.*$' , r'\g<package_name>' , package )
        package_version = re . sub ( r'.*[^=]==' , '' , line . strip ( ) )
        if package :
            dependency_links . append ( '{package_link}#egg={package}-{package_version}' . format ( package_link = package_link , package = package , package_version = package_version , ) )
    else : # Ignore any trailing comment
        package , __ , __ = package . partition ( '#' )
        # Remove any whitespace and assume non - empty results are dependencies
        package = package . strip ( )
    if package :
        dependencies . append ( package )
return dependencies , dependency_links


signature:
def transmit_learner_data ( self , user ) :
    """Iterate over each learner data record and transmit it to the integrated channel ."""

implementation:
exporter = self . get_learner_data_exporter ( user )
transmitter = self . get_learner_data_transmitter ( )
transmitter . transmit ( exporter )


signature:
def transmit_content_metadata ( self , user ) :
    """Transmit content metadata to integrated channel ."""

implementation:
exporter = self . get_content_metadata_exporter ( user )
transmitter = self . get_content_metadata_transmitter ( )
transmitter . transmit ( exporter . export ( ) )


signature:
def get_learner_data_records ( self , enterprise_enrollment , completed_date = None , is_passing = False , ** kwargs ) : # pylint : disable = arguments - differ , unused - argument
    """Return a DegreedLearnerDataTransmissionAudit with the given enrollment and course completion data .
 If completed _ date is None , then course completion has not been met .
 If no remote ID can be found , return None ."""

implementation:
# Degreed expects completion dates of the form ' yyyy - mm - dd ' .
completed_timestamp = completed_date . strftime ( "%F" ) if isinstance ( completed_date , datetime ) else None
if enterprise_enrollment . enterprise_customer_user . get_remote_id ( ) is not None :
    DegreedLearnerDataTransmissionAudit = apps . get_model ( # pylint : disable = invalid - name
    'degreed' , 'DegreedLearnerDataTransmissionAudit' )
    # We return two records here , one with the course key and one with the course run id , to account for
    # uncertainty about the type of content ( course vs . course run ) that was sent to the integrated channel .
    return [ DegreedLearnerDataTransmissionAudit ( enterprise_course_enrollment_id = enterprise_enrollment . id , degreed_user_email = enterprise_enrollment . enterprise_customer_user . user_email , course_id = parse_course_key ( enterprise_enrollment . course_id ) , course_completed = completed_date is not None and is_passing , completed_timestamp = completed_timestamp , ) , DegreedLearnerDataTransmissionAudit ( enterprise_course_enrollment_id = enterprise_enrollment . id , degreed_user_email = enterprise_enrollment . enterprise_customer_user . user_email , course_id = enterprise_enrollment . course_id , course_completed = completed_date is not None and is_passing , completed_timestamp = completed_timestamp , ) ]
else :
    LOGGER . debug ( 'No learner data was sent for user [%s] because a Degreed user ID could not be found.' , enterprise_enrollment . enterprise_customer_user . username )


signature:
def get ( self , request , template_id , view_type ) :
    """Render the given template with the stock data ."""

implementation:
template = get_object_or_404 ( EnrollmentNotificationEmailTemplate , pk = template_id )
if view_type not in self . view_type_contexts :
    return HttpResponse ( status = 404 )
base_context = self . view_type_contexts [ view_type ] . copy ( )
base_context . update ( { 'user_name' : self . get_user_name ( request ) } )
return HttpResponse ( template . render_html_template ( base_context ) , content_type = 'text/html' )


signature:
def _build_admin_context ( request , customer ) :
    """Build common admin context ."""

implementation:
opts = customer . _meta
codename = get_permission_codename ( 'change' , opts )
has_change_permission = request . user . has_perm ( '%s.%s' % ( opts . app_label , codename ) )
return { 'has_change_permission' : has_change_permission , 'opts' : opts }


signature:
def _build_context ( self , request , enterprise_customer_uuid ) :
    """Build common context parts used by different handlers in this view ."""

implementation:
enterprise_customer = EnterpriseCustomer . objects . get ( uuid = enterprise_customer_uuid )
# pylint : disable = no - member
context = { self . ContextParameters . ENTERPRISE_CUSTOMER : enterprise_customer , }
context . update ( admin . site . each_context ( request ) )
context . update ( self . _build_admin_context ( request , enterprise_customer ) )
return context


signature:
def get ( self , request , enterprise_customer_uuid ) :
    """Handle GET request - render " Transmit courses metadata " form .
 Arguments :
 request ( django . http . request . HttpRequest ) : Request instance
 enterprise _ customer _ uuid ( str ) : Enterprise Customer UUID
 Returns :
 django . http . response . HttpResponse : HttpResponse"""

implementation:
context = self . _build_context ( request , enterprise_customer_uuid )
transmit_courses_metadata_form = TransmitEnterpriseCoursesForm ( )
context . update ( { self . ContextParameters . TRANSMIT_COURSES_METADATA_FORM : transmit_courses_metadata_form } )
return render ( request , self . template , context )


signature:
def post ( self , request , enterprise_customer_uuid ) :
    """Handle POST request - handle form submissions .
 Arguments :
 request ( django . http . request . HttpRequest ) : Request instance
 enterprise _ customer _ uuid ( str ) : Enterprise Customer UUID"""

implementation:
transmit_courses_metadata_form = TransmitEnterpriseCoursesForm ( request . POST )
# check that form data is well - formed
if transmit_courses_metadata_form . is_valid ( ) :
    channel_worker_username = transmit_courses_metadata_form . cleaned_data [ 'channel_worker_username' ]
    # call ` transmit _ content _ metadata ` management command to trigger
    # transmission of enterprise courses metadata
    call_command ( 'transmit_content_metadata' , '--catalog_user' , channel_worker_username , enterprise_customer = enterprise_customer_uuid )
    # Redirect to GET
    return HttpResponseRedirect ( '' )
context = self . _build_context ( request , enterprise_customer_uuid )
context . update ( { self . ContextParameters . TRANSMIT_COURSES_METADATA_FORM : transmit_courses_metadata_form } )
return render ( request , self . template , context )


signature:
def _build_context ( self , request , customer_uuid ) :
    """Build common context parts used by different handlers in this view ."""

implementation:
# TODO : pylint acts stupid - find a way around it without suppressing
enterprise_customer = EnterpriseCustomer . objects . get ( uuid = customer_uuid )
# pylint : disable = no - member
search_keyword = self . get_search_keyword ( request )
linked_learners = self . get_enterprise_customer_user_queryset ( request , search_keyword , customer_uuid )
pending_linked_learners = self . get_pending_users_queryset ( search_keyword , customer_uuid )
context = { self . ContextParameters . ENTERPRISE_CUSTOMER : enterprise_customer , self . ContextParameters . PENDING_LEARNERS : pending_linked_learners , self . ContextParameters . LEARNERS : linked_learners , self . ContextParameters . SEARCH_KEYWORD : search_keyword or '' , self . ContextParameters . ENROLLMENT_URL : settings . LMS_ENROLLMENT_API_PATH , }
context . update ( admin . site . each_context ( request ) )
context . update ( self . _build_admin_context ( request , enterprise_customer ) )
return context


signature:
def get_enterprise_customer_user_queryset ( self , request , search_keyword , customer_uuid , page_size = PAGE_SIZE ) :
    """Get the list of EnterpriseCustomerUsers we want to render .
 Arguments :
 request ( HttpRequest ) : HTTP Request instance .
 search _ keyword ( str ) : The keyword to search for in users ' email addresses and usernames .
 customer _ uuid ( str ) : A unique identifier to filter down to only users linked to a
 particular EnterpriseCustomer .
 page _ size ( int ) : Number of learners displayed in each paginated set ."""

implementation:
page = request . GET . get ( 'page' , 1 )
learners = EnterpriseCustomerUser . objects . filter ( enterprise_customer__uuid = customer_uuid )
user_ids = learners . values_list ( 'user_id' , flat = True )
matching_users = User . objects . filter ( pk__in = user_ids )
if search_keyword is not None :
    matching_users = matching_users . filter ( Q ( email__icontains = search_keyword ) | Q ( username__icontains = search_keyword ) )
matching_user_ids = matching_users . values_list ( 'pk' , flat = True )
learners = learners . filter ( user_id__in = matching_user_ids )
return paginated_list ( learners , page , page_size )


signature:
def get_pending_users_queryset ( self , search_keyword , customer_uuid ) :
    """Get the list of PendingEnterpriseCustomerUsers we want to render .
 Args :
 search _ keyword ( str ) : The keyword to search for in pending users ' email addresses .
 customer _ uuid ( str ) : A unique identifier to filter down to only pending users
 linked to a particular EnterpriseCustomer ."""

implementation:
queryset = PendingEnterpriseCustomerUser . objects . filter ( enterprise_customer__uuid = customer_uuid )
if search_keyword is not None :
    queryset = queryset . filter ( user_email__icontains = search_keyword )
return queryset


signature:
def _handle_singular ( cls , enterprise_customer , manage_learners_form ) :
    """Link single user by email or username .
 Arguments :
 enterprise _ customer ( EnterpriseCustomer ) : learners will be linked to this Enterprise Customer instance
 manage _ learners _ form ( ManageLearnersForm ) : bound ManageLearners form instance"""

implementation:
form_field_value = manage_learners_form . cleaned_data [ ManageLearnersForm . Fields . EMAIL_OR_USERNAME ]
email = email_or_username__to__email ( form_field_value )
try :
    validate_email_to_link ( email , form_field_value , ValidationMessages . INVALID_EMAIL_OR_USERNAME , True )
except ValidationError as exc :
    manage_learners_form . add_error ( ManageLearnersForm . Fields . EMAIL_OR_USERNAME , exc )
else :
    EnterpriseCustomerUser . objects . link_user ( enterprise_customer , email )
    return [ email ]


signature:
def _handle_bulk_upload ( cls , enterprise_customer , manage_learners_form , request , email_list = None ) :
    """Bulk link users by email .
 Arguments :
 enterprise _ customer ( EnterpriseCustomer ) : learners will be linked to this Enterprise Customer instance
 manage _ learners _ form ( ManageLearnersForm ) : bound ManageLearners form instance
 request ( django . http . request . HttpRequest ) : HTTP Request instance
 email _ list ( iterable ) : A list of pre - processed email addresses to handle using the form"""

implementation:
errors = [ ]
emails = set ( )
already_linked_emails = [ ]
duplicate_emails = [ ]
csv_file = manage_learners_form . cleaned_data [ ManageLearnersForm . Fields . BULK_UPLOAD ]
if email_list :
    parsed_csv = [ { ManageLearnersForm . CsvColumns . EMAIL : email } for email in email_list ]
else :
    parsed_csv = parse_csv ( csv_file , expected_columns = { ManageLearnersForm . CsvColumns . EMAIL } )
try :
    for index , row in enumerate ( parsed_csv ) :
        email = row [ ManageLearnersForm . CsvColumns . EMAIL ]
        try :
            already_linked = validate_email_to_link ( email , ignore_existing = True )
        except ValidationError as exc :
            message = _ ( "Error at line {line}: {message}\n" ) . format ( line = index + 1 , message = exc )
            errors . append ( message )
        else :
            if already_linked :
                already_linked_emails . append ( ( email , already_linked . enterprise_customer ) )
            elif email in emails :
                duplicate_emails . append ( email )
            else :
                emails . add ( email )
except ValidationError as exc :
    errors . append ( exc )
if errors :
    manage_learners_form . add_error ( ManageLearnersForm . Fields . GENERAL_ERRORS , ValidationMessages . BULK_LINK_FAILED )
    for error in errors :
        manage_learners_form . add_error ( ManageLearnersForm . Fields . BULK_UPLOAD , error )
    return
# There were no errors . Now do the actual linking :
for email in emails :
    EnterpriseCustomerUser . objects . link_user ( enterprise_customer , email )
# Report what happened :
count = len ( emails )
messages . success ( request , ungettext ( "{count} new learner was added to {enterprise_customer_name}." , "{count} new learners were added to {enterprise_customer_name}." , count ) . format ( count = count , enterprise_customer_name = enterprise_customer . name ) )
this_customer_linked_emails = [ email for email , customer in already_linked_emails if customer == enterprise_customer ]
other_customer_linked_emails = [ email for email , __ in already_linked_emails if email not in this_customer_linked_emails ]
if this_customer_linked_emails :
    messages . warning ( request , _ ( "The following learners were already associated with this Enterprise " "Customer: {list_of_emails}" ) . format ( list_of_emails = ", " . join ( this_customer_linked_emails ) ) )
if other_customer_linked_emails :
    messages . warning ( request , _ ( "The following learners are already associated with " "another Enterprise Customer. These learners were not " "added to {enterprise_customer_name}: {list_of_emails}" ) . format ( enterprise_customer_name = enterprise_customer . name , list_of_emails = ", " . join ( other_customer_linked_emails ) , ) )
if duplicate_emails :
    messages . warning ( request , _ ( "The following duplicate email addresses were not added: " "{list_of_emails}" ) . format ( list_of_emails = ", " . join ( duplicate_emails ) ) )
# Build a list of all the emails that we can act on further ; that is ,
# emails that we either linked to this customer , or that were linked already .
all_processable_emails = list ( emails ) + this_customer_linked_emails
return all_processable_emails


signature:
def enroll_user ( cls , enterprise_customer , user , course_mode , * course_ids ) :
    """Enroll a single user in any number of courses using a particular course mode .
 Args :
 enterprise _ customer : The EnterpriseCustomer which is sponsoring the enrollment
 user : The user who needs to be enrolled in the course
 course _ mode : The mode with which the enrollment should be created
 * course _ ids : An iterable containing any number of course IDs to eventually enroll the user in .
 Returns :
 Boolean : Whether or not enrollment succeeded for all courses specified"""

implementation:
enterprise_customer_user , __ = EnterpriseCustomerUser . objects . get_or_create ( enterprise_customer = enterprise_customer , user_id = user . id )
enrollment_client = EnrollmentApiClient ( )
succeeded = True
for course_id in course_ids :
    try :
        enrollment_client . enroll_user_in_course ( user . username , course_id , course_mode )
    except HttpClientError as exc : # Check if user is already enrolled then we should ignore exception
        if cls . is_user_enrolled ( user , course_id , course_mode ) :
            succeeded = True
        else :
            succeeded = False
            default_message = 'No error message provided'
            try :
                error_message = json . loads ( exc . content . decode ( ) ) . get ( 'message' , default_message )
            except ValueError :
                error_message = default_message
            logging . error ( 'Error while enrolling user %(user)s: %(message)s' , dict ( user = user . username , message = error_message ) )
    if succeeded :
        __ , created = EnterpriseCourseEnrollment . objects . get_or_create ( enterprise_customer_user = enterprise_customer_user , course_id = course_id )
        if created :
            track_enrollment ( 'admin-enrollment' , user . id , course_id )
return succeeded


signature:
def is_user_enrolled ( cls , user , course_id , course_mode ) :
    """Query the enrollment API and determine if a learner is enrolled in a given course run track .
 Args :
 user : The user whose enrollment needs to be checked
 course _ mode : The mode with which the enrollment should be checked
 course _ id : course id of the course where enrollment should be checked .
 Returns :
 Boolean : Whether or not enrollment exists"""

implementation:
enrollment_client = EnrollmentApiClient ( )
try :
    enrollments = enrollment_client . get_course_enrollment ( user . username , course_id )
    if enrollments and course_mode == enrollments . get ( 'mode' ) :
        return True
except HttpClientError as exc :
    logging . error ( 'Error while checking enrollment status of user %(user)s: %(message)s' , dict ( user = user . username , message = str ( exc ) ) )
except KeyError as exc :
    logging . warning ( 'Error while parsing enrollment data of user %(user)s: %(message)s' , dict ( user = user . username , message = str ( exc ) ) )
return False


signature:
def get_users_by_email ( cls , emails ) :
    """Accept a list of emails , and separate them into users that exist on OpenEdX and users who don ' t .
 Args :
 emails : An iterable of email addresses to split between existing and nonexisting
 Returns :
 users : Queryset of users who exist in the OpenEdX platform and who were in the list of email addresses
 missing _ emails : List of unique emails which were in the original list , but do not yet exist as users"""

implementation:
users = User . objects . filter ( email__in = emails )
present_emails = users . values_list ( 'email' , flat = True )
missing_emails = list ( set ( emails ) - set ( present_emails ) )
return users , missing_emails


signature:
def enroll_users_in_program ( cls , enterprise_customer , program_details , course_mode , emails , cohort = None ) :
    """Enroll existing users in all courses in a program , and create pending enrollments for nonexisting users .
 Args :
 enterprise _ customer : The EnterpriseCustomer which is sponsoring the enrollment
 program _ details : The details of the program in which we ' re enrolling
 course _ mode ( str ) : The mode with which we ' re enrolling in the program
 emails : An iterable of email addresses which need to be enrolled
 Returns :
 successes : A list of users who were successfully enrolled in all courses of the program
 pending : A list of PendingEnterpriseCustomerUsers who were successfully linked and had
 pending enrollments created for them in the database
 failures : A list of users who could not be enrolled in the program"""

implementation:
existing_users , unregistered_emails = cls . get_users_by_email ( emails )
course_ids = get_course_runs_from_program ( program_details )
successes = [ ]
pending = [ ]
failures = [ ]
for user in existing_users :
    succeeded = cls . enroll_user ( enterprise_customer , user , course_mode , * course_ids )
    if succeeded :
        successes . append ( user )
    else :
        failures . append ( user )
for email in unregistered_emails :
    pending_user = enterprise_customer . enroll_user_pending_registration ( email , course_mode , * course_ids , cohort = cohort )
    pending . append ( pending_user )
return successes , pending , failures


signature:
def enroll_users_in_course ( cls , enterprise_customer , course_id , course_mode , emails ) :
    """Enroll existing users in a course , and create a pending enrollment for nonexisting users .
 Args :
 enterprise _ customer : The EnterpriseCustomer which is sponsoring the enrollment
 course _ id ( str ) : The unique identifier of the course in which we ' re enrolling
 course _ mode ( str ) : The mode with which we ' re enrolling in the course
 emails : An iterable of email addresses which need to be enrolled
 Returns :
 successes : A list of users who were successfully enrolled in the course
 pending : A list of PendingEnterpriseCustomerUsers who were successfully linked and had
 pending enrollments created for them in the database
 failures : A list of users who could not be enrolled in the course"""

implementation:
existing_users , unregistered_emails = cls . get_users_by_email ( emails )
successes = [ ]
pending = [ ]
failures = [ ]
for user in existing_users :
    succeeded = cls . enroll_user ( enterprise_customer , user , course_mode , course_id )
    if succeeded :
        successes . append ( user )
    else :
        failures . append ( user )
for email in unregistered_emails :
    pending_user = enterprise_customer . enroll_user_pending_registration ( email , course_mode , course_id )
    pending . append ( pending_user )
return successes , pending , failures


signature:
def send_messages ( cls , http_request , message_requests ) :
    """Deduplicate any outgoing message requests , and send the remainder .
 Args :
 http _ request : The HTTP request in whose response we want to embed the messages
 message _ requests : A list of undeduplicated messages in the form of tuples of message type
 and text - for example , ( ' error ' , ' Something went wrong ' )"""

implementation:
deduplicated_messages = set ( message_requests )
for msg_type , text in deduplicated_messages :
    message_function = getattr ( messages , msg_type )
    message_function ( http_request , text )


signature:
def notify_program_learners ( cls , enterprise_customer , program_details , users ) :
    """Notify learners about a program in which they ' ve been enrolled .
 Args :
 enterprise _ customer : The EnterpriseCustomer being linked to
 program _ details : Details about the specific program the learners were enrolled in
 users : An iterable of the users or pending users who were enrolled"""

implementation:
program_name = program_details . get ( 'title' )
program_branding = program_details . get ( 'type' )
program_uuid = program_details . get ( 'uuid' )
lms_root_url = get_configuration_value_for_site ( enterprise_customer . site , 'LMS_ROOT_URL' , settings . LMS_ROOT_URL )
program_path = urlquote ( '/dashboard/programs/{program_uuid}/?tpa_hint={tpa_hint}' . format ( program_uuid = program_uuid , tpa_hint = enterprise_customer . identity_provider , ) )
destination_url = '{site}/{login_or_register}?next={program_path}' . format ( site = lms_root_url , login_or_register = '{login_or_register}' , program_path = program_path )
program_type = 'program'
program_start = get_earliest_start_date_from_program ( program_details )
with mail . get_connection ( ) as email_conn :
    for user in users :
        login_or_register = 'register' if isinstance ( user , PendingEnterpriseCustomerUser ) else 'login'
        destination_url = destination_url . format ( login_or_register = login_or_register )
        send_email_notification_message ( user = user , enrolled_in = { 'name' : program_name , 'url' : destination_url , 'type' : program_type , 'start' : program_start , 'branding' : program_branding , } , enterprise_customer = enterprise_customer , email_connection = email_conn )


signature:
def get_success_enrollment_message ( cls , users , enrolled_in ) :
    """Create message for the users who were enrolled in a course or program .
 Args :
 users : An iterable of users who were successfully enrolled
 enrolled _ in ( str ) : A string identifier for the course or program the users were enrolled in
 Returns :
 tuple : A 2 - tuple containing a message type and message text"""

implementation:
enrolled_count = len ( users )
return ( 'success' , ungettext ( '{enrolled_count} learner was enrolled in {enrolled_in}.' , '{enrolled_count} learners were enrolled in {enrolled_in}.' , enrolled_count , ) . format ( enrolled_count = enrolled_count , enrolled_in = enrolled_in , ) )


signature:
def get_failed_enrollment_message ( cls , users , enrolled_in ) :
    """Create message for the users who were not able to be enrolled in a course or program .
 Args :
 users : An iterable of users who were not successfully enrolled
 enrolled _ in ( str ) : A string identifier for the course or program with which enrollment was attempted
 Returns :
 tuple : A 2 - tuple containing a message type and message text"""

implementation:
failed_emails = [ user . email for user in users ]
return ( 'error' , _ ( 'The following learners could not be enrolled in {enrolled_in}: {user_list}' ) . format ( enrolled_in = enrolled_in , user_list = ', ' . join ( failed_emails ) , ) )


signature:
def get_pending_enrollment_message ( cls , pending_users , enrolled_in ) :
    """Create message for the users who were enrolled in a course or program .
 Args :
 users : An iterable of PendingEnterpriseCustomerUsers who were successfully linked with a pending enrollment
 enrolled _ in ( str ) : A string identifier for the course or program the pending users were linked to
 Returns :
 tuple : A 2 - tuple containing a message type and message text"""

implementation:
pending_emails = [ pending_user . user_email for pending_user in pending_users ]
return ( 'warning' , _ ( "The following learners do not have an account on " "{platform_name}. They have not been enrolled in " "{enrolled_in}. When these learners create an account, they will " "be enrolled automatically: {pending_email_list}" ) . format ( platform_name = settings . PLATFORM_NAME , enrolled_in = enrolled_in , pending_email_list = ', ' . join ( pending_emails ) , ) )


signature:
def _enroll_users ( cls , request , enterprise_customer , emails , mode , course_id = None , program_details = None , notify = True ) :
    """Enroll the users with the given email addresses to the courses specified , either specifically or by program .
 Args :
 cls ( type ) : The EnterpriseCustomerManageLearnersView class itself
 request : The HTTP request the enrollment is being created by
 enterprise _ customer : The instance of EnterpriseCustomer whose attached users we ' re enrolling
 emails : An iterable of strings containing email addresses to enroll in a course
 mode : The enrollment mode the users will be enrolled in the course with
 course _ id : The ID of the course in which we want to enroll
 program _ details : Details about a program in which we want to enroll
 notify : Whether to notify ( by email ) the users that have been enrolled"""

implementation:
pending_messages = [ ]
if course_id :
    succeeded , pending , failed = cls . enroll_users_in_course ( enterprise_customer = enterprise_customer , course_id = course_id , course_mode = mode , emails = emails , )
    all_successes = succeeded + pending
    if notify :
        enterprise_customer . notify_enrolled_learners ( catalog_api_user = request . user , course_id = course_id , users = all_successes , )
    if succeeded :
        pending_messages . append ( cls . get_success_enrollment_message ( succeeded , course_id ) )
    if failed :
        pending_messages . append ( cls . get_failed_enrollment_message ( failed , course_id ) )
    if pending :
        pending_messages . append ( cls . get_pending_enrollment_message ( pending , course_id ) )
if program_details :
    succeeded , pending , failed = cls . enroll_users_in_program ( enterprise_customer = enterprise_customer , program_details = program_details , course_mode = mode , emails = emails , )
    all_successes = succeeded + pending
    if notify :
        cls . notify_program_learners ( enterprise_customer = enterprise_customer , program_details = program_details , users = all_successes )
    program_identifier = program_details . get ( 'title' , program_details . get ( 'uuid' , _ ( 'the program' ) ) )
    if succeeded :
        pending_messages . append ( cls . get_success_enrollment_message ( succeeded , program_identifier ) )
    if failed :
        pending_messages . append ( cls . get_failed_enrollment_message ( failed , program_identifier ) )
    if pending :
        pending_messages . append ( cls . get_pending_enrollment_message ( pending , program_identifier ) )
cls . send_messages ( request , pending_messages )


signature:
def get ( self , request , customer_uuid ) :
    """Handle GET request - render linked learners list and " Link learner " form .
 Arguments :
 request ( django . http . request . HttpRequest ) : Request instance
 customer _ uuid ( str ) : Enterprise Customer UUID
 Returns :
 django . http . response . HttpResponse : HttpResponse"""

implementation:
context = self . _build_context ( request , customer_uuid )
manage_learners_form = ManageLearnersForm ( user = request . user , enterprise_customer = context [ self . ContextParameters . ENTERPRISE_CUSTOMER ] )
context . update ( { self . ContextParameters . MANAGE_LEARNERS_FORM : manage_learners_form } )
return render ( request , self . template , context )


signature:
def post ( self , request , customer_uuid ) :
    """Handle POST request - handle form submissions .
 Arguments :
 request ( django . http . request . HttpRequest ) : Request instance
 customer _ uuid ( str ) : Enterprise Customer UUID
 Returns :
 django . http . response . HttpResponse : HttpResponse"""

implementation:
enterprise_customer = EnterpriseCustomer . objects . get ( uuid = customer_uuid )
# pylint : disable = no - member
manage_learners_form = ManageLearnersForm ( request . POST , request . FILES , user = request . user , enterprise_customer = enterprise_customer )
# initial form validation - check that form data is well - formed
if manage_learners_form . is_valid ( ) :
    email_field_as_bulk_input = split_usernames_and_emails ( manage_learners_form . cleaned_data [ ManageLearnersForm . Fields . EMAIL_OR_USERNAME ] )
    is_bulk_entry = len ( email_field_as_bulk_input ) > 1
    # The form is valid . Call the appropriate helper depending on the mode :
    mode = manage_learners_form . cleaned_data [ ManageLearnersForm . Fields . MODE ]
    if mode == ManageLearnersForm . Modes . MODE_SINGULAR and not is_bulk_entry :
        linked_learners = self . _handle_singular ( enterprise_customer , manage_learners_form )
    elif mode == ManageLearnersForm . Modes . MODE_SINGULAR :
        linked_learners = self . _handle_bulk_upload ( enterprise_customer , manage_learners_form , request , email_list = email_field_as_bulk_input )
    else :
        linked_learners = self . _handle_bulk_upload ( enterprise_customer , manage_learners_form , request )
# _ handle _ form might add form errors , so we check if it is still valid
if manage_learners_form . is_valid ( ) :
    course_details = manage_learners_form . cleaned_data . get ( ManageLearnersForm . Fields . COURSE )
    program_details = manage_learners_form . cleaned_data . get ( ManageLearnersForm . Fields . PROGRAM )
    notification_type = manage_learners_form . cleaned_data . get ( ManageLearnersForm . Fields . NOTIFY )
    notify = notification_type == ManageLearnersForm . NotificationTypes . BY_EMAIL
    course_id = None
    if course_details :
        course_id = course_details [ 'course_id' ]
    if course_id or program_details :
        course_mode = manage_learners_form . cleaned_data [ ManageLearnersForm . Fields . COURSE_MODE ]
        self . _enroll_users ( request = request , enterprise_customer = enterprise_customer , emails = linked_learners , mode = course_mode , course_id = course_id , program_details = program_details , notify = notify , )
    # Redirect to GET if everything went smooth .
    manage_learners_url = reverse ( "admin:" + UrlNames . MANAGE_LEARNERS , args = ( customer_uuid , ) )
    search_keyword = self . get_search_keyword ( request )
    if search_keyword :
        manage_learners_url = manage_learners_url + "?q=" + search_keyword
    return HttpResponseRedirect ( manage_learners_url )
# if something went wrong - display bound form on the page
context = self . _build_context ( request , customer_uuid )
context . update ( { self . ContextParameters . MANAGE_LEARNERS_FORM : manage_learners_form } )
return render ( request , self . template , context )


signature:
def delete ( self , request , customer_uuid ) :
    """Handle DELETE request - handle unlinking learner .
 Arguments :
 request ( django . http . request . HttpRequest ) : Request instance
 customer _ uuid ( str ) : Enterprise Customer UUID
 Returns :
 django . http . response . HttpResponse : HttpResponse"""

implementation:
# TODO : pylint acts stupid - find a way around it without suppressing
enterprise_customer = EnterpriseCustomer . objects . get ( uuid = customer_uuid )
# pylint : disable = no - member
email_to_unlink = request . GET [ "unlink_email" ]
try :
    EnterpriseCustomerUser . objects . unlink_user ( enterprise_customer = enterprise_customer , user_email = email_to_unlink )
except ( EnterpriseCustomerUser . DoesNotExist , PendingEnterpriseCustomerUser . DoesNotExist ) :
    message = _ ( "Email {email} is not associated with Enterprise " "Customer {ec_name}" ) . format ( email = email_to_unlink , ec_name = enterprise_customer . name )
    return HttpResponse ( message , content_type = "application/json" , status = 404 )
return HttpResponse ( json . dumps ( { } ) , content_type = "application/json" )


signature:
def proxied_get ( self , * args , ** kwargs ) :
    """Perform the query and returns a single object matching the given keyword arguments .
 This customizes the queryset to return an instance of ` ` ProxyDataSharingConsent ` ` when
 the searched - for ` ` DataSharingConsent ` ` instance does not exist ."""

implementation:
original_kwargs = kwargs . copy ( )
if 'course_id' in kwargs :
    try : # Check if we have a course ID or a course run ID
        course_run_key = str ( CourseKey . from_string ( kwargs [ 'course_id' ] ) )
    except InvalidKeyError : # The ID we have is for a course instead of a course run ; fall through
    # to the second check .
        pass
    else :
        try : # Try to get the record for the course run specifically
            return self . get ( * args , ** kwargs )
        except DataSharingConsent . DoesNotExist : # A record for the course run didn ' t exist , so modify the query
        # parameters to look for just a course record on the second pass .
            kwargs [ 'course_id' ] = parse_course_key ( course_run_key )
try :
    return self . get ( * args , ** kwargs )
except DataSharingConsent . DoesNotExist :
    return ProxyDataSharingConsent ( ** original_kwargs )


signature:
def from_children ( cls , program_uuid , * children ) :
    """Build a ProxyDataSharingConsent using the details of the received consent records ."""

implementation:
if not children or any ( child is None for child in children ) :
    return None
granted = all ( ( child . granted for child in children ) )
exists = any ( ( child . exists for child in children ) )
usernames = set ( [ child . username for child in children ] )
enterprises = set ( [ child . enterprise_customer for child in children ] )
if not len ( usernames ) == len ( enterprises ) == 1 :
    raise InvalidProxyConsent ( 'Children used to create a bulk proxy consent object must ' 'share a single common username and EnterpriseCustomer.' )
username = children [ 0 ] . username
enterprise_customer = children [ 0 ] . enterprise_customer
return cls ( enterprise_customer = enterprise_customer , username = username , program_uuid = program_uuid , exists = exists , granted = granted , child_consents = children )


signature:
def commit ( self ) :
    """Commit a real ` ` DataSharingConsent ` ` object to the database , mirroring current field settings .
 : return : A ` ` DataSharingConsent ` ` object if validation is successful , otherwise ` ` None ` ` ."""

implementation:
if self . _child_consents :
    consents = [ ]
    for consent in self . _child_consents :
        consent . granted = self . granted
        consents . append ( consent . save ( ) or consent )
    return ProxyDataSharingConsent . from_children ( self . program_uuid , * consents )
consent , _ = DataSharingConsent . objects . update_or_create ( enterprise_customer = self . enterprise_customer , username = self . username , course_id = self . course_id , defaults = { 'granted' : self . granted } )
self . _exists = consent . exists
return consent


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:
persistent_course_grades = self . get_course_completions ( lrs_configuration . enterprise_customer , days )
users = self . prefetch_users ( persistent_course_grades )
course_overviews = self . prefetch_courses ( persistent_course_grades )
for persistent_course_grade in persistent_course_grades :
    try :
        user = users . get ( persistent_course_grade . user_id )
        course_overview = course_overviews . get ( persistent_course_grade . course_id )
        course_grade = CourseGradeFactory ( ) . read ( user , course_key = persistent_course_grade . course_id )
        send_course_completion_statement ( lrs_configuration , user , course_overview , course_grade )
    except ClientError :
        LOGGER . exception ( 'Client error while sending course completion to xAPI for' ' enterprise customer {enterprise_customer}.' . format ( enterprise_customer = lrs_configuration . enterprise_customer . name ) )


signature:
def get_course_completions ( self , enterprise_customer , days ) :
    """Get course completions via PersistentCourseGrade 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 PersistentCourseGrade objects ."""

implementation: return PersistentCourseGrade . objects . filter ( passed_timestamp__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 prefetch_users ( persistent_course_grades ) :
    """Prefetch Users from the list of user _ ids present in the persistent _ course _ grades .
 Arguments :
 persistent _ course _ grades ( list ) : A list of PersistentCourseGrade .
 Returns :
 ( dict ) : A dictionary containing user _ id to user mapping ."""

implementation:
users = User . objects . filter ( id__in = [ grade . user_id for grade in persistent_course_grades ] )
return { user . id : user for user in users }


signature:
def create_roles ( apps , schema_editor ) :
    """Create the enterprise roles if they do not already exist ."""

implementation:
EnterpriseFeatureRole = apps . get_model ( 'enterprise' , 'EnterpriseFeatureRole' )
EnterpriseFeatureRole . objects . update_or_create ( name = ENTERPRISE_CATALOG_ADMIN_ROLE )
EnterpriseFeatureRole . objects . update_or_create ( name = ENTERPRISE_DASHBOARD_ADMIN_ROLE )
EnterpriseFeatureRole . objects . update_or_create ( name = ENTERPRISE_ENROLLMENT_API_ADMIN_ROLE )


signature:
def delete_roles ( apps , schema_editor ) :
    """Delete the enterprise roles ."""

implementation:
EnterpriseFeatureRole = apps . get_model ( 'enterprise' , 'EnterpriseFeatureRole' )
EnterpriseFeatureRole . objects . filter ( name__in = [ ENTERPRISE_CATALOG_ADMIN_ROLE , ENTERPRISE_DASHBOARD_ADMIN_ROLE , ENTERPRISE_ENROLLMENT_API_ADMIN_ROLE ] ) . delete ( )


signature:
def get_identity_provider ( provider_id ) :
    """Get Identity Provider with given id .
 Return :
 Instance of ProviderConfig or None ."""

implementation:
try :
    from third_party_auth . provider import Registry
    # pylint : disable = redefined - outer - name
except ImportError as exception :
    LOGGER . warning ( "Could not import Registry from third_party_auth.provider" )
    LOGGER . warning ( exception )
    Registry = None
# pylint : disable = redefined - outer - name
try :
    return Registry and Registry . get ( provider_id )
except ValueError :
    return None


signature:
def get_idp_choices ( ) :
    """Get a list of identity providers choices for enterprise customer .
 Return :
 A list of choices of all identity providers , None if it can not get any available identity provider ."""

implementation:
try :
    from third_party_auth . provider import Registry
    # pylint : disable = redefined - outer - name
except ImportError as exception :
    LOGGER . warning ( "Could not import Registry from third_party_auth.provider" )
    LOGGER . warning ( exception )
    Registry = None
# pylint : disable = redefined - outer - name
first = [ ( "" , "-" * 7 ) ]
if Registry :
    return first + [ ( idp . provider_id , idp . name ) for idp in Registry . enabled ( ) ]
return None


signature:
def get_catalog_admin_url_template ( mode = 'change' ) :
    """Get template of catalog admin url .
 URL template will contain a placeholder ' { catalog _ id } ' for catalog id .
 Arguments :
 mode e . g . change / add .
 Returns :
 A string containing template for catalog url .
 Example :
 > > > get _ catalog _ admin _ url _ template ( ' change ' )
 " http : / / localhost : 18381 / admin / catalogs / catalog / { catalog _ id } / change / " """

implementation:
api_base_url = getattr ( settings , "COURSE_CATALOG_API_URL" , "" )
# Extract FQDN ( Fully Qualified Domain Name ) from API URL .
match = re . match ( r"^(?P<fqdn>(?:https?://)?[^/]+)" , api_base_url )
if not match :
    return ""
# Return matched FQDN from catalog api url appended with catalog admin path
if mode == 'change' :
    return match . group ( "fqdn" ) . rstrip ( "/" ) + "/admin/catalogs/catalog/{catalog_id}/change/"
elif mode == 'add' :
    return match . group ( "fqdn" ) . rstrip ( "/" ) + "/admin/catalogs/catalog/add/"


signature:
def build_notification_message ( template_context , template_configuration = None ) :
    """Create HTML and plaintext message bodies for a notification .
 We receive a context with data we can use to render , as well as an optional site
 template configration - if we don ' t get a template configuration , we ' ll use the
 standard , built - in template .
 Arguments :
 template _ context ( dict ) : A set of data to render
 template _ configuration : A database - backed object with templates
 stored that can be used to render a notification ."""

implementation:
if ( template_configuration is not None and template_configuration . html_template and template_configuration . plaintext_template ) :
    plain_msg , html_msg = template_configuration . render_all_templates ( template_context )
else :
    plain_msg = render_to_string ( 'enterprise/emails/user_notification.txt' , template_context )
    html_msg = render_to_string ( 'enterprise/emails/user_notification.html' , template_context )
return plain_msg , html_msg