feat(portaswitch): add endpoint to retrieve user contact by ID

Author: yakimenko73

app/bss/adapters/__init__.py

@@ -246,6 +246,11 @@ def retrieve_contacts(self, session: SessionInfo, user: UserInfo) -> List[Contac
         """List of other extensions in the PBX"""
         raise NotImplementedError("Override this method in your sub-class")
 
+    @abstractmethod
+    def retrieve_contact_by_user_id(self, session: SessionInfo, user: UserInfo, user_id: str) -> ContactInfo:
+        """Retrieve extension by User ID in the PBX"""
+        raise NotImplementedError("Override this method in your sub-class")
+
     @abstractmethod
     def retrieve_calls(
             self,

app/bss/adapters/portaswitch/adapter.py

@@ -540,6 +540,14 @@ def retrieve_contacts(self, session: SessionInfo, user: UserInfo) -> list[Contac
 
             raise error
 
+    def retrieve_contact_by_user_id(self, session: SessionInfo, user: UserInfo, user_id: str) -> ContactInfo:
+        """Retrieve extension by User ID in the PBX"""
+        account_info = self._admin_api.get_account_info(i_account=int(user_id)).get("account_info")
+        if not account_info:
+            raise WebTritErrorException(404, f"There is no an account with such id: {user_id}")
+
+        return Serializer.get_contact_info_by_account(account_info, int(user.user_id))
+
     def retrieve_calls(
             self,
             session: SessionInfo,

app/bss/adapters/portaswitch/api/admin.py

@@ -146,7 +146,13 @@ def get_account_info(self, **params) -> dict:
             dict: The API method execution result that contains an account info.
         """
         return self._send_request(
-            module="Account", method="get_account_info", params={"without_service_features": 1, **params}
+            module="Account",
+            method="get_account_info",
+            params={
+                "without_service_features": 1,
+                "detailed_info": 1,
+                **params
+            }
         )
 
     def get_env_info(self) -> dict:

app/main.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import logging
 import os
 import sys
 from datetime import datetime
@@ -59,6 +58,7 @@
     UpdateSessionNotFoundErrorResponse,
     UpdateSessionUnprocessableEntityErrorResponse,
     Contacts,
+    ContactInfo,
     # signup
     UserCreateRequest,
     UserCreateResponse,
@@ -193,7 +193,7 @@ def create_session(
 
     if eval_as_bool(config.get_conf_val("Capabilities",
                                         "Password", "Force",
-                            default="False")):
+                                        default="False")):
         # forcefully allow this method to be called even if the
         # adapter declares it as not implemented. This is useful
         # in case when app UI should not display this type of login
@@ -205,7 +205,7 @@ def create_session(
         is_method_allowed(Capabilities.passwordSignin)
 
     user_ref = safely_extract_scalar_value(body.user_ref) or \
-        safely_extract_scalar_value(body.login)
+               safely_extract_scalar_value(body.login)
     if not (user_ref and body.password):
         # missing parameters
         raise_webtrit_error(422, "Missing username & password")
@@ -579,6 +579,49 @@ def get_user_contact_list(
     return Contacts(items=[], )
 
 
+@router.get(
+    '/user/contacts/{user_id}',
+    response_model=ContactInfo,
+    responses={
+        '401': {'model': GetUserContactListUnauthorizedErrorResponse},
+        '404': {'model': GetUserContactListNotFoundErrorResponse},
+        '422': {'model': GetUserContactListUnprocessableEntityErrorResponse},
+        '500': {'model': GetUserContactListInternalServerErrorErrorResponse},
+    },
+    tags=['user'],
+)
+def get_user_contact_by_id(
+        user_id: str,
+        auth_data: HTTPAuthorizationCredentials = Depends(security),
+        x_webtrit_tenant_id: Optional[str] = Header(None, alias=TENANT_ID_HTTP_HEADER),
+) -> (
+        Union[
+            ContactInfo,
+            GetUserContactListUnauthorizedErrorResponse,
+            GetUserContactListNotFoundErrorResponse,
+            GetUserContactListUnprocessableEntityErrorResponse,
+            GetUserContactListInternalServerErrorErrorResponse,
+        ]
+):
+    """
+    Get contact information by user ID
+    """
+    global bss
+
+    access_token = auth_data.credentials
+    session = bss.validate_session(access_token)
+
+    contact = bss.retrieve_contact_by_user_id(
+        session,
+        ExtendedUserInfo(
+            user_id=safely_extract_scalar_value(session.user_id),
+            tenant_id=bss.default_id_if_none(x_webtrit_tenant_id)
+        ),
+        user_id
+    )
+    return contact
+
+
 @router.get(
     '/user/history',
     response_model=Calls,
@@ -949,7 +992,7 @@ def create_user_event(
 ) -> Union[
     Response,
     CreateUserEventUnauthorizedErrorResponse,
-    CreateUserEventUnprocessableEntityErrorResponse, 
+    CreateUserEventUnprocessableEntityErrorResponse,
     CreateUserEventInternalServerErrorErrorResponse,
 ]:
     """