refactor(portaswitch): enhance type handling for user_id and access_token

Author: yakimenko73

app/bss/adapters/portaswitch/adapter.py

@@ -1,5 +1,4 @@
 import logging
-from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timedelta, UTC
 from typing import Final, Iterator, Optional, Dict
 
@@ -8,7 +7,16 @@
 from app_config import AppConfig
 from bss.adapters import BSSAdapter
 from bss.dbs import TiedKeyValue
-from bss.models import DeliveryChannel, SipServer, UserCreateResponse, CustomRequest, CustomResponse, CustomPage
+from bss.models import (
+    DeliveryChannel,
+    SipServer,
+    UserCreateResponse,
+    CustomRequest,
+    CustomResponse,
+    CustomPage,
+    UserId,
+    AccessToken,
+)
 from bss.types import (
     CallRecordingId,
     Capabilities,
@@ -43,11 +51,11 @@
 
 
 class PortaSwitchAdapter(BSSAdapter):
-    """Connects WebTrit and PortaSwitch. Authenticate a user using his/her data in PortaSwitch,
-    retrieve user's SIP credentials to be used by WebTrit and return a list of other configured
-    extenstions (to be provided as 'Cloud PBX' contacts).
-    Currently does not support OTP login.
+    """Bridges WebTrit Core with PortaSwitch APIs.
 
+    Provides authentication (password, OTP, auto-provision), session lifecycle management,
+    SIP credential retrieval, contacts, call history, voicemail, notifications, and other
+    capabilities required by WebTrit clients.
     """
 
     VERSION: Final[str] = "0.1.20"
@@ -129,8 +137,8 @@ def authenticate(self, user: UserInfo, password: str = None) -> SessionInfo:
             session_data = self._account_api.login(account_info["login"], account_info["password"])
 
             return SessionInfo(
-                user_id=account_info["i_account"],
-                access_token=session_data["access_token"],
+                user_id=UserId(str(account_info["i_account"])),
+                access_token=AccessToken(session_data["access_token"]),
                 refresh_token=session_data["refresh_token"],
                 expires_at=datetime.now() + timedelta(seconds=session_data["expires_in"]),
             )
@@ -224,7 +232,7 @@ def validate_otp(self, otp: OTPVerifyRequest) -> SessionInfo:
             session_data: dict = self._account_api.login(account_info["login"], account_info["password"])
 
             return SessionInfo(
-                user_id=account_info["i_account"],
+                user_id=str(account_info["i_account"]),
                 access_token=session_data["access_token"],
                 refresh_token=session_data["refresh_token"],
                 expires_at=datetime.now() + timedelta(seconds=session_data["expires_in"]),
@@ -264,7 +272,7 @@ def validate_session(self, access_token: str) -> SessionInfo:
                     code="access_token_invalid",
                 )
 
-            return SessionInfo(user_id=user_id, access_token=access_token)
+            return SessionInfo(user_id=str(user_id), access_token=access_token)
         except ExpiredSignatureError:
             raise WebTritErrorException(
                 status_code=401,
@@ -303,7 +311,7 @@ def refresh_session(self, refresh_token: str) -> SessionInfo:
             account_info: dict = self._account_api.get_account_info(access_token=access_token)["account_info"]
 
             return SessionInfo(
-                user_id=account_info["i_account"],
+                user_id=str(account_info["i_account"]),
                 access_token=session_data["access_token"],
                 refresh_token=session_data["refresh_token"],
                 expires_at=datetime.now() + timedelta(seconds=session_data["expires_in"]),

app/bss/adapters/portaswitch/serializer.py

@@ -1,7 +1,7 @@
 from datetime import datetime, timezone
 from typing import Optional
 
-from bss.models import SIPTransport
+from bss.models import SIPTransport, UserId
 from bss.types import (
     Balance,
     BalanceType,
@@ -20,7 +20,6 @@
     VoicemailMessageAttachment,
     Direction,
 )
-
 from .types import PortaSwitchMailboxMessageFlag
 
 #: dict: Contains a map between a PortaSwitch AccountInfo.billing_model and BalanceType.
@@ -100,7 +99,7 @@ def get_contact_info_by_account(account_info: dict, current_user: int) -> Contac
         """
 
         return ContactInfo(
-            user_id=account_info["i_account"],
+            user_id=UserId(str(account_info["i_account"])),
             is_current_user=account_info["i_account"] == current_user,
             alias_name=account_info.get("extension_name"),
             company_name=account_info.get("companyname", ""),
@@ -132,9 +131,11 @@ def get_contact_info_by_extension(extension_info: dict, aliases: list, current_u
             ContactInfo: The filled structure of ContactInfo.
         """
 
+        i_account = extension_info.get("i_account")
+
         return ContactInfo(
-            user_id=extension_info.get("i_account"),
-            is_current_user=extension_info.get("i_account") == current_user,
+            user_id=UserId(str(i_account)) if i_account else None,
+            is_current_user=i_account == current_user,
             alias_name=extension_info.get("name", ""),
             first_name=extension_info.get("firstname", ""),
             last_name=extension_info.get("lastname", ""),

app/bss/models.py

@@ -140,7 +140,7 @@ class UserId(RootModel[str]):
     The **Adaptee** must consistently return the same `UserId` for the same user,
     regardless of the `UserRef` used for sign-in.
     """
-    
+
     model_config = {"json_schema_extra": {"example": "123456789abcdef0123456789abcdef0"}}
 
 
@@ -182,7 +182,7 @@ class UserRef(RootModel[str]):
     with the user. When the same user is accessed using different references,
     it is crucial to ensure that the same `UserId` is assigned to this user.
     """
-    
+
     model_config = {"json_schema_extra": {"example": "1234567890"}}
 
 
@@ -272,7 +272,8 @@ class BalanceType(Enum):
 
 
 class Balance(BaseModel):
-    amount: Optional[float] = Field(None, description="The user's current balance.", json_schema_extra={"example": "50.00"})
+    amount: Optional[float] = Field(None, description="The user's current balance.",
+                                    json_schema_extra={"example": "50.00"})
     balance_type: Optional[BalanceType] = Field(
         None,
         description="Meaning of the balance figure for this user.\n\n* `inapplicable` means the **Adaptee** does not handle\n  billing and does not have the balance data.\n* `prepaid` means the number reflects the funds that\n  the user has available for spending.\n* `postpaid` means the balance reflects the amount of\n  previously accumulated charges (how much the user\n  owes - to be used in conjunction with a `credit_limit`).\n",
@@ -352,7 +353,8 @@ class SipInfo(BaseModel):
         description="The visible identification of the caller to be included in the SIP request.\nThis will be shown to the called party as the caller's name. If not provided,\nthe `display_name` will be populated with the `username`.\n",
         json_schema_extra={"example": "Thomas A. Anderson"},
     )
-    password: str = Field(..., description="The password for the SIP account.", json_schema_extra={"example": "strong_password"})
+    password: str = Field(..., description="The password for the SIP account.",
+                          json_schema_extra={"example": "strong_password"})
     transport: Optional[SIPTransport] = Field(
         default_factory=lambda: SIPTransport.UDP,
         description="The transport protocol for SIP communication. UDP by default.",
@@ -542,7 +544,7 @@ class OtpId(RootModel[str]):
     Note: This ID is NOT the code that the user will enter. It serves
     to match the originally generated OTP with the one provided by the user.
     """
-    
+
     model_config = {"json_schema_extra": {"example": "12345678-9abc-def0-1234-56789abcdef0"}}
 
 
@@ -657,7 +659,10 @@ class SessionOtpVerifyRequest(BaseModel):
 
 
 class Contact(BaseModel):
-    user_id: Optional[UserId]
+    user_id: Optional[UserId] = Field(
+        default=None,
+        description="Unique identifier of the contact on the **Adaptee**. Optional for synthetic entries.",
+    )
     is_current_user: Optional[bool] = Field(
         None, description="Indicates whether the contact is associated with the same user who making the request."
     )
@@ -671,8 +676,10 @@ class Contact(BaseModel):
         description="The name of the company the user is associated with.",
         json_schema_extra={"example": "Matrix"},
     )
-    email: Optional[EmailStr] = Field(None, description="The user's email address.", json_schema_extra={"example": "a.black@matrix.com"})
-    first_name: Optional[str] = Field(None, description="The user's first name.", json_schema_extra={"example": "Annabelle"})
+    email: Optional[EmailStr] = Field(None, description="The user's email address.",
+                                      json_schema_extra={"example": "a.black@matrix.com"})
+    first_name: Optional[str] = Field(None, description="The user's first name.",
+                                      json_schema_extra={"example": "Annabelle"})
     last_name: Optional[str] = Field(None, description="The user's last name.", json_schema_extra={"example": "Black"})
     numbers: Numbers
     sip_status: Optional[SipStatus] = Field(
@@ -688,10 +695,14 @@ class UserInfoShowResponse(BaseModel):
         json_schema_extra={"example": "CTO"},
     )
     balance: Optional[Balance] = None
-    company_name: Optional[str] = Field(None, description="The company the user is associated with.", json_schema_extra={"example": "Matrix"})
-    email: Optional[EmailStr] = Field(None, description="The user's email address.", json_schema_extra={"example": "neo@matrix.com"})
-    first_name: Optional[str] = Field(None, description="The user's first name.", json_schema_extra={"example": "Thomas"})
-    last_name: Optional[str] = Field(None, description="The user's last name.", json_schema_extra={"example": "Anderson"})
+    company_name: Optional[str] = Field(None, description="The company the user is associated with.",
+                                        json_schema_extra={"example": "Matrix"})
+    email: Optional[EmailStr] = Field(None, description="The user's email address.",
+                                      json_schema_extra={"example": "neo@matrix.com"})
+    first_name: Optional[str] = Field(None, description="The user's first name.",
+                                      json_schema_extra={"example": "Thomas"})
+    last_name: Optional[str] = Field(None, description="The user's last name.",
+                                     json_schema_extra={"example": "Anderson"})
     numbers: Numbers
     sip: SipInfo
     status: Optional[Status] = Field(