Skip to content

Commit

Permalink
chore: refactor models
Browse files Browse the repository at this point in the history 
- Refactor models to remove redundancy
- Mapping approach for VoiceConfiguration to dynamically select the correct model

Signed-off-by: Jessica Matsuoka <[email protected]>
  • Loading branch information
@matsk-sinch
matsk-sinch committed Feb 3, 2025
1 parent 5e561a8 commit d3a515c
Show file tree
Hide file tree
Showing 21 changed files with 1,025 additions and 163 deletions.
5 changes: 4 additions & 1 deletion .gitignore
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
tox.ini
.nox/
.coverage
.coverage.*
Expand Down Expand Up @@ -132,4 +133,6 @@ cython_debug/
poetry.lock

# .DS_Store files
.DS_Store
.DS_Store

qodana.yaml
79 changes: 0 additions & 79 deletions sinch/core/models/base_model.py
View file
Original file line number Diff line number Diff line change
@@ -1,9 +1,5 @@
import json
import re
from datetime import datetime
from dataclasses import asdict, dataclass
from typing import Any
from pydantic import BaseModel, ConfigDict


@dataclass
Expand All @@ -19,78 +15,3 @@ def as_json(self):
class SinchRequestBaseModel(SinchBaseModel):
def as_dict(self):
return {k: v for k, v in asdict(self).items() if v is not None}


class BaseModelConfigRequest(BaseModel):
"""
A base model that allows extra fields and converts snake_case to camelCase.
"""

@staticmethod
def _to_camel_case(snake_str: str) -> str:
"""Converts snake_case to camelCase."""
components = snake_str.split('_')
return components[0] + ''.join(x.title() for x in components[1:])

model_config = ConfigDict(
# Allows using both alias (camelCase) and field name (snake_case)
populate_by_name=True,
# Allows extra values in input
extra="allow"
)

def model_dump(self, **kwargs) -> dict:
"""Converts extra fields from snake_case to camelCase when dumping the model in endpoint."""
# Get the standard model dump
data = super().model_dump(**kwargs)

# Get extra fields
extra_data = self.__pydantic_extra__ or {}

# Convert extra fields to camelCase and collect the original snake_case keys
converted_extra = {}
for key, value in extra_data.items():
camel_case_key = self._to_camel_case(key)
converted_extra[camel_case_key] = value

# Remove snake_case keys from `data` before merging converted extras
for key in extra_data.keys():
data.pop(key, None) # Ensure snake_case fields are removed from final output

# Merge the cleaned base data with the converted extra fields
return {**data, **converted_extra}


class BaseModelConfigResponse(BaseModel):
"""
A base model that allows extra fields and converts camelCase to snake_case,
and serializes datetime fields to ISO format.
"""

@staticmethod
def datetime_encoder(v: datetime) -> str:
""""Converts a datetime object to a string in ISO 8601 format """
return v.strftime("%Y-%m-%dT%H:%M:%S.%fZ")[:-3] + "Z"

@staticmethod
def _to_snake_case(camel_str: str) -> str:
"""Helper to convert camelCase string to snake_case."""
return re.sub(r'(?<!^)(?=[A-Z])', '_', camel_str).lower()

model_config = ConfigDict(
# Allows using both alias (camelCase) and field name (snake_case)
populate_by_name=True,
# Allows extra values in input
extra="allow",
# Custom datetime serialization
ser_json_typed={datetime: datetime_encoder}
)

def model_post_init(self, __context: Any) -> None:
""" Converts unknown fields from camelCase to snake_case."""
if self.__pydantic_extra__:
converted_extra = {
self._to_snake_case(key): value for key, value in self.__pydantic_extra__.items()
}
self.__pydantic_extra__.clear()
self.__pydantic_extra__.update(converted_extra)
203 changes: 181 additions & 22 deletions sinch/domains/numbers/available_numbers.py
View file
Original file line number Diff line number Diff line change
@@ -1,31 +1,61 @@
from typing import Optional, TypedDict, overload
from typing import Optional, TypedDict, overload, Literal, Union, Annotated
from typing_extensions import NotRequired
from pydantic import conlist, StrictInt, StrictStr
from pydantic import conlist, StrictInt, StrictStr, Field
from sinch.domains.numbers.endpoints.available.search_for_number import SearchForNumberEndpoint
from sinch.domains.numbers.endpoints.available.list_available_numbers import AvailableNumbersEndpoint
from sinch.domains.numbers.endpoints.available.activate_number import ActivateNumberEndpoint
from sinch.domains.numbers.endpoints.available.rent_any_number import RentAnyNumberEndpoint

from sinch.domains.numbers.models.available.list_available_numbers_request import ListAvailableNumbersRequest
from sinch.domains.numbers.models.available.activate_number_request import ActivateNumberRequest
from sinch.domains.numbers.models.available.check_number_availability_request import CheckNumberAvailabilityRequest
from sinch.domains.numbers.models.available.rent_any_number_request import RentAnyNumberRequest

