diff --git a/tests/test_events.py b/tests/test_events.py index 8805517d..50d909df 100644 --- a/tests/test_events.py +++ b/tests/test_events.py @@ -1,6 +1,7 @@ import pytest from workos.events import Events from tests.utils.fixtures.mock_event import MockEvent +from tests.utils.fixtures.mock_directory_sync_event import MockDirectorySyncEvent class TestEvents(object): @@ -8,6 +9,27 @@ class TestEvents(object): def setup(self, set_api_key, set_client_id): self.events = Events() + @pytest.fixture + def mock_directory_sync_events(self): + events = [MockDirectorySyncEvent(id=str(i)).to_dict() for i in range(100)] + + return { + "data": events, + "list_metadata": {"after": None}, + "metadata": { + "params": { + "events": None, + "limit": None, + "organization_id": None, + "after": None, + "range_start": None, + "range_end": None, + "default_limit": True, + }, + "method": Events.list_directory_sync_events, + }, + } + @pytest.fixture def mock_events(self): events = [MockEvent(id=str(i)).to_dict() for i in range(100)] @@ -29,6 +51,15 @@ def mock_events(self): }, } + def test_list_directory_sync_events( + self, mock_directory_sync_events, mock_request_method + ): + mock_request_method("get", mock_directory_sync_events, 200) + + events = self.events.list_directory_sync_events() + + assert events == mock_directory_sync_events + def test_list_events(self, mock_events, mock_request_method): mock_request_method("get", mock_events, 200) diff --git a/tests/utils/fixtures/mock_directory_sync_event.py b/tests/utils/fixtures/mock_directory_sync_event.py new file mode 100644 index 00000000..93070829 --- /dev/null +++ b/tests/utils/fixtures/mock_directory_sync_event.py @@ -0,0 +1,39 @@ +import datetime +from workos.resources.base import WorkOSBaseResource +from workos.event_objects.directory_user import DirectoryUserEvent + + +class MockDirectorySyncEvent(WorkOSBaseResource): + def __init__(self, id): + self.object = "event" + self.id = id + self.event = "dsync.user.created" + self.data = { + "id": "directory_user_01E1X1B89NH8Z3SDFJR4H7RGX7", + "directory_id": "directory_01ECAZ4NV9QMV47GW873HDCX74", + "organization_id": "org_01EZTR6WYX1A0DSE2CYMGXQ24Y", + "idp_id": "8931", + "emails": [ + {"primary": True, "type": "work", "value": "lela.block@example.com"} + ], + "first_name": "Lela", + "last_name": "Block", + "job_title": "Software Engineer", + "username": "lela.block@example.com", + "state": "active", + "created_at": "2021-06-25T19:07:33.155Z", + "updated_at": "2021-06-25T19:07:33.155Z", + "custom_attributes": {"department": "Engineering"}, + "raw_attributes": {}, + "object": "directory_user", + } + + self.created_at = datetime.datetime.now() + + OBJECT_FIELDS = [ + "object", + "id", + "event", + "data", + "created_at", + ] diff --git a/tests/utils/fixtures/mock_event.py b/tests/utils/fixtures/mock_event.py index 0acdda4d..a7a756e3 100644 --- a/tests/utils/fixtures/mock_event.py +++ b/tests/utils/fixtures/mock_event.py @@ -6,8 +6,11 @@ class MockEvent(WorkOSBaseResource): def __init__(self, id): self.object = "event" self.id = id - self.event = "dsync.user.created" - self.data = {"id": "event_01234ABCD", "organization_id": "org_1234"} + self.event = "user.created" + self.data = { + "id": "user_01E4ZCR3C5A4QZ2Z2JQXGKZJ9E", + "email": "todd@example.com", + } self.created_at = datetime.datetime.now() OBJECT_FIELDS = [ diff --git a/workos/directory_sync.py b/workos/directory_sync.py index 2a1dd43a..32098b0a 100644 --- a/workos/directory_sync.py +++ b/workos/directory_sync.py @@ -1,4 +1,6 @@ from warnings import warn +from typing import Optional + import workos from workos.utils.pagination_order import Order from workos.utils.request import ( @@ -6,6 +8,7 @@ REQUEST_METHOD_DELETE, REQUEST_METHOD_GET, ) +from workos.utils.types import JsonDict from workos.utils.validation import DIRECTORY_SYNC_MODULE, validate_settings from workos.resources.directory_sync import ( @@ -107,13 +110,13 @@ def list_users( def list_users_v2( self, - directory=None, - group=None, - limit=None, - before=None, - after=None, - order=None, - ): + directory: Optional[str] = None, + group: Optional[str] = None, + limit: Optional[int] = None, + before: Optional[str] = None, + after: Optional[str] = None, + order: Optional[Order] = None, + ) -> WorkOSListResource: """Gets a list of provisioned Users for a Directory. Note, either 'directory' or 'group' must be provided. @@ -247,13 +250,13 @@ def list_groups( def list_groups_v2( self, - directory=None, - user=None, - limit=None, - before=None, - after=None, - order=None, - ): + directory: Optional[str] = None, + user: Optional[str] = None, + limit: Optional[int] = None, + before: Optional[str] = None, + after: Optional[str] = None, + order: Optional[Order] = None, + ) -> WorkOSListResource: """Gets a list of provisioned Groups for a Directory . Note, either 'directory' or 'user' must be provided. @@ -312,7 +315,7 @@ def list_groups_v2( return self.construct_from_response(response) - def get_user(self, user): + def get_user(self, user: str) -> WorkOSDirectoryUser: """Gets details for a single provisioned Directory User. Args: @@ -329,7 +332,7 @@ def get_user(self, user): return WorkOSDirectoryUser.construct_from_response(response).to_dict() - def get_group(self, group): + def get_group(self, group: str) -> WorkOSDirectoryGroup: """Gets details for a single provisioned Directory Group. Args: @@ -346,7 +349,7 @@ def get_group(self, group): return WorkOSDirectoryGroup.construct_from_response(response).to_dict() - def get_directory(self, directory): + def get_directory(self, directory: str) -> WorkOSDirectory: """Gets details for a single Directory Args: @@ -439,19 +442,19 @@ def list_directories( def list_directories_v2( self, - domain=None, - search=None, - limit=None, - before=None, - after=None, - organization=None, - order=None, + domain: Optional[str] = None, + search: Optional[str] = None, + limit: Optional[int] = None, + before: Optional[str] = None, + after: Optional[str] = None, + organization: Optional[str] = None, + order: Optional[Order] = None, ): """Gets details for existing Directories. Args: domain (str): Domain of a Directory. (Optional) - organization: ID of an Organization (Optional) + organization (str): ID of an Organization (Optional) search (str): Searchable text for a Directory. (Optional) limit (int): Maximum number of records to return. (Optional) before (str): Pagination cursor to receive records before a provided Directory ID. (Optional) @@ -505,26 +508,7 @@ def list_directories_v2( return self.construct_from_response(response) - def get_directory(self, directory): - """Gets details for a single Directory - - Args: - directory (str): Directory unique identifier. - - Returns: - dict: Directory response from WorkOS - - """ - - response = self.request_helper.request( - "directories/{directory}".format(directory=directory), - method=REQUEST_METHOD_GET, - token=workos.api_key, - ) - - return WorkOSDirectory.construct_from_response(response).to_dict() - - def delete_directory(self, directory): + def delete_directory(self, directory: str) -> JsonDict: """Delete one existing Directory. Args: diff --git a/workos/event_objects/__init__.py b/workos/event_objects/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/workos/event_objects/authentication.py b/workos/event_objects/authentication.py new file mode 100644 index 00000000..211b6629 --- /dev/null +++ b/workos/event_objects/authentication.py @@ -0,0 +1,188 @@ +from typing import List, Literal, NamedTuple +from enum import Enum +from workos.utils.types import JsonDict + + +class AuthenticationType(Enum): + EMAIL_VERIFICATION = "email_verification" + MAGIC_AUTH = "magic_auth" + MFA = "mfa" + OAUTH = "oauth" + PASSWORD = "password" + SSO = "sso" + + +class Error(NamedTuple): + code: str + message: str + + +class AuthenticationFailedFields: + def __init__(self, attributes: JsonDict) -> None: + self.type: AuthenticationType = AuthenticationType(attributes["type"]) + self.status: str = attributes["status"] + self.user_id: str = attributes["user_id"] + self.email: str = attributes["email"] + self.ip_address: str = attributes["ip_address"] + self.user_agent: str = attributes["user_agent"] + self.error: Error = Error( + code=attributes["error"]["code"], message=attributes["error"]["message"] + ) + + +class AuthenticationSucceededFields: + def __init__(self, attributes: JsonDict) -> None: + self.type: AuthenticationType = AuthenticationType(attributes["type"]) + self.status: str = attributes["status"] + self.user_id: str = attributes["user_id"] + self.email: str = attributes["email"] + self.ip_address: str = attributes["ip_address"] + self.user_agent: str = attributes["user_agent"] + + +class AuthenticationEmailVerificationFailedEvent: + event_name: str = "authentication.email_verification_failed" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.email_verification_failed"] = attributes[ + "event" + ] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationFailedFields = AuthenticationFailedFields( + attributes["data"] + ) + + +class AuthenticationEmailVerificationSucceededEvent: + event_name: str = "authentication.email_verification_succeeded" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.email_verification_succeeded"] = attributes[ + "event" + ] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationSucceededFields = AuthenticationSucceededFields( + attributes["data"] + ) + + +class AuthenticationMagicAuthFailedEvent: + event_name: str = "authentication.magic_auth_failed" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.magic_auth_failed"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationFailedFields = AuthenticationFailedFields( + attributes["data"] + ) + + +class AuthenticationMagicAuthSucceededEvent: + event_name: str = "authentication.magic_auth_succeeded" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.magic_auth_succeeded"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationSucceededFields = AuthenticationSucceededFields( + attributes["data"] + ) + + +class AuthenticationMFAFailedEvent: + event_name: str = "authentication.mfa_failed" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.mfa_failed"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationFailedFields = AuthenticationFailedFields( + attributes["data"] + ) + + +class AuthenticationMFASucceededEvent: + event_name: str = "authentication.mfa_succeeded" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.mfa_succeeded"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationSucceededFields = AuthenticationSucceededFields( + attributes["data"] + ) + + +class AuthenticationOAuthFailedEvent: + event_name: str = "authentication.oauth_failed" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.oauth_failed"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationFailedFields = AuthenticationFailedFields( + attributes["data"] + ) + + +class AuthenticationOAuthSucceededEvent: + event_name: str = "authentication.oauth_succeeded" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.oauth_succeeded"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationSucceededFields = AuthenticationSucceededFields( + attributes["data"] + ) + + +class AuthenticationPasswordFailedEvent: + event_name: str = "authentication.password_failed" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.password_failed"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationFailedFields = AuthenticationFailedFields( + attributes["data"] + ) + + +class AuthenticationPasswordSucceededEvent: + event_name: str = "authentication.password_succeeded" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.password_succeeded"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationSucceededFields = AuthenticationSucceededFields( + attributes["data"] + ) + + +class AuthenticationSSOFailedEvent: + event_name: str = "authentication.sso_failed" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.sso_failed"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationFailedFields = AuthenticationFailedFields( + attributes["data"] + ) + + +class AuthenticationSSOSucceededEvent: + event_name: str = "authentication.sso_succeeded" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["authentication.sso_succeeded"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: AuthenticationSucceededFields = AuthenticationSucceededFields( + attributes["data"] + ) diff --git a/workos/event_objects/connection.py b/workos/event_objects/connection.py new file mode 100644 index 00000000..b20b686f --- /dev/null +++ b/workos/event_objects/connection.py @@ -0,0 +1,95 @@ +from typing import List, Literal, NamedTuple +from enum import Enum +from workos.utils.types import JsonDict + + +class ConnectionState(Enum): + INACTIVE = "inactive" + ACTIVE = "active" + + +class ConnectionType(Enum): + ADFS_SAML = "ADFSSAML" + ADP_OIDC = "AdpOidc" + AUTH0_SAML = "Auth0SAML" + AZURE_SAML = "AzureSAML" + CAS_SAML = "CasSAML" + CLASS_LINK_SAML = "ClassLinkSAML" + CLOUDFLARE_SAML = "CloudflareSAML" + CYBER_ARK_SAML = "CyberArkSAML" + DUO_SAML = "DuoSAML" + GENERIC_OIDC = "GenericOIDC" + GENERIC_SAML = "GenericSAML" + GOOGLE_OAUTH = "GoogleOAuth" + GOOGLE_SAML = "GoogleSAML" + JUMP_CLOUD_SAML = "JumpCloudSAML" + KEYCLOAK_SAML = "KeycloakSAML" + LAST_PASS_SAML = "LastPassSAML" + LOGIN_GOV_OIDC = "LoginGovOidc" + MAGIC_LINK = "MagicLink" + MICROSOFT_OAUTH = "MicrosoftOAuth" + MINI_ORANGE_SAML = "MiniOrangeSAML" + NET_IQ_SAML = "NetIqSAML" + OKTA_SAML = "OktaSAML" + ONE_LOGIN_SAML = "OneLoginSAML" + ORACLE_SAML = "OracleSAML" + PING_FEDERATE_SAML = "PingFederateSAML" + PING_ONE_SAML = "PingOneSAML" + RIPPLING_SAML = "RipplingSAML" + SALESFORCE_SAML = "SalesforceSAML" + SHIBBOLETH_GENERIC_SAML = "ShibbolethGenericSAML" + SHIBBOLETH_SAML = "ShibbolethSAML" + SIMPLE_SAML_PHP_SAML = "SimpleSamlPhpSAML" + VM_WARE_SAML = "VMwareSAML" + + +class Domain: + def __init__(self, attributes: JsonDict) -> None: + self.id: str = attributes["id"] + self.object: Literal["connection_domain"] = attributes["object"] + self.domain: str = attributes["domain"] + + +class ConnectionEvent: + def __init__(self, attributes: JsonDict) -> None: + self.object: Literal["connection"] = attributes["object"] + self.id: str = attributes["id"] + self.organization_id: str = attributes["organization_id"] + self.state: str = ConnectionState(attributes["state"]) + self.connection_type: str = ConnectionType(attributes["connection_type"]) + self.name: str = attributes["name"] + self.created_at: str = attributes["created_at"] + self.updated_at: str = attributes["updated_at"] + self.domains = [] + for domain in attributes["domains"]: + self.domains.push(Domain(attributes=domain)) + + +class ConnectionActivatedEvent: + event_name = "connection.activated" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["connection.activated"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: ConnectionEvent = ConnectionEvent(attributes["data"]) + + +class ConnectionDeactivatedEvent: + event_name = "connection.deactivated" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["connection.deactivated"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: ConnectionEvent = ConnectionEvent(attributes["data"]) + + +class ConnectionDeactivatedEvent: + event_name = "connection.deleted" + + def __init__(self, attributes: JsonDict) -> None: + self.event: Literal["connection.deleted"] = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: ConnectionEvent = ConnectionEvent(attributes["data"]) diff --git a/workos/event_objects/directory.py b/workos/event_objects/directory.py new file mode 100644 index 00000000..3ed1993d --- /dev/null +++ b/workos/event_objects/directory.py @@ -0,0 +1,87 @@ +from typing import List, Literal +from enum import Enum +from workos.utils.types import JsonDict + + +class DirectoryType(Enum): + AZURE_SCIM_v2 = "azure scim v2.0" + BAMBOO_HR = "bamboohr" + BREATHE_HR = "breathe hr" + CEZANNE_HR = "cezanne hr" + CYBERARK_SCIM_v2 = "cyperark scim v2.0" + FOURTH_HR = "fourth hr" + GENERIC_SCIM_v2 = "generic scim v2.0" + GOOGLE = "gsuite directory" + HIBOB = "hibob" + JUMPCLOUD_SCIM_v2 = "jump cloud scim v2.0" + OKTA_SCIM_v2 = "okta scim v2.0" + ONELOGIN_SCIM_v2 = "onelogin scim v2.0" + PEOPLE_HR = "people hr" + PERSONIO = "personio" + PINGFEDERATE_SCIM_v2 = "pingfederate scim v2.0" + RIPPLING_SCIM_v2 = "rippling v2.0" + SFTP = "sftp" + SFTP_WORKDAY = "sftp workday" + WORKDAY = "workday" + + +class DirectoryState(Enum): + ACTIVE = "active" + DELETING = "deleting" + + +class OrganizationDomain: + def __init__(self, attributes: JsonDict) -> None: + self.id: str = attributes["id"] + self.domain: str = attributes["domain"] + + +class DirectoryEventWithLegacyFields: + def __init__(self, attributes) -> None: + self.id: str = attributes["id"] + self.name: str = attributes["name"] + self.type: DirectoryType = DirectoryType(attributes["type"]) + self.state: DirectoryState = DirectoryState.ACTIVE + self.domains: List[OrganizationDomain] = [] + for domain in attributes["domains"]: + self.domains.push(OrganizationDomain(domain)) + self.organization_id: str = attributes["organization_id"] + self.created_at: str = attributes["created_at"] + self.updated_at: str = attributes["updated_at"] + self.external_key: str = attributes["external_key"] + self.object: Literal["directory"] = attributes["object"] + + +class DirectoryEvent: + def __init__(self, attributes) -> None: + self.id: str = attributes["id"] + self.name: str = attributes["name"] + self.type: DirectoryType = DirectoryType(attributes["type"]) + self.state: DirectoryState = DirectoryState.ACTIVE + self.organization_id: str = attributes["organization_id"] + self.created_at: str = attributes["created_at"] + self.updated_at: str = attributes["updated_at"] + self.object: Literal["directory"] = attributes["object"] + + +class DirectoryActivatedEvent: + event_name = "dsync.activated" + + def __init__(self, attributes: JsonDict) -> None: + self.event: str = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.data: DirectoryEventWithLegacyFields = DirectoryEventWithLegacyFields( + attributes["data"] + ) + + +class DirectoryDeletedEvent: + event_name = "dsync.deleted" + + def __init__(self, attributes: JsonDict) -> None: + self.event: str = attributes["event"] + self.id: str = attributes["id"] + self.created_at = attributes["created_at"] + self.state: DirectoryState = DirectoryState.DELETING + self.data: DirectoryEvent = DirectoryEvent(attributes["data"]) diff --git a/workos/event_objects/directory_group.py b/workos/event_objects/directory_group.py new file mode 100644 index 00000000..488f3970 --- /dev/null +++ b/workos/event_objects/directory_group.py @@ -0,0 +1,46 @@ +from typing import Literal +from workos.utils.types import JsonDict + + +class DirectoryGroupEvent: + def __init__(self, attributes) -> None: + self.id: str = attributes["id"] + self.name: str = attributes["name"] + self.idp_id: str = attributes["idp_id"] + self.directory_id: str = attributes["directory_id"] + self.organization_id: str = attributes["organization_id"] + self.created_at: str = attributes["created_at"] + self.updated_at: str = attributes["updated_at"] + self.raw_attributes: JsonDict = attributes.get("raw_attributes", {}) + self.previous_attributes: JsonDict = attributes.get("previous_attributes", {}) + self.object: Literal["directory_group"] = attributes["object"] + + +class DirectoryGroupCreatedEvent: + event_name = "dsync.group.created" + + def __init__(self, attributes: JsonDict) -> None: + self.event: str = attributes["event"] + self.id: str = attributes["id"] + self.created_at: str = attributes["created_at"] + self.data: DirectoryGroupEvent = DirectoryGroupEvent(attributes["data"]) + + +class DirectoryGroupDeletedEvent: + event_name = "dsync.group.deleted" + + def __init__(self, attributes: JsonDict) -> None: + self.event: str = attributes["event"] + self.id: str = attributes["id"] + self.created_at: str = attributes["created_at"] + self.data: DirectoryGroupEvent = DirectoryGroupEvent(attributes["data"]) + + +class DirectoryGroupUpdatedEvent: + event_name = "dsync.group.updated" + + def __init__(self, attributes: JsonDict) -> None: + self.event: str = attributes["event"] + self.id: str = attributes["id"] + self.created_at: str = attributes["created_at"] + self.data: DirectoryGroupEvent = DirectoryGroupEvent(attributes["data"]) diff --git a/workos/event_objects/directory_user.py b/workos/event_objects/directory_user.py new file mode 100644 index 00000000..fe6c74f5 --- /dev/null +++ b/workos/event_objects/directory_user.py @@ -0,0 +1,77 @@ +from typing import Optional, List, Literal +from enum import Enum +from workos.utils.types import JsonDict + + +class DirectoryUserRole: + def __init__(self, attributes: JsonDict) -> None: + self.slug: str = attributes["slug"] + + +class DirectoryUserEmail: + def __init__(self, attributes: JsonDict) -> None: + self.type: Optional[str] = attributes.get("type") + self.value: Optional[str] = attributes.get("value") + self.primary: Optional[bool] = attributes.get("primary") + + +class DirectoryUserState(Enum): + ACTIVE = "active" + INACTIVE = "inactive" + + +class DirectoryUserEvent: + def __init__(self, attributes: JsonDict) -> None: + self.id: str = attributes["id"] + self.idp_id: str = attributes["idp_id"] + self.directory_id: str = attributes["directory_id"] + self.organization_id: str = attributes["organization_id"] + self.first_name: Optional[str] = attributes.get("first_name") + self.last_name: Optional[str] = attributes.get("last_name") + self.job_title: Optional[str] = attributes.get("job_title") + + self.emails: List[DirectoryUserEmail] = [] + for email in attributes.get("emails"): + self.emails.append(DirectoryUserEmail(email)) + + self.username: Optional[str] = attributes.get("username") + self.state: DirectoryUserState = DirectoryUserState(attributes["state"]) + self.custom_attributes: JsonDict = attributes.get("custom_attributes", {}) + self.raw_attributes: JsonDict = attributes.get("raw_attributes", {}) + self.created_at: str = attributes["created_at"] + self.updated_at: str = attributes["updated_at"] + self.role: Optional[DirectoryUserRole] = ( + DirectoryUserRole(attributes["role"]) if attributes.get("role") else None + ) + self.previous_attributes: JsonDict = attributes.get("previous_attributes", {}) + self.object: Literal["directory_user"] = attributes["object"] + + +class DirectoryUserCreatedEvent: + event_name = "dsync.user.created" + + def __init__(self, attributes: JsonDict) -> None: + self.id: str = attributes["id"] + self.event: str = attributes["event"] + self.created_at: str = attributes["created_at"] + self.data: DirectoryUserEvent = DirectoryUserEvent(attributes["data"]) + + +class DirectoryUserDeletedEvent: + event_name = "dsync.user.deleted" + + def __init__(self, attributes: JsonDict) -> None: + self.id: str = attributes["id"] + self.event: str = attributes["event"] + self.created_at: str = attributes["created_at"] + self.data: DirectoryUserEvent = DirectoryUserEvent(attributes["data"]) + + +class DirectoryUserUpdatedEvent: + event_name = "dsync.user.updated" + + def __init__(self, attributes: JsonDict) -> None: + self.id: str = attributes["id"] + self.event: str = attributes["event"] + self.created_at: str = attributes["created_at"] + self.data: DirectoryUserEvent = DirectoryUserEvent(attributes["data"]) diff --git a/workos/events.py b/workos/events.py index d4aef529..bbd70b54 100644 --- a/workos/events.py +++ b/workos/events.py @@ -1,4 +1,3 @@ -from warnings import warn import workos from workos.utils.request import ( RequestHelper, @@ -6,7 +5,51 @@ ) from workos.utils.validation import EVENTS_MODULE, validate_settings +from workos.utils.types import JsonDict from workos.resources.list import WorkOSListResource +from typing import Optional, TypedDict, List, Union +from workos.event_objects.directory import ( + DirectoryActivatedEvent, + DirectoryDeletedEvent, +) +from workos.event_objects.directory_group import ( + DirectoryGroupCreatedEvent, + DirectoryGroupDeletedEvent, + DirectoryGroupUpdatedEvent, +) +from workos.event_objects.directory_user import ( + DirectoryUserCreatedEvent, + DirectoryUserDeletedEvent, + DirectoryUserUpdatedEvent, +) + +EVENT_TO_OBJECT = { + DirectoryUserUpdatedEvent.event_name: DirectoryUserUpdatedEvent, + DirectoryUserCreatedEvent.event_name: DirectoryUserCreatedEvent, + DirectoryUserDeletedEvent.event_name: DirectoryUserDeletedEvent, + DirectoryGroupCreatedEvent.event_name: DirectoryGroupCreatedEvent, + DirectoryGroupDeletedEvent.event_name: DirectoryGroupDeletedEvent, + DirectoryGroupUpdatedEvent.event_name: DirectoryGroupUpdatedEvent, + DirectoryActivatedEvent.event_name: DirectoryActivatedEvent, + DirectoryActivatedEvent.event_name: DirectoryDeletedEvent, +} +EVENT_TYPES = Union[ + DirectoryActivatedEvent, + DirectoryDeletedEvent, + DirectoryUserCreatedEvent, + DirectoryUserDeletedEvent, + DirectoryUserUpdatedEvent, + DirectoryGroupCreatedEvent, + DirectoryGroupDeletedEvent, + DirectoryGroupUpdatedEvent, +] + + +class ListEventsResponse(TypedDict): + object: str + data: List[EVENT_TYPES] + metadata: JsonDict + RESPONSE_LIMIT = 10 @@ -26,13 +69,13 @@ def request_helper(self): def list_events( self, - events=None, - limit=None, - organization_id=None, - after=None, - range_start=None, - range_end=None, - ): + events: Optional[List[EVENT_TYPES]] = None, + limit: Optional[int] = None, + organization_id: Optional[str] = None, + after: Optional[str] = None, + range_start: Optional[str] = None, + range_end: Optional[str] = None, + ) -> ListEventsResponse: """Gets a list of Events . Kwargs: events (list): Filter to only return events of particular types. (Optional) @@ -78,4 +121,11 @@ def list_events( "method": Events.list_events, } + data = [] + for list_data in response["data"]: + if list_data["event"] in EVENT_TO_OBJECT.keys(): + data.append(EVENT_TO_OBJECT[list_data["event"]](list_data)) + + response["data"] = data + return response diff --git a/workos/utils/types.py b/workos/utils/types.py new file mode 100644 index 00000000..73ff6af8 --- /dev/null +++ b/workos/utils/types.py @@ -0,0 +1,3 @@ +from typing import Dict, Any + +JsonDict = Dict[str, Any]