Code revamping (#20)

* Added get_workzones

* completed tests

* Fixed Workskill and added model

* Fixed error in definition

Co-authored-by: Boris Anthony <boris@st3am.us>

Author: btoron

.gitignore

@@ -3,3 +3,4 @@ dist/
 .env
 *.pyc
 .DS_Store
+*.xslx

README.md

@@ -62,14 +62,18 @@ Currently implemented
 
 ### Metadata / properties
     get_properties (self, offset=0, limit=100, response_type=FULL_RESPONSE)
-     get_all_properties(self, initial_offset=0, limit=100)
+    get_all_properties(self, initial_offset=0, limit=100)
 
 ### Metadata / workskills
     get_workskills (self, offset=0, limit=100, response_type=FULL_RESPONSE)
     get_workskill(self, label: str, response_type=FULL_RESPONSE)
     create_or_update_workskill(self, skill: Workskill, response_type=FULL_RESPONSE)
     delete_workskill(self, label: str, response_type=FULL_RESPONSE)
+    get_workskill_conditions(self, response_type=FULL_RESPONSE):
+    replace_workskill_conditions(self, data: WorskillConditionList, response_type=FULL_RESPONSE
 
+### Metadata / workzones
+    get_workzones(self, response_type=FULL_RESPONSE)
 ## Test History
 
 OFS REST API Version | PyOFSC

examples/get_work_skill_conditions.py

@@ -6,7 +6,7 @@
 
 import ofsc
 from ofsc import FULL_RESPONSE, JSON_RESPONSE, OFSC
-from ofsc.models import WorkskillCondition
+from ofsc.models import WorkskillCondition, WorskillConditionList
 from openpyxl import Workbook
 
 from config import Config
@@ -45,11 +45,11 @@ def init_script():
 
 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"]]
+    ws_list = WorskillConditionList.parse_obj(response["items"])
     return ws_list
 
 
-def write_xls(filename: str, wsc_list: List[WorkskillCondition]):
+def write_xls(filename: str, wsc_list: WorskillConditionList):
     def convert(data):
         match data:
             case None:

examples/get_workzones.py

@@ -0,0 +1,84 @@
+#!python
+import argparse
+import logging
+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 Workzone, WorkzoneList
+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_workzone_list():
+    response = instance.metadata.get_workzones(response_type=JSON_RESPONSE)
+    return WorkzoneList.parse_obj(response["items"])
+
+
+def write_xls(filename: str, wz_list: WorkzoneList):
+    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_workzones = wb.active
+    ws_workzones.title = "Workzones"
+    workzone_field_names = None
+    for zone in wz_list:
+        logging.warning(f"ZONE: {zone} {type(zone)}")
+        if workzone_field_names is None:
+            workzone_field_names = list(zone.dict().keys())
+            print(f"Zone: Workzone Fields: {workzone_field_names}")
+            ws_workzones.append(workzone_field_names)
+        data = [convert(value) for value in zone.dict().values()]
+        ws_workzones.append(data)
+    wb.save(filename)
+    return None
+
+
+init_script()
+wsc_list = get_workzone_list()
+write_xls(args.output, wsc_list)

ofsc/__init__.py

@@ -1,10 +1,11 @@
 import base64
 import logging
+from functools import wraps
 from http import client
 from urllib import response
 from warnings import warn
 
-from .common import FULL_RESPONSE, JSON_RESPONSE, TEXT_RESPONSE
+from .common import FULL_RESPONSE, JSON_RESPONSE, TEXT_RESPONSE, wrap_return
 from .core import OFSCore
 from .metadata import OFSMetadata
 from .models import OFSConfig

ofsc/common.py

@@ -1,3 +1,36 @@
-TEXT_RESPONSE=1
-FULL_RESPONSE=2
-JSON_RESPONSE=3
+import logging
+from functools import wraps
+
+TEXT_RESPONSE = 1
+FULL_RESPONSE = 2
+JSON_RESPONSE = 3
+
+
+def wrap_return(*a, **kw):
+    """
+    Decorator @return_as wraps the function
+    and decides the return type and if we launch an exception
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            # Pre:
+            response_type = kwargs.get(
+                "response_type", kw.get("response_type", FULL_RESPONSE)
+            )
+            kwargs.pop("response_type", None)
+            response = func(*args, **kwargs)
+            # post:
+            logging.warning(response)
+            if response_type == FULL_RESPONSE:
+                return response
+            elif response_type == JSON_RESPONSE:
+                return response.json()
+            else:
+                return response.text
+            return result
+
+        return wrapper
+
+    return decorator

ofsc/metadata.py

@@ -7,7 +7,7 @@
 
 import requests
 
-from .common import FULL_RESPONSE, JSON_RESPONSE, TEXT_RESPONSE
+from .common import FULL_RESPONSE, JSON_RESPONSE, TEXT_RESPONSE, wrap_return
 from .models import (
     OFSApi,
     OFSConfig,
@@ -199,7 +199,7 @@ def get_workskill_conditions(self, response_type=FULL_RESPONSE):
             return response.text
 
     def replace_workskill_conditions(
-        self, data=WorskillConditionList, response_type=FULL_RESPONSE
+        self, data: WorskillConditionList, response_type=FULL_RESPONSE
     ):
         url = urljoin(self.baseUrl, f"/rest/ofscMetadata/v1/workSkillConditions")
         content = '{"items":' + data.json(exclude_none=True) + "}"
@@ -212,3 +212,19 @@ def replace_workskill_conditions(
             return response.json()
         else:
             return response.text
+
+    # 202208 Workzones
+    @wrap_return(response_type=JSON_RESPONSE, expected=[200])
+    def get_workzones(
+        self,
+        offset=0,
+        limit=100,
+    ):
+        url = urljoin(self.baseUrl, "/rest/ofscMetadata/v1/workZones")
+        params = {"offset": offset, "limit": limit}
+        response = requests.get(
+            url,
+            headers=self.headers,
+            params=params,
+        )
+        return response

ofsc/models.py

@@ -2,7 +2,7 @@
 import typing
 from enum import Enum
 from functools import lru_cache
-from typing import Any, List
+from typing import Any, List, Optional
 
 from pydantic import BaseModel, validator
 
@@ -45,28 +45,53 @@ class SharingEnum(str, Enum):
 
 
 class TranslationEnum(str, Enum):
-    en: "en"
-    es: "es"
-    pt: "pt"
+    en = "en"
+    es = "es"
+    pt = "pt"
+    fr = "fr"
+    br = "br"
+    el = "el"
 
 
+# Work skills
 class Translation(BaseModel):
     language: TranslationEnum = "en"
     name: str
+    languageISO: Optional[str]
+
+
+class TranslationList(BaseModel):
+    __root__: List[Translation]
+
+    def __iter__(self):
+        return iter(self.__root__)
+
+    def __getitem__(self, item):
+        return self.__root__[item]
 
 
 class Workskill(BaseModel):
     label: str
     active: bool = True
     name: str = ""
     sharing: SharingEnum
-    translations: List[Translation] = []
+    translations: TranslationList = []
 
     @validator("translations", always=True)
     def set_default(cls, field_value, values):
         return field_value or [Translation(name=values["name"])]
 
 
+class WorkskillList(BaseModel):
+    __root__: List[Workskill]
+
+    def __iter__(self):
+        return iter(self.__root__)
+
+    def __getitem__(self, item):
+        return self.__root__[item]
+
+
 class Condition(BaseModel):
     label: str
     function: str
@@ -85,3 +110,28 @@ class WorkskillCondition(BaseModel):
 
 class WorskillConditionList(BaseModel):
     __root__: List[WorkskillCondition]
+
+    def __iter__(self):
+        return iter(self.__root__)
+
+    def __getitem__(self, item):
+        return self.__root__[item]
+
+
+# Workzones
+class Workzone(BaseModel):
+    workZoneLabel: str
+    workZoneName: str
+    status: str
+    travelArea: str
+    keys: List[Any]
+
+
+class WorkzoneList(BaseModel):
+    __root__: List[Workzone]
+
+    def __iter__(self):
+        return iter(self.__root__)
+
+    def __getitem__(self, item):
+        return self.__root__[item]

poetry.lock

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