from sinch.domains.numbers.models.available.list_available_numbers_response import ListAvailableNumbersResponse
from sinch.domains.numbers.models.available.activate_number_response import ActivateNumberResponse
from sinch.domains.numbers.models.available.check_number_availability_response import CheckNumberAvailabilityResponse
from sinch.domains.numbers.models.available.rent_any_number_response import RentAnyNumberResponse

# Define type aliases
NumberType = Union[Literal["MOBILE", "LOCAL", "TOLL_FREE"], StrictStr]
CapabilityType = conlist(Union[Literal["SMS", "VOICE"], StrictStr], min_length=1)
NumberSearchPatternType = Union[Literal["START", "CONTAINS", "END"], StrictStr]


class SmsConfigurationDict(TypedDict):
service_plan_id: str
campaign_id: NotRequired[str]


class VoiceConfigurationDict(TypedDict):
type: str
class VoiceConfigurationDictRTC(TypedDict):
type: Literal["RTC"]
app_id: NotRequired[str]


class VoiceConfigurationDictEST(TypedDict):
type: Literal["EST"]
trunk_id: NotRequired[str]


class VoiceConfigurationDictFAX(TypedDict):
type: Literal["FAX"]
service_id: NotRequired[str]


class VoiceConfigurationDictCustom(TypedDict):
type: str


class NumberPatternDict(TypedDict):
pattern: NotRequired[str]
search_pattern: NotRequired[str]
search_pattern: NotRequired[NumberSearchPatternType]


VoiceConfigurationDictType = Annotated[
Union[VoiceConfigurationDictFAX, VoiceConfigurationDictRTC,
VoiceConfigurationDictEST, VoiceConfigurationDictCustom],
Field(discriminator="type")
]


class AvailableNumbers:
Expand Down Expand Up @@ -53,23 +83,24 @@ def _request(self, endpoint_class, request_data):
def list(
self,
region_code: StrictStr,
number_type: StrictStr,
number_type: NumberType,
number_pattern: Optional[StrictStr] = None,
number_search_pattern: Optional[StrictStr] = None,
capabilities: Optional[conlist] = None,
number_search_pattern: Optional[NumberSearchPatternType] = None,
capabilities: Optional[CapabilityType] = None,
page_size: Optional[StrictInt] = None,
**kwargs
) -> ListAvailableNumbersResponse:
"""
Search for available virtual numbers for you to activate using a variety of parameters to filter results.
Args:
region_code (str): ISO 3166-1 alpha-2 country code of the phone number.
number_type (str): Type of number (MOBILE, LOCAL, TOLL_FREE).
number_pattern (str): Specific sequence of digits to search for.
number_search_pattern (str): Pattern to apply (START, CONTAIN, END).
capabilities (list): Capabilities (SMS, VOICE) required for the number.
page_size (int): Maximum number of items to return.
region_code (StrictStr): ISO 3166-1 alpha-2 country code of the phone number.
number_type (NumberType): Type of number (e.g., "MOBILE", "LOCAL", "TOLL_FREE").
number_pattern (Optional[StrictStr]): Specific sequence of digits to search for.
number_search_pattern (Optional[NumberSearchPatternType]):
Pattern to apply (e.g., "START", "CONTAINS", "END").
capabilities (Optional[CapabilityType]): Capabilities required for the number. (e.g., ["SMS", "VOICE"])
page_size (StrictInt): Maximum number of items to return.
**kwargs: Additional filters for the request.
Returns:
Expand All @@ -93,8 +124,8 @@ def list(
def activate(
self,
phone_number: StrictStr,
sms_configuration: None = None,
voice_configuration: None = None,
sms_configuration: None,
voice_configuration: None,
callback_url: Optional[StrictStr] = None
) -> ActivateNumberResponse:
pass
Expand All @@ -104,7 +135,27 @@ def activate(
self,
phone_number: StrictStr,
sms_configuration: SmsConfigurationDict,
voice_configuration: VoiceConfigurationDict,
voice_configuration: VoiceConfigurationDictEST,
callback_url: Optional[StrictStr] = None
) -> ActivateNumberResponse:
pass

@overload
def activate(
self,
phone_number: StrictStr,
sms_configuration: SmsConfigurationDict,
voice_configuration: VoiceConfigurationDictFAX,
callback_url: Optional[StrictStr] = None
) -> ActivateNumberResponse:
pass

@overload
def activate(
self,
phone_number: StrictStr,
sms_configuration: SmsConfigurationDict,
voice_configuration: VoiceConfigurationDictRTC,
callback_url: Optional[StrictStr] = None
) -> ActivateNumberResponse:
pass
Expand All @@ -113,18 +164,25 @@ def activate(
self,
phone_number: StrictStr,
sms_configuration: Optional[SmsConfigurationDict] = None,
voice_configuration: Optional[VoiceConfigurationDict] = None,
voice_configuration: Optional[VoiceConfigurationDictType] = None,
callback_url: Optional[StrictStr] = None,
**kwargs
) -> ActivateNumberResponse:
"""
Activate a virtual number to use with SMS products, Voice products, or both.
Activate a virtual number to use with SMS, Voice, or both products.
Args:
phone_number (StrictStr): The phone number in E.164 format with leading +.
sms_configuration (SmsConfigurationDict): Configuration for SMS activation.
voice_configuration (VoiceConfigurationDict): Configuration for Voice activation.
callback_url (StrictStr): The callback URL to be called.
sms_configuration (Optional[SmsConfigurationDict]): A dictionary defining the SMS configuration.
Including fields such as:
- service_plan_id (str): The service plan ID.
- campaign_id (Optional[str]): The campaign ID.
voice_configuration (Optional[VoiceConfigurationDictType]): A dictionary defining the Voice configuration.
Supported types include:
- `VoiceConfigurationDictRTC`: type 'RTC' with an `app_id` field.
- `VoiceConfigurationDictEST`: type 'EST' with a `trunk_id` field.
- `VoiceConfigurationDictFAX`: type 'FAX' with a `service_id` field.
callback_url (Optional[StrictStr]): The callback URL to be called.
**kwargs: Additional parameters for the request.
Returns:
Expand All @@ -141,6 +199,107 @@ def activate(
)
return self._request(ActivateNumberEndpoint, request_data)

