Merge pull request #811 from burzekj/genesys_get_users_feature

✨ new logic to extracting users from genesys

CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added `_rename_duplicated_fields` method to `SharepointListToDF` task class for finding and rename duplicated columns
 - Added new view type `agent_interaction_view_type` in `Genesys`source.
 - Added libraries `nltk` and `sklearn` to `requirements`.
+- Added new logic for endpoint `users` in `Genesys`task.
 
 ### Fixed
 - Fixed bug for endpoint `conversations` in GET method in `Genesys` Task.

tests/unit/test_utils.py

@@ -9,6 +9,7 @@
     add_viadot_metadata_columns,
     check_if_empty_file,
     gen_bulk_insert_query_from_df,
+    check_value,
 )
 
 EMPTY_CSV_PATH = "empty.csv"
@@ -153,3 +154,51 @@ def test_add_viadot_metadata_columns_with_parameter():
     assert df_base.columns.to_list() == ["a", "b"]
     assert df_decorated.columns.to_list() == ["a", "b", "_viadot_source"]
     assert df_decorated["_viadot_source"][0] == "Source_name"
+
+
+# Sample test checking the correctness of the function when the key is found
+def test_check_value_found():
+    json_data = {
+        "first_known_lvl": {
+            "second_known_lvl": {"third_known_lvl": {"searched_phrase": "phrase"}}
+        }
+    }
+    result = check_value(
+        json_data["first_known_lvl"]["second_known_lvl"]["third_known_lvl"],
+        ["searched_phrase"],
+    )
+    assert result == "phrase"
+
+
+# Sample test checking the correctness of the function when the key is not found
+def test_check_value_not_found():
+    json_data = {
+        "first_known_lvl": {
+            "second_known_lvl": {
+                "third_known_lvl": {"other_phrase": "This won't be found"}
+            }
+        }
+    }
+    result = check_value(
+        json_data["first_known_lvl"]["second_known_lvl"]["third_known_lvl"],
+        ["searched_phrase"],
+    )
+    assert result is None
+
+
+# Sample test checking the correctness of the function with an empty dictionary
+def test_check_value_empty_dict():
+    json_data = {}
+    result = check_value(json_data, ["searched_phrase"])
+    assert result is None
+
+
+# Sample test checking the correctness of the function with a nonexistent key
+def test_check_value_nonexistent_key():
+    json_data = {
+        "first_known_lvl": {
+            "second_known_lvl": {"third_known_lvl": {"searched_phrase": "phrase"}}
+        }
+    }
+    result = check_value(json_data, ["nonexistent_key"])
+    assert result is None

viadot/tasks/genesys.py

@@ -13,6 +13,7 @@
 
 from viadot.exceptions import APIError
 from viadot.sources import Genesys
+from viadot.utils import check_value
 from viadot.task_utils import *
 
 logger = logging.get_logger()
@@ -524,3 +525,72 @@ def run(
             logger.info("Downloaded the data from the Genesys into the CSV.")
 
             return [file_name]
+
+        elif view_type is None and end_point == "users":
+            # First call to API to get information about amount of pages to extract
+            temp_json = genesys.genesys_api_connection(
+                post_data_list=post_data_list,
+                end_point=f"{end_point}/?pageSize=500&pageNumber=1&expand=presence,dateLastLogin,groups,employerInfo,lasttokenissued&state=any",
+                method="GET",
+            )
+            last_page = temp_json["pageCount"] + 1
+
+            data_list = []
+
+            # For loop to download all pages from Genesys GET API
+            for n in range(1, last_page):
+                json_file = genesys.genesys_api_connection(
+                    post_data_list=post_data_list,
+                    end_point=f"{end_point}/?pageSize=500&pageNumber={n}&expand=presence,dateLastLogin,groups,employerInfo,lasttokenissued&state=any",
+                    method="GET",
+                )
+                logger.info(f"Downloaded: {n} page")
+
+                num_ids = len(json_file["entities"])
+
+                # For loop to extract data from specific page
+                for id in range(0, num_ids):
+                    record_dict = {}
+                    record_dict["Id"] = check_value(json_file["entities"][id], ["id"])
+                    record_dict["Name"] = check_value(
+                        json_file["entities"][id], ["name"]
+                    )
+                    record_dict["DivisionName"] = check_value(
+                        json_file["entities"][id], ["division", "name"]
+                    )
+                    record_dict["Email"] = check_value(
+                        json_file["entities"][id], ["email"]
+                    )
+                    record_dict["State"] = check_value(
+                        json_file["entities"][id], ["state"]
+                    )
+                    record_dict["Title"] = check_value(
+                        json_file["entities"][id], ["title"]
+                    )
+                    record_dict["Username"] = check_value(
+                        json_file["entities"][id], ["username"]
+                    )
+                    record_dict["SystemPresence"] = check_value(
+                        json_file["entities"][id],
+                        ["presence", "presenceDefinition", "systemPresence"],
+                    )
+                    record_dict["DateLastLogin"] = check_value(
+                        json_file["entities"][id], ["dateLastLogin"]
+                    )
+
+                    data_list.append(record_dict)
+
+            df = pd.DataFrame(data_list)
+
+            # data validation function (optional)
+            if validate_df_dict:
+                validate_df.run(df=df, tests=validate_df_dict)
+
+            file_name = "All_Genesys_Users.csv"
+            df.to_csv(
+                os.path.join(file_name),
+                index=False,
+                sep="\t",
+            )
+
+            return [file_name]

viadot/utils.py

@@ -2,7 +2,7 @@
 import os
 import re
 from itertools import chain
-from typing import Any, Callable, Dict, List, Literal
+from typing import Union, Any, Callable, Dict, List, Literal
 
 import pandas as pd
 import prefect
@@ -460,3 +460,28 @@ def get_nested_dict(d):
                 return d
     else:
         return None
+
+
+def check_value(base: Union[Dict, Any], levels: List) -> Union[None, Any]:
+    """
+    Task to extract data from nested json file if there is any under passed parameters.
+    Otherwise return None.
+
+    Args:
+        base (Dict, Any): variable with base lvl of the json, for example:
+                          json_file["first_known_lvl"]["second_known_lvl"]["third_known_lvl"]
+        levels (List): List of potential lower levels of nested json for data retrieval. For example:
+                       ["first_lvl_below_base", "second_lvl_below_base", "searched_phrase"]
+
+    Returns:
+        Union[None, Any]: Searched value for the lowest level, in example data under "searched_phrase" key.
+    """
+
+    for lvl in levels:
+        if isinstance(base, dict):
+            base = base.get(lvl)
+            if base is None:
+                return None
+        else:
+            return base
+    return base