def align_by_id()

in microservices/skill_service/src/routes/skill_unified_alignment.py [0:0]

• 64 lines of code
• 22 McCabe index (conditional complexity)

def align_by_id(req_body: AlignByIdsRequestModel):
  """
  Given the Firestore (Skill/Job/Knowledge/Curriculum) id(s), this method
  returns all the candidates to be mapped to the id(s).
  Args:
    req_body (AlignByIdsRequestModel): Required body of Unified Alignment.

  Raises:
    HTTPException: 500 Internal Server Error if something fails.

  Returns:
    [JSON] (AlignByIdsResponseModel): Mapping of given ids with Knowledge,
    Skill, Job, Curriculum items.
  """
  try:
    request_body = req_body.__dict__
    input_type = request_body["input_type"]
    alignment_sources = request_body["output_alignment_sources"]
    if input_type == "skill":
      for skill_id in request_body["ids"]:
        Skill.find_by_uuid(skill_id)
      allowed_output_alignment_keys = {"skill_sources", "learning_resource_ids"}
      if alignment_sources.keys() < allowed_output_alignment_keys:
        raise ValidationError("{0} missing in output_alignment_sources".\
        format(allowed_output_alignment_keys-alignment_sources.keys()))
      for key in alignment_sources.keys():
        if key not in allowed_output_alignment_keys:
          raise ValidationError("Invalid key {0} in output_alignment_sources. "
          "Only {1} are allowed for input_type skill.".\
          format(key, allowed_output_alignment_keys))
      if all(value == [] for value in alignment_sources.values()):
        raise ValidationError("No source is provided for alignment.")
      for key, value in alignment_sources.items():
        if len(value) == 1 and value[0] == "*":
          raise ValidationError(
              "\"*\" is not allowed as value in"
              " \"output_alignment_sources\". Please use specific "
              "source(s) as value.")
        if key == "skill_sources":
          data_sources = get_data_sources("skill")[0]
          SKILL_SOURCES = data_sources["source"]
          MATCHING_ENGINE_INDEX_IDS = data_sources["matching_engine_index_id"]
          for source in value:
            if source not in SKILL_SOURCES:
              raise ValidationError("{0} not a valid skill source. Allowed "
                                     "\"skill_sources\" are {1}.".format(
                                         source, SKILL_SOURCES))
            elif source not in MATCHING_ENGINE_INDEX_IDS:
              raise Exception("Index is not created for {0}. Please use {1} "
                              "to create index.".format(
                                  source,
                                  "skill-service/api/v1/skill/embeddings"))
        elif key == "learning_resource_ids":
          if len(value) > 1:
            raise ValidationError\
              ("Only 1 learning_resource_id is allowed currently")
          for source in value:
            KnowledgeServiceLearningContent.find_by_id(source)
      skill_alignment_sources = alignment_sources.get("skill_sources")
      learning_resource_ids = alignment_sources.get("learning_resource_ids")
      request_body["skill_alignment_sources"] = skill_alignment_sources
      request_body["learning_resource_ids"] = learning_resource_ids
    else:
      raise NotImplementedError("Only \"skill\" is allowed as \"input_type\"")
    alignment_obj = SkillUnifiedAlignment()
    response = alignment_obj.align_by_id(request_body)
    return response
  except ValidationError as e:
    raise BadRequest(str(e)) from e
  except ResourceNotFoundException as e:
    raise ResourceNotFound(str(e)) from e
  except NotImplementedError as e:
    raise APINotImplemented(str(e)) from e
  except Exception as e:
    Logger.error(e)
    Logger.error(traceback.print_exc())
    raise InternalServerError(str(e)) from e