@overload
def rent_any(
self,
region_code: StrictStr,
type_: NumberType,
sms_configuration: None,
voice_configuration: None,
number_pattern: Optional[NumberPatternDict] = None,
capabilities: Optional[CapabilityType] = None,
callback_url: Optional[str] = None,
) -> RentAnyNumberResponse:
pass

@overload
def rent_any(
self,
region_code: StrictStr,
type_: NumberType,
sms_configuration: SmsConfigurationDict,
voice_configuration: VoiceConfigurationDictRTC,
number_pattern: Optional[NumberPatternDict] = None,
capabilities: Optional[CapabilityType] = None,
callback_url: Optional[str] = None,
) -> RentAnyNumberResponse:
pass

@overload
def rent_any(
self,
region_code: StrictStr,
type_: NumberType,
sms_configuration: SmsConfigurationDict,
voice_configuration: VoiceConfigurationDictFAX,
number_pattern: Optional[NumberPatternDict] = None,
capabilities: Optional[CapabilityType] = None,
callback_url: Optional[str] = None,
) -> RentAnyNumberResponse:
pass

@overload
def rent_any(
self,
region_code: StrictStr,
type_: NumberType,
sms_configuration: SmsConfigurationDict,
voice_configuration: VoiceConfigurationDictEST,
number_pattern: Optional[NumberPatternDict] = None,
capabilities: Optional[CapabilityType] = None,
callback_url: Optional[str] = None,
) -> RentAnyNumberResponse:
pass

def rent_any(
self,
region_code: StrictStr,
type_: NumberType,
number_pattern: Optional[NumberPatternDict] = None,
capabilities: Optional[CapabilityType] = None,
sms_configuration: Optional[SmsConfigurationDict] = None,
voice_configuration: Optional[VoiceConfigurationDictType] = None,
callback_url: Optional[str] = None,
**kwargs
) -> RentAnyNumberResponse:
"""
Search for and activate an available Sinch virtual number all in one API call.
Currently, the rentAny operation works only for US 10DLC numbers
Args:
region_code (str): ISO 3166-1 alpha-2 country code of the phone number.
type_ (NumberType): Type of number (e.g., "MOBILE", "LOCAL", "TOLL_FREE").
number_pattern (Optional[NumberPatternDict]): Specific sequence of digits to search for.
capabilities (Optional[CapabilityType]): Capabilities required for the number. (e.g., ["SMS", "VOICE"])
sms_configuration (Optional[SmsConfigurationDict]): A dictionary defining the SMS configuration.
Including fields such as:
- service_plan_id (str): The service plan ID.
- campaign_id (Optional[str]): The campaign ID.
voice_configuration (Optional[VoiceConfigurationDictType]): A dictionary defining the Voice configuration.
Supported types include:
- `VoiceConfigurationDictRTC`: type 'RTC' with an `app_id` field.
- `VoiceConfigurationDictEST`: type 'EST' with a `trunk_id` field.
- `VoiceConfigurationDictFAX`: type 'FAX' with a `service_id` field.
callback_url (str): The callback URL to receive notifications.
**kwargs: Additional parameters for the request.
Returns:
RentAnyNumberRequest: A response object with the activated number and its details.
For detailed documentation, visit https://developers.sinch.com
"""
request_data = RentAnyNumberRequest(
region_code=region_code,
type_=type_,
number_pattern=number_pattern,
capabilities=capabilities,
sms_configuration=sms_configuration,
voice_configuration=voice_configuration,
callback_url=callback_url,
**kwargs
)
return self._request(RentAnyNumberEndpoint, request_data)

def check_availability(self, phone_number: StrictStr, **kwargs) -> CheckNumberAvailabilityResponse:
"""
Enter a specific phone number to check availability.
Expand Down
Loading

0 comments on commit d3a515c

Please sign in to comment.