Added bulk update (#28)

btoron · Aug 1, 2023 · 0d5a386 · 0d5a386
1 parent 61b528e
commit 0d5a386
Show file tree

Hide file tree

Showing 4 changed files with 143 additions and 2 deletions.
diff --git a/ofsc/core.py b/ofsc/core.py
@@ -7,7 +7,7 @@
 import requests
 
 from .common import FULL_RESPONSE, JSON_RESPONSE, TEXT_RESPONSE, wrap_return
-from .models import OFSApi, OFSConfig
+from .models import BulkUpdateRequest, OFSApi, OFSConfig
 
 
 class OFSCore(OFSApi):
@@ -483,3 +483,16 @@ def get_subscription_details(self, subscription_id):
         )
         response = requests.get(url, headers=self.headers)
         return response
+
+    ###
+    # 2. Core / Activities
+    ###
+
+    @wrap_return(response_type=JSON_RESPONSE, expected=[200])
+    def bulk_update(self, data: BulkUpdateRequest):
+        url = urljoin(
+            self.baseUrl,
+            "/rest/ofscCore/v1/activities/custom-actions/bulkUpdate",
+        )
+        response = requests.post(url, headers=self.headers, data=data.json())
+        return response
diff --git a/ofsc/models.py b/ofsc/models.py
@@ -288,3 +288,57 @@ def __iter__(self):
 
     def __getitem__(self, item):
         return self.__root__[item]
+
+
+# Core / Activities
+class BulkUpdateActivityItem(BaseModel):
+    activityId: Optional[int]
+    activityType: Optional[str]
+    date: Optional[str]
+
+    class Config:
+        extra = Extra.allow  # or 'allow' str
+
+
+# CORE / BulkUpdaterequest
+
+
+class BulkUpdateParameters(BaseModel):
+    fallbackResource: Optional[str]
+    identifyActivityBy: Optional[str]
+    ifExistsThenDoNotUpdateFields: Optional[List[str]]
+    ifInFinalStatusThen: Optional[str]
+    inventoryPropertiesUpdateMode: Optional[str]
+
+
+class BulkUpdateRequest(BaseModel):
+    activities: List[BulkUpdateActivityItem]
+    updateParameters: BulkUpdateParameters
+
+
+class ActivityKeys(BaseModel):
+    activityId: Optional[int]
+    apptNumber: Optional[str]
+    customerNumber: Optional[str]
+
+
+class BulkUpdateError(BaseModel):
+    errorDetail: Optional[str]
+    operation: Optional[str]
+
+
+class BulkUpdateWarning(BaseModel):
+    code: Optional[int]
+    message: Optional[int]
+
+
+class BulkUpdateResult(BaseModel):
+    activityKeys: Optional[ActivityKeys]
+    errors: Optional[List[BulkUpdateError]]
+    operationsFailed: Optional[List[str]]
+    operationsPerformed: Optional[List[str]]
+    warnings: Optional[List[BulkUpdateWarning]]
+
+
+class BulkUpdateResponse(BaseModel):
+    results: Optional[List[BulkUpdateResult]]
diff --git a/pyproject.toml b/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "ofsc"
-version = "1.19"
+version = "1.20"
 license = "MIT"
 description = "Python wrapper for Oracle Field Service API"
 authors = ["Borja Toron <borja.toron@gmail.com>"]

diff --git a/tests/test_core_activities.py b/tests/test_core_activities.py
@@ -4,6 +4,7 @@
 import pytest
 
 from ofsc.common import FULL_RESPONSE
+from ofsc.models import BulkUpdateRequest, BulkUpdateResponse
 
 
 def test_search_activities_001(instance):
@@ -82,3 +83,76 @@ def test_get_activities_offset(instance, current_date, demo_data, request_loggin
         "activityId": expected_id,
         "resourceId": expected_postalcode,
     }
+
+
+def test_model_bulk_update_simple(instance, request_logging):
+    logging.info("...104. Bulk Update")
+    data = {
+        "updateParameters": {
+            "identifyActivityBy": "apptNumber",
+            "ifInFinalStatusThen": "doNothing",
+        },
+        "activities": [
+            {
+                "apptNumber": "XS3453456",
+                "resourceId": "CAUSA",
+                "date": "2023-08-09",
+                "activityType": "09",
+                "timeZone": "America/New_York",
+                "language": "en-US",
+                "inventories": {
+                    "items": [
+                        {
+                            "inventoryType": "RG6BLK",
+                            "inventory_model": "Z222",
+                            "quantity": 10,
+                        },
+                        {
+                            "inventoryType": "RG6BLK",
+                            "inventory_model": "X333",
+                            "quantity": 5,
+                        },
+                    ]
+                },
+                "requiredInventories": {
+                    "items": [
+                        {"inventoryType": "NS", "model": "A22", "quantity": 10},
+                        {"inventoryType": "NS", "model": "B22", "quantity": 10},
+                    ]
+                },
+                "linkedActivities": {
+                    "items": [
+                        {
+                            "fromActivity": {
+                                "apptNumber": "A0001",
+                                "linkType": "start-after",
+                            }
+                        },
+                        {
+                            "toActivity": {
+                                "apptNumber": "B0002",
+                                "linkType": "start-after",
+                            }
+                        },
+                    ]
+                },
+                "resourcePreferences": {
+                    "items": [
+                        {
+                            "resourceId": "testActivityCustomAcQYNWZCUYYP",
+                            "preferenceType": "preferred",
+                        },
+                        {
+                            "resourceId": "testActivityCustomAcCKIVJZFSSB",
+                            "preferenceType": "preferred",
+                        },
+                    ]
+                },
+            }
+        ],
+    }
+    input = BulkUpdateRequest.parse_obj(data)
+    raw_response = instance.core.bulk_update(input, response_type=FULL_RESPONSE)
+    assert raw_response.status_code == 200
+    response = raw_response.json()
+    output = BulkUpdateResponse.parse_obj(response)