add new models, change structure, start implementing pagination

Author: maxkahan

testutils/__init__.py

@@ -1,4 +1,4 @@
-from .testutils import build_response
 from .mock_auth import get_mock_api_key_auth, get_mock_jwt_auth
+from .testutils import build_response
 
 __all__ = ['build_response', 'get_mock_api_key_auth', 'get_mock_jwt_auth']

testutils/mock_auth.py

@@ -1,4 +1,4 @@
-from os.path import join, dirname
+from os.path import dirname, join
 
 from vonage_http_client.auth import Auth
 

users/src/vonage_users/common.py

@@ -0,0 +1,70 @@
+from typing import Dict, List, Optional
+
+from pydantic import BaseModel, Field, HttpUrl
+from typing_extensions import Annotated
+
+PhoneNumber = Annotated[str, Field(pattern='^[1-9]\d{6,14}$')]
+
+
+class PstnChannel(BaseModel):
+    number: int
+
+
+class SipChannel(BaseModel):
+    uri: str = Field(..., pattern='^(sip|sips):\+?([\w|:.\-@;,=%&]+)')
+    username: str = None
+    password: str = None
+
+
+class VbcChannel(BaseModel):
+    extension: str
+
+
+class WebsocketChannel(BaseModel):
+    uri: str = Field(pattern='^(ws|wss)://[a-zA-Z0-9~#%@&-_?\/.,:;)(][]*$')
+    content_type: str = Field(pattern="^audio/l16;rate=(8000|16000)$")
+    headers: Optional[Dict[str, str]] = None
+
+
+class SmsChannel(BaseModel):
+    number: PhoneNumber
+
+
+class MmsChannel(BaseModel):
+    number: PhoneNumber
+
+
+class WhatsappChannel(BaseModel):
+    number: PhoneNumber
+
+
+class ViberChannel(BaseModel):
+    number: PhoneNumber
+
+
+class MessengerChannel(BaseModel):
+    id: str
+
+
+class Channels(BaseModel):
+    pstn: Optional[List[PstnChannel]] = None
+    sip: Optional[List[SipChannel]] = None
+    vbc: Optional[List[VbcChannel]] = None
+    websocket: Optional[List[WebsocketChannel]] = None
+    sms: Optional[List[SmsChannel]] = None
+    mms: Optional[List[MmsChannel]] = None
+    whatsapp: Optional[List[WhatsappChannel]] = None
+    viber: Optional[List[ViberChannel]] = None
+    messenger: Optional[List[MessengerChannel]] = None
+
+
+class Properties(BaseModel):
+    custom_data: Optional[Dict[str, str]]
+
+
+class User(BaseModel):
+    name: Optional[str] = Field(None, example="my_user_name")
+    display_name: Optional[str] = Field(None, example="My User Name")
+    image_url: Optional[HttpUrl] = Field(None, example="https://example.com/image.png")
+    properties: Optional[Properties]
+    channels: Optional[Channels]

users/src/vonage_users/requests.py

@@ -1,12 +1,12 @@
 from typing import Literal, Optional
 
-from pydantic import BaseModel, Field, ValidationInfo, field_validator, model_validator
+from pydantic import BaseModel, Field
 
 
 class ListUsersRequest(BaseModel):
     """Request object for listing users."""
 
