Create Nylas credentials API

nylas/models/credentials.py

@@ -0,0 +1,82 @@
+from dataclasses import dataclass
+from type import List, Any, Dict, Optional, Literal, Union, NotRequired
+
+from dataclasses_json import dataclass_json
+from typing_extensions import TypedDict
+
+
+CredentialType = Literal["adminconsent", "serviceaccount", "connector"]
+
+@dataclass_json
+@dataclass
+class Credential:
+    """
+    Interface representing a Nylas Credential object.
+    Attributes
+    id: Globally unique object identifier;
+    name: Name of the credential
+    credential_type: The type of credential
+    hashed_data: Hashed value of the credential that you created
+    created_at: Timestamp of when the credential was created
+    updated_at: Timestamp of when the credential was updated;
+    """
+    id: str
+    name: str
+    credential_type: Optional[CredentialType]
+    hashed_data: Optional[str] = None
+    created_at: Optional[int] = None
+    updated_at: Optional[int] = None
+
+
+class MicrosoftAdminConsentSettings(TypedDict):
+    client_id: str
+    client_secret: str
+
+
+class CreateCredentialBaseRequest(TypedDict):
+    """
+        Interface representing a request to create a credential.
+
+        Attributes:
+            provider: OAuth provider.
+            settings: Settings required to create a credential
+            credential_type: Type of credential you want to create.
+            credential_data: The data required to successfully create the credential object
+    """
+    provider: str
+    name: Optional[str]
+    credential_type: CredentialType
+
+
+class GoogleServiceAccountCredential(CreateCredentialBaseRequest):
+    credential_data: Dict[str, any]
+
+
+class AdminConsentCredential(CreateCredentialBaseRequest):
+    credential_data: MicrosoftAdminConsentSettings
+
+
+class ConnectorOverrideCredential(CreateCredentialBaseRequest):
+    credential_data: Dict[str, any]
+
+
+class ListCredentialQueryParams(TypedDict):
+    """
+    Interface representing the query parameters for credentials .
+
+    Attributes:
+        offset: Offset results
+        sort_by: Sort entries by field name
+        order_by: Order results by the specified field.
+            Currently only start is supported.
+        limit: The maximum number of objects to return.
+            This field defaults to 50. The maximum allowed value is 200.
+    """
+
+    limit: NotRequired[int]
+    offset: NotRequired[int]
+    order_by: NotRequired[str]
+    sort_by: NotRequired[str]
+
+
+CredentialRequest = Union[AdminConsentCredential, GoogleServiceAccountCredential, ConnectorOverrideCredential]

nylas/resources/credentials.py

@@ -0,0 +1,44 @@
+from nylas.handler.api_resources import (
+    ListableApiResource,
+    FindableApiResource,
+    CreatableApiResource,
+    UpdatableApiResource,
+    DestroyableApiResource,
+)
+from nylas.models.auth import Provider
+from nylas.models.credentials import (
+    Credential,
+    CredentialRequest, ListCredentialQueryParams
+)
+from nylas.models.response import Response, ListResponse, DeleteResponse
+
+
+class Credentials(
+    ListableApiResource,
+    FindableApiResource,
+    CreatableApiResource,
+    UpdatableApiResource,
+    DestroyableApiResource,
+):
+    def list(self, provider: Provider, query_params: ListCredentialQueryParams) -> ListResponse[Credential]:
+        return super(Credentials, self).list(path=f"/v3/connectors/{provider}/creds", response_type=Credential,
+                                             query_params=query_params)
+
+    def find(self, provider: Provider, credential_id: str) -> Response[Credential]:
+        return super(Credentials, self).find(
+            path=f"/v3/connectors/{provider}/creds/{credential_id}", response_type=Credential
+        )
+
+    def create(self, request_body: CredentialRequest) -> Response[Credential]:
+        return super(Credentials, self).create(
+            path=f"/v3/connectors/microsoft/creds", response_type=Credential, request_body=request_body
+        )
+
+    def update(self, credential_id: str, request_body: CredentialRequest) -> Response[Credential]:
+        return super(Credentials, self).update(
+            path=f"/v3/connectors/{request_body.get('provider')}/creds/{credential_id}", response_type=Credential,
+            request_body=request_body
+        )
+
+    def destroy(self, provider: Provider, credential_id: str) -> DeleteResponse[Credential]:
+        return super(Credentials, self).destroy(path=f"/v3/connectors/{provider}/creds/{credential_id}")