def import_skills()

in microservices/skill_service/src/services/ingest_csv.py [0:0]

• 94 lines of code
• 14 McCabe index (conditional complexity)

def import_skills(skills_json_array):
  """extracts and saves the skills as per skill fireo model"""
  skill_competency_mapping = {}
  inserted_skills = []
  for skill in skills_json_array:
    comp_code = skill.get("CompetencyAlignment")
    skill_code = skill.get("SkillCode")

    if skill_competency_mapping.get(skill_code):
      if comp_code not in skill_competency_mapping.get(skill_code).get(
          "parent_nodes")["competencies"]:
        skill_competency_mapping[skill_code]["parent_nodes"][
          "competencies"].append(comp_code)
    else:
      if skill.get("Keyword"):
        keywords = skill.get("Keyword").split(",")
        for i, item in enumerate(keywords):
          item = item.lstrip()
          keywords[i] = item.rstrip()
      else:
        keywords = []
      # adding id and score to the sme aligned skill
      # TODO: Istead of hardcoding skill_source("emsi")
      #       somehow get it from the csv
      aligned_skill = skill.get("ExternalSkillAlignment")
      skill_alignment = {"emsi": {"aligned": [], "suggested": []}}
      if aligned_skill:
        skill_alignment["emsi"]["aligned"].append({
            "id": "",
            "name": aligned_skill,
            "score": 1.0  # since human aligned this skill
        })
        aligned_skill_doc = Skill.collection.filter("name", "==",
                                                    aligned_skill).get()
        if aligned_skill_doc:
          skill_alignment["emsi"]["aligned"][0]["id"] = aligned_skill_doc.id

      occupations_major_group = []
      if skill.get("OccupationsMajorGroup"):
        occupations_major_group.append(skill.get("OccupationsMajorGroup"))

      occupations_minor_group = []
      if skill.get("OccupationsMinorGroup"):
        occupations_minor_group.append(skill.get("OccupationsMinorGroup"))

      broad_occupation = []
      if skill.get("BroadOccupation"):
        broad_occupation.append(skill.get("BroadOccupation"))

      detailed_occupation = []
      if skill.get("DetailedOccupation"):
        detailed_occupation.append(skill.get("DetailedOccupation"))

      skill_competency_mapping[skill_code] = {
          "name": skill.get("Skill"),
          "description": skill.get("SkillStatement"),
          "parent_nodes": {"competencies": [comp_code]},
          "reference_id": skill.get("SkillCode"),
          "keywords": keywords,
          "occupations": {
              "occupations_major_group": occupations_major_group,
              "occupations_minor_group": occupations_minor_group,
              "broad_occupation": broad_occupation,
              "detailed_occupation": detailed_occupation
          },
          "onet_job": skill.get("ONETJob"),
          "alignments": {
              "standard_alignment": skill.get("StandardAlignment"),
              "credential_alignment": skill.get("CredentialAlignment"),
              "skill_alignment": {
                  "emsi": {
                      "aligned": [{
                          "id": "",
                          "name": skill_alignment,
                          "score": 1.0
                      }],
                      "suggested": []
                  }
              },
              "knowledge_alignment": {},
              "role_alignment": {},
              "organizational_alignment": skill.get("OrganizationalAlignment"),
          },
          "author": "",
          "creator": "",
          "organizations": [],
          "certifications": [],
          "type": {
              "id": "",
              "name": ""
          },
          "source_uri": "snhu",
          "source_name": "snhu",
      }

  for skill in skill_competency_mapping.values():
    skill_obj = Skill()
    skill_obj = skill_obj.from_dict(skill)
    skill_obj.uuid = ""
    skill_obj.save()
    skill_obj.uuid = skill_obj.id
    skill_obj.update()
    inserted_skills.append(skill_obj.to_dict())

  # Insert/Update doc in data_sources collection after ingestion:
  _ = upsert_data_source_doc("skill", "snhu")

  return inserted_skills