-    page_size: Optional[int] = Field(None, ge=1, le=100)
+    page_size: Optional[int] = Field(10, ge=1, le=100)
     order: Optional[Literal['asc', 'desc', 'ASC', 'DESC']] = None
     cursor: Optional[str] = Field(
         None,

users/src/vonage_users/responses.py

@@ -35,6 +35,13 @@ class ListUsersResponse(BaseModel):
     links: Links = Field(..., validation_alias='_links')
 
 
+class CreateUserResponse(BaseModel):
+    id: str
+    name: str
+    display_name: str
+    links: UserLinks = Field(..., validation_alias='_links')
+
+
 # class MessageResponse(BaseModel):
 #     to: str
 #     message_id: str = Field(..., validation_alias='message-id')

users/src/vonage_users/users.py

@@ -1,30 +1,69 @@
-from typing import Optional
-from pydantic import validate_call
+from typing import Generator, Literal, Optional
+
+from pydantic import BaseModel, validate_call
 from vonage_http_client.http_client import HttpClient
 
-from .errors import UsersError
+from .common import User
 from .requests import ListUsersRequest
-from .responses import ListUsersResponse
+from .responses import CreateUserResponse, ListUsersResponse
+
+
+class Filters(BaseModel):
+    order: Optional[Literal['asc', 'desc', 'ASC', 'DESC']] = None
+
+
+import urllib.parse
+
+
+def parse_cursor_from_url(url: str) -> Optional[str]:
+    """Extract the cursor from the "next" URL."""
+    query_string = urllib.parse.urlparse(url).query
+    params = urllib.parse.parse_qs(query_string)
+    return params.get('cursor', [None])[0]
 
 
 class Users:
     """Class containing methods for user management.
 
-    When using APIs that require a Vonage Application to be created,
-    you can create users to associate with that application.
+    When using APIs that require a Vonage Application to be created, you can create users to
+    associate with that application.
     """
 
     def __init__(self, http_client: HttpClient) -> None:
         self._http_client = http_client
         self._auth_type = 'jwt'
 
     @validate_call
-    def list_users(self, params: Optional[ListUsersRequest] = None) -> ListUsersResponse:
-        """List all users."""
-        response = self._http_client.get(
+    def list_users(
+        self,
+        order: Literal['asc', 'desc', 'ASC', 'DESC'] = None,
+        name: str = None,
+    ) -> Generator:
+        """List all users with pagination handled by a generator."""
+        cursor = None
+        while True:
+            params = ListUsersRequest(order=order, cursor=cursor, name=name)
+            response = self._http_client.get(
+                self._http_client.api_host,
+                '/v1/users',
+                # need to send the right stuff to the api
+                params.model_dump() if params is not None else None,
+                self._auth_type,
+            )
+            users = ListUsersResponse(**response)
+            for user in users.embedded.users:
+                yield user
+            if not users.links.next:
+                break
+            cursor = parse_cursor_from_url(users.links.next.href)
+
+    @validate_call
+    def create_user(self, params: Optional[User]):
+        """Create a user."""
+        response = self._http_client.post(
             self._http_client.api_host,
             '/v1/users',
             params.model_dump() if params is not None else None,
             self._auth_type,
         )
-        return ListUsersResponse(**response)
+        return CreateUserResponse(**response)

users/tests/test_users.py

@@ -1,15 +1,9 @@
 from os.path import abspath
 
 import responses
-from pydantic import ValidationError
-from pytest import raises
-from vonage_http_client.auth import Auth
-from vonage_http_client.errors import HttpRequestError
 from vonage_http_client.http_client import HttpClient
 from vonage_users import Users
-from vonage_users.errors import UsersError
 from vonage_users.requests import ListUsersRequest
-from vonage_users.responses import ListUsersResponse
 
 from testutils import build_response, get_mock_jwt_auth
 
@@ -33,18 +27,23 @@ def test_create_list_users_request():
 @responses.activate
 def test_list_users():
     build_response(path, 'GET', 'https://api.nexmo.com/v1/users', 'list_users.json')
-    response: ListUsersResponse = users.list_users()
-    assert response.page_size == 10
-    assert response.embedded.users[0].id == 'USR-82e028d9-5201-4f1e-8188-604b2d3471fd'
-    assert response.embedded.users[0].name == 'my_user_name'
-    assert response.embedded.users[0].display_name == 'My User Name'
-    assert (
-        response.embedded.users[0].links.self.href
-        == 'https://api.nexmo.com/v1/users/USR-82e028d9-5201-4f1e-8188-604b2d3471fd'
-    )
-    assert (
-        response.links.self.href
-        == 'https://api.nexmo.com/v1/users?order=desc&page_size=10&cursor=7EjDNQrAcipmOnc0HCzpQRkhBULzY44ljGUX4lXKyUIVfiZay5pv9wg='
-    )
+    users_generator = users.list_users()
+    print(next(users_generator))
+    # for user in users_generator:
+    #     print(user)
+    assert 0
+
+    # assert response.page_size == 10
+    # assert response.embedded.users[0].id == 'USR-82e028d9-5201-4f1e-8188-604b2d3471fd'
+    # assert response.embedded.users[0].name == 'my_user_name'
+    # assert response.embedded.users[0].display_name == 'My User Name'
+    # assert (
+    #     response.embedded.users[0].links.self.href
+    #     == 'https://api.nexmo.com/v1/users/USR-82e028d9-5201-4f1e-8188-604b2d3471fd'
+    # )
+    # assert (
+    #     response.links.self.href
+    #     == 'https://api.nexmo.com/v1/users?order=desc&page_size=10&cursor=7EjDNQrAcipmOnc0HCzpQRkhBULzY44ljGUX4lXKyUIVfiZay5pv9wg='
+    # )
 
     # print(response.model_dump_json(indent=2))