Feature: workskill conditions

LICENSE.txt

@@ -1,5 +1,5 @@
 MIT License
-Copyright (c) 2018 Borja Toron
+Copyright (c) 2018-2022 Borja Toron
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal
 in the Software without restriction, including without limitation the rights

examples/get_work_skill_conditions.py

@@ -0,0 +1,97 @@
+#!python
+import argparse
+import pprint
+from logging import basicConfig, debug, info, warning
+from typing import AnyStr, List
+
+import ofsc
+from ofsc import FULL_RESPONSE, JSON_RESPONSE, OFSC
+from ofsc.models import WorkskillCondition
+from openpyxl import Workbook
+
+from config import Config
+
+
+def init_script():
+    global args, instance, pp
+    pp = pprint.PrettyPrinter(indent=4)
+    parser = argparse.ArgumentParser(description="Extract users from OFSC instance")
+    parser.add_argument(
+        "--verbose",
+        type=int,
+        choices={0, 1, 2, 3},
+        default=1,
+        help="Additional messages. 0 is None, 3 is detailed debug",
+    )
+    parser.add_argument(
+        "--output",
+        type=str,
+        default="output.xlsx",
+        help="Excel sheet to update",
+    )
+    args = parser.parse_args()
+
+    # create logger
+    basicConfig(level=40 - 10 * int(args.verbose))
+    instance = OFSC(
+        clientID=Config.OFSC_CLIENT_ID,
+        secret=Config.OFSC_CLIENT_SECRET,
+        companyName=Config.OFSC_COMPANY,
+        baseUrl=Config.OFSC_BASE_URL,
+    )
+    info(f"Creating connection for {Config.OFSC_COMPANY} {Config.OFSC_CLIENT_ID}")
+    return instance
+
+
+def get_workskill_list():
+    response = instance.metadata.get_workskill_conditions(response_type=JSON_RESPONSE)
+    ws_list = [WorkskillCondition.parse_obj(item) for item in response["items"]]
+    return ws_list
+
+
+def write_xls(filename: str, wsc_list: List[WorkskillCondition]):
+    def convert(data):
+        match data:
+            case None:
+                return data
+            case str():
+                return data
+            case bool():
+                return data
+            case int():
+                return data
+            case _:
+                return str(data)
+
+    wb = Workbook()
+    ws_conditions = wb.active
+    ws_conditions.title = "Workskill Conditions"
+    ws_rules = wb.create_sheet("Workskill Conditions Rules")
+    condition_field_names = None
+    rules_field_names = ["condition_id"]
+    for condition in wsc_list:
+        if condition_field_names is None:
+            condition_field_names = list(
+                condition.dict(exclude={"conditions", "dependencies"}).keys()
+            )
+            print(f"Condition Fields: {condition_field_names}")
+            ws_conditions.append(condition_field_names)
+        data = list(condition.dict(exclude={"conditions", "dependencies"}).values())
+        ws_conditions.append(data)
+        for rule in condition.conditions:
+            if len(rules_field_names) == 1:
+                rules_field_names += [str(field) for field in rule.dict().keys()]
+                print(f"Rules Fields: {rules_field_names}")
+                ws_rules.append(rules_field_names)
+            data = [condition.internalId] + [
+                convert(field) for field in rule.dict().values()
+            ]
+            ws_rules.append(data)
+
+    wb.save(filename)
+    return None
+
+
+init_script()
+wsc_list = get_workskill_list()
+write_xls(args.output, wsc_list)

ofsc/metadata.py

@@ -2,12 +2,19 @@
 import json
 import logging
 import urllib
+from typing import List
 from urllib.parse import urljoin
 
 import requests
 
 from .common import FULL_RESPONSE, JSON_RESPONSE, TEXT_RESPONSE
-from .models import OFSApi, OFSConfig, Workskill
+from .models import (
+    OFSApi,
+    OFSConfig,
+    Workskill,
+    WorkskillCondition,
+    WorskillConditionList,
+)
 
 
 class OFSMetadata(OFSApi):
@@ -176,3 +183,32 @@ def delete_workskill(self, label: str, response_type=FULL_RESPONSE):
             return response.json()
         else:
             return response.text
+
+    # Workskill conditions
+    def get_workskill_conditions(self, response_type=FULL_RESPONSE):
+        url = urljoin(self.baseUrl, f"/rest/ofscMetadata/v1/workSkillConditions")
+        response = requests.get(
+            url,
+            headers=self.headers,
+        )
+        if response_type == FULL_RESPONSE:
+            return response
+        elif response_type == JSON_RESPONSE:
+            return response.json()
+        else:
+            return response.text
+
+    def replace_workskill_conditions(
+        self, data=WorskillConditionList, response_type=FULL_RESPONSE
+    ):
+        url = urljoin(self.baseUrl, f"/rest/ofscMetadata/v1/workSkillConditions")
+        content = '{"items":' + data.json(exclude_none=True) + "}"
+        headers = self.headers
+        headers["Content-Type"] = "application/json"
+        response = requests.put(url, headers=headers, data=content)
+        if response_type == FULL_RESPONSE:
+            return response
+        elif response_type == JSON_RESPONSE:
+            return response.json()
+        else:
+            return response.text

ofsc/models.py

@@ -1,7 +1,8 @@
 import base64
+import typing
 from enum import Enum
 from functools import lru_cache
-from typing import List
+from typing import Any, List
 
 from pydantic import BaseModel, validator
 
@@ -64,3 +65,23 @@ class Workskill(BaseModel):
     @validator("translations", always=True)
     def set_default(cls, field_value, values):
         return field_value or [Translation(name=values["name"])]
+
+
+class Condition(BaseModel):
+    label: str
+    function: str
+    value: Any = None
+    valueList: list = []
+
+
+class WorkskillCondition(BaseModel):
+    internalId: int
+    label: str
+    requiredLevel: int
+    preferableLevel: int
+    conditions: List[Condition]
+    dependencies: Any
+
+
+class WorskillConditionList(BaseModel):
+    __root__: List[WorkskillCondition]

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "ofsc"
-version = "1.17.1"
+version = "1.17.2"
 license = "MIT"
 description = "Python wrapper for Oracle Field Service API"
 authors = ["Borja Toron <borja.toron@gmail.com>"]
@@ -15,6 +15,7 @@ pytest = "^7.1.2"
 pydantic = "^1.9.1"
 
 [tool.poetry.dev-dependencies]
+openpyxl = "^3.0.10"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

tests/conftest.py

@@ -1,7 +1,10 @@
 import logging
 import os
+import pprint
+from http.client import HTTPConnection  # py3
 
 import pytest
+import requests
 from ofsc import OFSC
 
 
@@ -20,3 +23,23 @@ def instance():
 @pytest.fixture
 def current_date():
     return os.environ.get("OFSC_TEST_DATE")
+
+
+@pytest.fixture
+def pp():
+    pp = pprint.PrettyPrinter(indent=4)
+    return pp
+
+
+@pytest.fixture
+def request_logging():
+    log = logging.getLogger("urllib3")
+    log.setLevel(logging.DEBUG)
+
+    # logging from urllib3 to console
+    ch = logging.StreamHandler()
+    ch.setLevel(logging.DEBUG)
+    log.addHandler(ch)
+
+    # print statements from `http.client.HTTPConnection` to console/stdout
+    HTTPConnection.debuglevel = 1

tests/test_metadata.py

@@ -1,7 +1,14 @@
+import json
 import logging
 
-from ofsc.common import FULL_RESPONSE
-from ofsc.models import SharingEnum, Workskill
+from ofsc.common import FULL_RESPONSE, JSON_RESPONSE
+from ofsc.models import (
+    Condition,
+    SharingEnum,
+    Workskill,
+    WorkskillCondition,
+    WorskillConditionList,
+)
 
 
 def test_get_workskills(instance):
@@ -53,3 +60,35 @@ def test_delete_workskill(instance):
         label=skill.label, response_type=FULL_RESPONSE
     )
     assert metadata_response.status_code == 204
+
+
+def test_get_workskill_conditions(instance, pp):
+    logging.info("... get workskill conditions")
+    metadata_response = instance.metadata.get_workskill_conditions(
+        response_type=FULL_RESPONSE
+    )
+    response = metadata_response.json()
+    assert metadata_response.status_code == 200
+    logging.debug(pp.pformat(response))
+    assert response["totalResults"] is not None
+    assert response["totalResults"] == 7
+    for item in response["items"]:
+        logging.debug(pp.pformat(item))
+        ws_item = WorkskillCondition.parse_obj(item)
+        logging.debug(pp.pformat(ws_item))
+        assert ws_item.label == item["label"]
+        for condition in ws_item.conditions:
+            assert type(condition) == Condition
+
+
+def test_replace_workskill_conditions(instance, pp, request_logging):
+    logging.info("... replace workskill conditions")
+    response = instance.metadata.get_workskill_conditions(response_type=JSON_RESPONSE)
+    assert response["totalResults"] is not None
+    assert response["totalResults"] == 7
+    ws_list = WorskillConditionList.parse_obj(response["items"])
+    metadata_response = instance.metadata.replace_workskill_conditions(ws_list)
+    logging.debug(pp.pformat(metadata_response.text))
+    assert metadata_response.status_code == 200
+    assert response["totalResults"] is not None
+    assert response["totalResults"] == 7