diff --git a/Makefile b/Makefile index 72cabbcfb1..21ef24d3ca 100644 --- a/Makefile +++ b/Makefile @@ -15,7 +15,7 @@ test-docker: docker run twilio/twilio-python pytest tests --ignore=tests/cluster develop: venv - . venv/bin/activate; pip install -e . --use-mirrors + . venv/bin/activate; pip install -e . . venv/bin/activate; pip install -r tests/requirements.txt analysis: diff --git a/twilio/rest/monitor/v2/__init__.py b/twilio/rest/monitor/v2/__init__.py new file mode 100644 index 0000000000..d2d420c1df --- /dev/null +++ b/twilio/rest/monitor/v2/__init__.py @@ -0,0 +1,51 @@ +r""" + This code was generated by + ___ _ _ _ _ _ _ ____ ____ ____ _ ____ ____ _ _ ____ ____ ____ ___ __ __ + | | | | | | | | | __ | | |__| | __ | __ |___ |\ | |___ |__/ |__| | | | |__/ + | |_|_| | |___ | |__| |__| | | | |__] |___ | \| |___ | \ | | | |__| | \ + + Twilio - Alarms + Manages Twilio Alarms + + NOTE: This class is auto generated by OpenAPI Generator. + https://openapi-generator.tech + Do not edit the class manually. +""" + +from typing import Optional +from twilio.base.version import Version +from twilio.base.domain import Domain +from twilio.rest.monitor.v2.alarm import AlarmList +from twilio.rest.monitor.v2.status import StatusList + + +class V2(Version): + + def __init__(self, domain: Domain): + """ + Initialize the V2 version of Monitor + + :param domain: The Twilio.monitor domain + """ + super().__init__(domain, "v2") + self._alarms: Optional[AlarmList] = None + self._status: Optional[StatusList] = None + + @property + def alarms(self) -> AlarmList: + if self._alarms is None: + self._alarms = AlarmList(self) + return self._alarms + + @property + def status(self) -> StatusList: + if self._status is None: + self._status = StatusList(self) + return self._status + + def __repr__(self) -> str: + """ + Provide a friendly representation + :returns: Machine friendly representation + """ + return "" diff --git a/twilio/rest/monitor/v2/alarm.py b/twilio/rest/monitor/v2/alarm.py new file mode 100644 index 0000000000..3eace5a461 --- /dev/null +++ b/twilio/rest/monitor/v2/alarm.py @@ -0,0 +1,789 @@ +r""" + This code was generated by + ___ _ _ _ _ _ _ ____ ____ ____ _ ____ ____ _ _ ____ ____ ____ ___ __ __ + | | | | | | | | | __ | | |__| | __ | __ |___ |\ | |___ |__/ |__| | | | |__/ + | |_|_| | |___ | |__| |__| | | | |__] |___ | \| |___ | \ | | | |__| | \ + + Twilio - Alarms + Manages Twilio Alarms + + NOTE: This class is auto generated by OpenAPI Generator. + https://openapi-generator.tech + Do not edit the class manually. +""" + +from datetime import datetime +from typing import Any, Dict, List, Optional, Union, Iterator, AsyncIterator +from twilio.base import deserialize, values +from twilio.base.instance_context import InstanceContext +from twilio.base.instance_resource import InstanceResource +from twilio.base.list_resource import ListResource +from twilio.base.version import Version +from twilio.base.page import Page + + +class AlarmInstance(InstanceResource): + + class MonitorV2AlarmCreateObject(object): + """ + :ivar friendly_name: Friendly name for alarm + :ivar query_type: + :ivar query: Value to query for + :ivar trigger_value: Threshold to send customer alarm notification + :ivar time_window: + :ivar email: Email notifications to send + :ivar webhook: Webhook notification to send + :ivar console_indicator: Whether to send console notifications + :ivar description: Description for the alarm. + :ivar enabled: Is alarm enbled? Default is true. + """ + + def __init__(self, payload: Dict[str, Any]): + + self.friendly_name: Optional[str] = payload.get("friendly_name") + self.query_type: Optional["AlarmInstance.QueryType"] = payload.get( + "query_type" + ) + self.query: Optional[str] = payload.get("query") + self.trigger_value: Optional[int] = payload.get("trigger_value") + self.time_window: Optional["AlarmInstance.TimeWindow"] = payload.get( + "time_window" + ) + self.email: Optional[List[str]] = payload.get("email") + self.webhook: Optional[str] = payload.get("webhook") + self.console_indicator: Optional[bool] = payload.get("console_indicator") + self.description: Optional[str] = payload.get("description") + self.enabled: Optional[bool] = payload.get("enabled") + + def to_dict(self): + return { + "friendly_name": self.friendly_name, + "query_type": self.query_type, + "query": self.query, + "trigger_value": self.trigger_value, + "time_window": self.time_window, + "email": self.email, + "webhook": self.webhook, + "console_indicator": self.console_indicator, + "description": self.description, + "enabled": self.enabled, + } + + class QueryType(object): + ERROR_CODE = "ERROR_CODE" + LOG_LEVEL = "LOG_LEVEL" + ALL = "ALL" + + class TimeWindow(object): + FIVE_MINS = "FIVE_MINS" + FIFTEEN_MINS = "FIFTEEN_MINS" + ONE_HOUR = "ONE_HOUR" + TWELVE_HOURS = "TWELVE_HOURS" + ONE_DAY = "ONE_DAY" + + """ + :ivar sid: + :ivar message: Error message + :ivar code: Twilio error code + :ivar user_error: Indicates if it was a user error + :ivar http_status_code: Http error code returned + :ivar params: + :ivar more_info: More information link + :ivar status: HTTP status code + :ivar friendly_name: Friendly name for alarm + :ivar query_type: + :ivar query: Value to query for + :ivar trigger_value: Threshold to send customer alarm notification + :ivar time_window: + :ivar email: Email notifications to send + :ivar webhook: Webhook notification to send + :ivar console_indicator: Whether to send console notifications + :ivar description: Description for the alarm. + :ivar enabled: Is alarm enbled? Default is true. + :ivar product: Product the alarm is related to. + :ivar alert_config_sid: Sid for the alarm + :ivar date_updated: The date and time in UTC when the alarm was last updated specified in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) format + :ivar date_created: The date and time in UTC when the alarm was created specified in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) format. + """ + + def __init__( + self, version: Version, payload: Dict[str, Any], sid: Optional[str] = None + ): + super().__init__(version) + + self.sid: Optional[str] = payload.get("sid") + self.message: Optional[str] = payload.get("message") + self.code: Optional[int] = deserialize.integer(payload.get("code")) + self.user_error: Optional[bool] = payload.get("user_error") + self.http_status_code: Optional[int] = deserialize.integer( + payload.get("http_status_code") + ) + self.params: Optional[dict[str, str]] = payload.get("params") + self.more_info: Optional[str] = payload.get("more_info") + self.status: Optional[int] = deserialize.integer(payload.get("status")) + self.friendly_name: Optional[str] = payload.get("friendlyName") + self.query_type: Optional[str] = payload.get("queryType") + self.query: Optional[str] = payload.get("query") + self.trigger_value: Optional[int] = deserialize.integer( + payload.get("triggerValue") + ) + self.time_window: Optional[str] = payload.get("timeWindow") + self.email: Optional[List[str]] = payload.get("email") + self.webhook: Optional[str] = payload.get("webhook") + self.console_indicator: Optional[bool] = payload.get("consoleIndicator") + self.description: Optional[str] = payload.get("description") + self.enabled: Optional[bool] = payload.get("enabled") + self.product: Optional[str] = payload.get("product") + self.alert_config_sid: Optional[str] = payload.get("alertConfigSid") + self.date_updated: Optional[datetime] = deserialize.iso8601_datetime( + payload.get("dateUpdated") + ) + self.date_created: Optional[datetime] = deserialize.iso8601_datetime( + payload.get("dateCreated") + ) + + self._solution = { + "sid": sid or self.sid, + } + self._context: Optional[AlarmContext] = None + + @property + def _proxy(self) -> "AlarmContext": + """ + Generate an instance context for the instance, the context is capable of + performing various actions. All instance actions are proxied to the context + + :returns: AlarmContext for this AlarmInstance + """ + if self._context is None: + self._context = AlarmContext( + self._version, + sid=self._solution["sid"], + ) + return self._context + + def delete(self) -> bool: + """ + Deletes the AlarmInstance + + + :returns: True if delete succeeds, False otherwise + """ + return self._proxy.delete() + + async def delete_async(self) -> bool: + """ + Asynchronous coroutine that deletes the AlarmInstance + + + :returns: True if delete succeeds, False otherwise + """ + return await self._proxy.delete_async() + + def fetch(self) -> "AlarmInstance": + """ + Fetch the AlarmInstance + + + :returns: The fetched AlarmInstance + """ + return self._proxy.fetch() + + async def fetch_async(self) -> "AlarmInstance": + """ + Asynchronous coroutine to fetch the AlarmInstance + + + :returns: The fetched AlarmInstance + """ + return await self._proxy.fetch_async() + + def update( + self, monitor_v2_alarm_create_object: MonitorV2AlarmCreateObject + ) -> "AlarmInstance": + """ + Update the AlarmInstance + + :param monitor_v2_alarm_create_object: + + :returns: The updated AlarmInstance + """ + return self._proxy.update( + monitor_v2_alarm_create_object=monitor_v2_alarm_create_object, + ) + + async def update_async( + self, monitor_v2_alarm_create_object: MonitorV2AlarmCreateObject + ) -> "AlarmInstance": + """ + Asynchronous coroutine to update the AlarmInstance + + :param monitor_v2_alarm_create_object: + + :returns: The updated AlarmInstance + """ + return await self._proxy.update_async( + monitor_v2_alarm_create_object=monitor_v2_alarm_create_object, + ) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + context = " ".join("{}={}".format(k, v) for k, v in self._solution.items()) + return "".format(context) + + +class AlarmContext(InstanceContext): + + class MonitorV2AlarmCreateObject(object): + """ + :ivar friendly_name: Friendly name for alarm + :ivar query_type: + :ivar query: Value to query for + :ivar trigger_value: Threshold to send customer alarm notification + :ivar time_window: + :ivar email: Email notifications to send + :ivar webhook: Webhook notification to send + :ivar console_indicator: Whether to send console notifications + :ivar description: Description for the alarm. + :ivar enabled: Is alarm enbled? Default is true. + """ + + def __init__(self, payload: Dict[str, Any]): + + self.friendly_name: Optional[str] = payload.get("friendly_name") + self.query_type: Optional["AlarmInstance.QueryType"] = payload.get( + "query_type" + ) + self.query: Optional[str] = payload.get("query") + self.trigger_value: Optional[int] = payload.get("trigger_value") + self.time_window: Optional["AlarmInstance.TimeWindow"] = payload.get( + "time_window" + ) + self.email: Optional[List[str]] = payload.get("email") + self.webhook: Optional[str] = payload.get("webhook") + self.console_indicator: Optional[bool] = payload.get("console_indicator") + self.description: Optional[str] = payload.get("description") + self.enabled: Optional[bool] = payload.get("enabled") + + def to_dict(self): + return { + "friendly_name": self.friendly_name, + "query_type": self.query_type, + "query": self.query, + "trigger_value": self.trigger_value, + "time_window": self.time_window, + "email": self.email, + "webhook": self.webhook, + "console_indicator": self.console_indicator, + "description": self.description, + "enabled": self.enabled, + } + + def __init__(self, version: Version, sid: str): + """ + Initialize the AlarmContext + + :param version: Version that contains the resource + :param sid: Sid + """ + super().__init__(version) + + # Path Solution + self._solution = { + "sid": sid, + } + self._uri = "/Alarms/{sid}".format(**self._solution) + + def delete(self) -> bool: + """ + Deletes the AlarmInstance + + + :returns: True if delete succeeds, False otherwise + """ + + headers = values.of({}) + + headers["Accept"] = "application/json" + + return self._version.delete(method="DELETE", uri=self._uri, headers=headers) + + async def delete_async(self) -> bool: + """ + Asynchronous coroutine that deletes the AlarmInstance + + + :returns: True if delete succeeds, False otherwise + """ + + headers = values.of({}) + + headers["Accept"] = "application/json" + + return await self._version.delete_async( + method="DELETE", uri=self._uri, headers=headers + ) + + def fetch(self) -> AlarmInstance: + """ + Fetch the AlarmInstance + + + :returns: The fetched AlarmInstance + """ + + headers = values.of({}) + + headers["Accept"] = "application/json" + + payload = self._version.fetch(method="GET", uri=self._uri, headers=headers) + + return AlarmInstance( + self._version, + payload, + sid=self._solution["sid"], + ) + + async def fetch_async(self) -> AlarmInstance: + """ + Asynchronous coroutine to fetch the AlarmInstance + + + :returns: The fetched AlarmInstance + """ + + headers = values.of({}) + + headers["Accept"] = "application/json" + + payload = await self._version.fetch_async( + method="GET", uri=self._uri, headers=headers + ) + + return AlarmInstance( + self._version, + payload, + sid=self._solution["sid"], + ) + + def update( + self, monitor_v2_alarm_create_object: MonitorV2AlarmCreateObject + ) -> AlarmInstance: + """ + Update the AlarmInstance + + :param monitor_v2_alarm_create_object: + + :returns: The updated AlarmInstance + """ + data = monitor_v2_alarm_create_object.to_dict() + + headers = values.of({}) + + headers["Content-Type"] = "application/json" + + headers["Accept"] = "application/json" + + payload = self._version.update( + method="PUT", uri=self._uri, data=data, headers=headers + ) + + return AlarmInstance(self._version, payload, sid=self._solution["sid"]) + + async def update_async( + self, monitor_v2_alarm_create_object: MonitorV2AlarmCreateObject + ) -> AlarmInstance: + """ + Asynchronous coroutine to update the AlarmInstance + + :param monitor_v2_alarm_create_object: + + :returns: The updated AlarmInstance + """ + data = monitor_v2_alarm_create_object.to_dict() + + headers = values.of({}) + + headers["Content-Type"] = "application/json" + + headers["Accept"] = "application/json" + + payload = await self._version.update_async( + method="PUT", uri=self._uri, data=data, headers=headers + ) + + return AlarmInstance(self._version, payload, sid=self._solution["sid"]) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + context = " ".join("{}={}".format(k, v) for k, v in self._solution.items()) + return "".format(context) + + +class AlarmPage(Page): + + def get_instance(self, payload: Dict[str, Any]) -> AlarmInstance: + """ + Build an instance of AlarmInstance + + :param payload: Payload response from the API + """ + return AlarmInstance(self._version, payload) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + return "" + + +class AlarmList(ListResource): + + class MonitorV2AlarmCreateObject(object): + """ + :ivar friendly_name: Friendly name for alarm + :ivar query_type: + :ivar query: Value to query for + :ivar trigger_value: Threshold to send customer alarm notification + :ivar time_window: + :ivar email: Email notifications to send + :ivar webhook: Webhook notification to send + :ivar console_indicator: Whether to send console notifications + :ivar description: Description for the alarm. + :ivar enabled: Is alarm enbled? Default is true. + """ + + def __init__(self, payload: Dict[str, Any]): + + self.friendly_name: Optional[str] = payload.get("friendly_name") + self.query_type: Optional["AlarmInstance.QueryType"] = payload.get( + "query_type" + ) + self.query: Optional[str] = payload.get("query") + self.trigger_value: Optional[int] = payload.get("trigger_value") + self.time_window: Optional["AlarmInstance.TimeWindow"] = payload.get( + "time_window" + ) + self.email: Optional[List[str]] = payload.get("email") + self.webhook: Optional[str] = payload.get("webhook") + self.console_indicator: Optional[bool] = payload.get("console_indicator") + self.description: Optional[str] = payload.get("description") + self.enabled: Optional[bool] = payload.get("enabled") + + def to_dict(self): + return { + "friendly_name": self.friendly_name, + "query_type": self.query_type, + "query": self.query, + "trigger_value": self.trigger_value, + "time_window": self.time_window, + "email": self.email, + "webhook": self.webhook, + "console_indicator": self.console_indicator, + "description": self.description, + "enabled": self.enabled, + } + + def __init__(self, version: Version): + """ + Initialize the AlarmList + + :param version: Version that contains the resource + + """ + super().__init__(version) + + self._uri = "/Alarms" + + def create( + self, monitor_v2_alarm_create_object: MonitorV2AlarmCreateObject + ) -> AlarmInstance: + """ + Create the AlarmInstance + + :param monitor_v2_alarm_create_object: + + :returns: The created AlarmInstance + """ + data = monitor_v2_alarm_create_object.to_dict() + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Content-Type"] = "application/json" + + headers["Accept"] = "application/json" + + payload = self._version.create( + method="POST", uri=self._uri, data=data, headers=headers + ) + + return AlarmInstance(self._version, payload) + + async def create_async( + self, monitor_v2_alarm_create_object: MonitorV2AlarmCreateObject + ) -> AlarmInstance: + """ + Asynchronously create the AlarmInstance + + :param monitor_v2_alarm_create_object: + + :returns: The created AlarmInstance + """ + data = monitor_v2_alarm_create_object.to_dict() + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Content-Type"] = "application/json" + + headers["Accept"] = "application/json" + + payload = await self._version.create_async( + method="POST", uri=self._uri, data=data, headers=headers + ) + + return AlarmInstance(self._version, payload) + + def stream( + self, + sort_order: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> Iterator[AlarmInstance]: + """ + Streams AlarmInstance records from the API as a generator stream. + This operation lazily loads records as efficiently as possible until the limit + is reached. + The results are returned as a generator, so this operation is memory efficient. + + :param str sort_order: + :param limit: Upper limit for the number of records to return. stream() + guarantees to never return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, stream() will attempt to read the + limit with the most efficient page size, i.e. min(limit, 1000) + + :returns: Generator that will yield up to limit results + """ + limits = self._version.read_limits(limit, page_size) + page = self.page(sort_order=sort_order, page_size=limits["page_size"]) + + return self._version.stream(page, limits["limit"]) + + async def stream_async( + self, + sort_order: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> AsyncIterator[AlarmInstance]: + """ + Asynchronously streams AlarmInstance records from the API as a generator stream. + This operation lazily loads records as efficiently as possible until the limit + is reached. + The results are returned as a generator, so this operation is memory efficient. + + :param str sort_order: + :param limit: Upper limit for the number of records to return. stream() + guarantees to never return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, stream() will attempt to read the + limit with the most efficient page size, i.e. min(limit, 1000) + + :returns: Generator that will yield up to limit results + """ + limits = self._version.read_limits(limit, page_size) + page = await self.page_async( + sort_order=sort_order, page_size=limits["page_size"] + ) + + return self._version.stream_async(page, limits["limit"]) + + def list( + self, + sort_order: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> List[AlarmInstance]: + """ + Lists AlarmInstance records from the API as a list. + Unlike stream(), this operation is eager and will load `limit` records into + memory before returning. + + :param str sort_order: + :param limit: Upper limit for the number of records to return. list() guarantees + never to return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, list() will attempt to read the limit + with the most efficient page size, i.e. min(limit, 1000) + + :returns: list that will contain up to limit results + """ + return list( + self.stream( + sort_order=sort_order, + limit=limit, + page_size=page_size, + ) + ) + + async def list_async( + self, + sort_order: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> List[AlarmInstance]: + """ + Asynchronously lists AlarmInstance records from the API as a list. + Unlike stream(), this operation is eager and will load `limit` records into + memory before returning. + + :param str sort_order: + :param limit: Upper limit for the number of records to return. list() guarantees + never to return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, list() will attempt to read the limit + with the most efficient page size, i.e. min(limit, 1000) + + :returns: list that will contain up to limit results + """ + return [ + record + async for record in await self.stream_async( + sort_order=sort_order, + limit=limit, + page_size=page_size, + ) + ] + + def page( + self, + sort_order: Union[str, object] = values.unset, + page_token: Union[str, object] = values.unset, + page_number: Union[int, object] = values.unset, + page_size: Union[int, object] = values.unset, + ) -> AlarmPage: + """ + Retrieve a single page of AlarmInstance records from the API. + Request is executed immediately + + :param sort_order: + :param page_token: PageToken provided by the API + :param page_number: Page Number, this value is simply for client state + :param page_size: Number of records to return, defaults to 50 + + :returns: Page of AlarmInstance + """ + data = values.of( + { + "SortOrder": sort_order, + "PageToken": page_token, + "Page": page_number, + "PageSize": page_size, + } + ) + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Accept"] = "application/json" + + response = self._version.page( + method="GET", uri=self._uri, params=data, headers=headers + ) + return AlarmPage(self._version, response) + + async def page_async( + self, + sort_order: Union[str, object] = values.unset, + page_token: Union[str, object] = values.unset, + page_number: Union[int, object] = values.unset, + page_size: Union[int, object] = values.unset, + ) -> AlarmPage: + """ + Asynchronously retrieve a single page of AlarmInstance records from the API. + Request is executed immediately + + :param sort_order: + :param page_token: PageToken provided by the API + :param page_number: Page Number, this value is simply for client state + :param page_size: Number of records to return, defaults to 50 + + :returns: Page of AlarmInstance + """ + data = values.of( + { + "SortOrder": sort_order, + "PageToken": page_token, + "Page": page_number, + "PageSize": page_size, + } + ) + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Accept"] = "application/json" + + response = await self._version.page_async( + method="GET", uri=self._uri, params=data, headers=headers + ) + return AlarmPage(self._version, response) + + def get_page(self, target_url: str) -> AlarmPage: + """ + Retrieve a specific page of AlarmInstance records from the API. + Request is executed immediately + + :param target_url: API-generated URL for the requested results page + + :returns: Page of AlarmInstance + """ + response = self._version.domain.twilio.request("GET", target_url) + return AlarmPage(self._version, response) + + async def get_page_async(self, target_url: str) -> AlarmPage: + """ + Asynchronously retrieve a specific page of AlarmInstance records from the API. + Request is executed immediately + + :param target_url: API-generated URL for the requested results page + + :returns: Page of AlarmInstance + """ + response = await self._version.domain.twilio.request_async("GET", target_url) + return AlarmPage(self._version, response) + + def get(self, sid: str) -> AlarmContext: + """ + Constructs a AlarmContext + + :param sid: Sid + """ + return AlarmContext(self._version, sid=sid) + + def __call__(self, sid: str) -> AlarmContext: + """ + Constructs a AlarmContext + + :param sid: Sid + """ + return AlarmContext(self._version, sid=sid) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + return "" diff --git a/twilio/rest/monitor/v2/status.py b/twilio/rest/monitor/v2/status.py new file mode 100644 index 0000000000..a6b75e0d0e --- /dev/null +++ b/twilio/rest/monitor/v2/status.py @@ -0,0 +1,162 @@ +r""" + This code was generated by + ___ _ _ _ _ _ _ ____ ____ ____ _ ____ ____ _ _ ____ ____ ____ ___ __ __ + | | | | | | | | | __ | | |__| | __ | __ |___ |\ | |___ |__/ |__| | | | |__/ + | |_|_| | |___ | |__| |__| | | | |__] |___ | \| |___ | \ | | | |__| | \ + + Twilio - Alarms + Manages Twilio Alarms + + NOTE: This class is auto generated by OpenAPI Generator. + https://openapi-generator.tech + Do not edit the class manually. +""" + +from typing import Any, Dict, Optional +from twilio.base import deserialize, values + +from twilio.base.instance_resource import InstanceResource +from twilio.base.list_resource import ListResource +from twilio.base.version import Version + + +class StatusInstance(InstanceResource): + + class MonitorV2AlarmStatus(object): + """ + :ivar enabled: status of alarm + """ + + def __init__(self, payload: Dict[str, Any]): + + self.enabled: Optional[bool] = payload.get("enabled") + + def to_dict(self): + return { + "enabled": self.enabled, + } + + """ + :ivar message: Error message + :ivar code: Twilio error code + :ivar user_error: Indicates if it was a user error + :ivar http_status_code: Http error code returned + :ivar params: + :ivar more_info: More information link + :ivar status: HTTP status code + """ + + def __init__( + self, version: Version, payload: Dict[str, Any], sid: Optional[str] = None + ): + super().__init__(version) + + self.message: Optional[str] = payload.get("message") + self.code: Optional[int] = deserialize.integer(payload.get("code")) + self.user_error: Optional[bool] = payload.get("user_error") + self.http_status_code: Optional[int] = deserialize.integer( + payload.get("http_status_code") + ) + self.params: Optional[dict[str, str]] = payload.get("params") + self.more_info: Optional[str] = payload.get("more_info") + self.status: Optional[int] = deserialize.integer(payload.get("status")) + + self._solution = { + "sid": sid or self.sid, + } + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + context = " ".join("{}={}".format(k, v) for k, v in self._solution.items()) + return "".format(context) + + +class StatusList(ListResource): + + class MonitorV2AlarmStatus(object): + """ + :ivar enabled: status of alarm + """ + + def __init__(self, payload: Dict[str, Any]): + + self.enabled: Optional[bool] = payload.get("enabled") + + def to_dict(self): + return { + "enabled": self.enabled, + } + + def __init__(self, version: Version, sid: str): + """ + Initialize the StatusList + + :param version: Version that contains the resource + :param sid: Sid + + """ + super().__init__(version) + + # Path Solution + self._solution = { + "sid": sid, + } + self._uri = "/Alarms/{sid}/status".format(**self._solution) + + def update(self, monitor_v2_alarm_status: MonitorV2AlarmStatus) -> StatusInstance: + """ + Update the StatusInstance + + :param monitor_v2_alarm_status: + + :returns: The created StatusInstance + """ + data = monitor_v2_alarm_status.to_dict() + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Content-Type"] = "application/json" + + headers["Accept"] = "application/json" + + payload = self._version.update( + method="PUT", uri=self._uri, data=data, headers=headers + ) + + return StatusInstance(self._version, payload, sid=self._solution["sid"]) + + async def update_async( + self, monitor_v2_alarm_status: MonitorV2AlarmStatus + ) -> StatusInstance: + """ + Asynchronously update the StatusInstance + + :param monitor_v2_alarm_status: + + :returns: The created StatusInstance + """ + data = monitor_v2_alarm_status.to_dict() + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Content-Type"] = "application/json" + + headers["Accept"] = "application/json" + + payload = await self._version.update_async( + method="PUT", uri=self._uri, data=data, headers=headers + ) + + return StatusInstance(self._version, payload, sid=self._solution["sid"]) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + return "" diff --git a/twilio/rest/preview_iam/__init__.py b/twilio/rest/preview_iam/__init__.py index 9a5bf85b24..dd51b97489 100644 --- a/twilio/rest/preview_iam/__init__.py +++ b/twilio/rest/preview_iam/__init__.py @@ -1,26 +1,59 @@ -from twilio.rest.preview_iam.PreviewIamBase import PreviewIamBase +r""" + This code was generated by + ___ _ _ _ _ _ _ ____ ____ ____ _ ____ ____ _ _ ____ ____ ____ ___ __ __ + | | | | | | | | | __ | | |__| | __ | __ |___ |\ | |___ |__/ |__| | | | |__/ + | |_|_| | |___ | |__| |__| | | | |__] |___ | \| |___ | \ | | | |__| | \ -from twilio.rest.preview_iam.v1.authorize import ( - AuthorizeList, -) -from twilio.rest.preview_iam.v1.token import ( - TokenList, -) -from twilio.rest.preview_iam.versionless.organization import ( - OrganizationList, -) -from twilio.rest.preview_iam.versionless import Versionless + Organization Public API + No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator) + NOTE: This class is auto generated by OpenAPI Generator. + https://openapi-generator.tech + Do not edit the class manually. +""" + +from typing import Optional +from twilio.base.version import Version +from twilio.base.domain import Domain +from twilio.rest.preview_iam.organizations.account import AccountList +from twilio.rest.preview_iam.organizations.role_assignment import RoleAssignmentList +from twilio.rest.preview_iam.organizations.user import UserList + + +class Organizations(Version): + + def __init__(self, domain: Domain): + """ + Initialize the Organizations version of PreviewIam + + :param domain: The Twilio.preview_iam domain + """ + super().__init__(domain, "Organizations") + self._accounts: Optional[AccountList] = None + self._role_assignments: Optional[RoleAssignmentList] = None + self._users: Optional[UserList] = None -class PreviewIam(PreviewIamBase): @property - def organization(self) -> OrganizationList: - return Versionless(self).organization + def accounts(self) -> AccountList: + if self._accounts is None: + self._accounts = AccountList(self) + return self._accounts @property - def authorize(self) -> AuthorizeList: - return self.v1.authorize + def role_assignments(self) -> RoleAssignmentList: + if self._role_assignments is None: + self._role_assignments = RoleAssignmentList(self) + return self._role_assignments @property - def token(self) -> TokenList: - return self.v1.token + def users(self) -> UserList: + if self._users is None: + self._users = UserList(self) + return self._users + + def __repr__(self) -> str: + """ + Provide a friendly representation + :returns: Machine friendly representation + """ + return "" diff --git a/twilio/rest/preview_iam/organizations/__init__.py b/twilio/rest/preview_iam/organizations/__init__.py new file mode 100644 index 0000000000..dd51b97489 --- /dev/null +++ b/twilio/rest/preview_iam/organizations/__init__.py @@ -0,0 +1,59 @@ +r""" + This code was generated by + ___ _ _ _ _ _ _ ____ ____ ____ _ ____ ____ _ _ ____ ____ ____ ___ __ __ + | | | | | | | | | __ | | |__| | __ | __ |___ |\ | |___ |__/ |__| | | | |__/ + | |_|_| | |___ | |__| |__| | | | |__] |___ | \| |___ | \ | | | |__| | \ + + Organization Public API + No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator) + + NOTE: This class is auto generated by OpenAPI Generator. + https://openapi-generator.tech + Do not edit the class manually. +""" + +from typing import Optional +from twilio.base.version import Version +from twilio.base.domain import Domain +from twilio.rest.preview_iam.organizations.account import AccountList +from twilio.rest.preview_iam.organizations.role_assignment import RoleAssignmentList +from twilio.rest.preview_iam.organizations.user import UserList + + +class Organizations(Version): + + def __init__(self, domain: Domain): + """ + Initialize the Organizations version of PreviewIam + + :param domain: The Twilio.preview_iam domain + """ + super().__init__(domain, "Organizations") + self._accounts: Optional[AccountList] = None + self._role_assignments: Optional[RoleAssignmentList] = None + self._users: Optional[UserList] = None + + @property + def accounts(self) -> AccountList: + if self._accounts is None: + self._accounts = AccountList(self) + return self._accounts + + @property + def role_assignments(self) -> RoleAssignmentList: + if self._role_assignments is None: + self._role_assignments = RoleAssignmentList(self) + return self._role_assignments + + @property + def users(self) -> UserList: + if self._users is None: + self._users = UserList(self) + return self._users + + def __repr__(self) -> str: + """ + Provide a friendly representation + :returns: Machine friendly representation + """ + return "" diff --git a/twilio/rest/preview_iam/organizations/account.py b/twilio/rest/preview_iam/organizations/account.py new file mode 100644 index 0000000000..bc5f49e28f --- /dev/null +++ b/twilio/rest/preview_iam/organizations/account.py @@ -0,0 +1,436 @@ +r""" + This code was generated by + ___ _ _ _ _ _ _ ____ ____ ____ _ ____ ____ _ _ ____ ____ ____ ___ __ __ + | | | | | | | | | __ | | |__| | __ | __ |___ |\ | |___ |__/ |__| | | | |__/ + | |_|_| | |___ | |__| |__| | | | |__] |___ | \| |___ | \ | | | |__| | \ + + Organization Public API + No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator) + + NOTE: This class is auto generated by OpenAPI Generator. + https://openapi-generator.tech + Do not edit the class manually. +""" + +from datetime import datetime +from typing import Any, Dict, List, Optional, Union, Iterator, AsyncIterator +from twilio.base import deserialize, values +from twilio.base.instance_context import InstanceContext +from twilio.base.instance_resource import InstanceResource +from twilio.base.list_resource import ListResource +from twilio.base.version import Version +from twilio.base.page import Page + + +class AccountInstance(InstanceResource): + """ + :ivar account_sid: Twilio account sid + :ivar friendly_name: Account friendly name + :ivar status: Account status + :ivar owner_sid: Twilio account sid + :ivar date_created: The date and time when the account was created in the system + """ + + def __init__( + self, + version: Version, + payload: Dict[str, Any], + organization_sid: Optional[str] = None, + account_sid: Optional[str] = None, + ): + super().__init__(version) + + self.account_sid: Optional[str] = payload.get("account_sid") + self.friendly_name: Optional[str] = payload.get("friendly_name") + self.status: Optional[str] = payload.get("status") + self.owner_sid: Optional[str] = payload.get("owner_sid") + self.date_created: Optional[datetime] = deserialize.iso8601_datetime( + payload.get("date_created") + ) + + self._solution = { + "organization_sid": organization_sid or self.organization_sid, + "account_sid": account_sid or self.account_sid, + } + self._context: Optional[AccountContext] = None + + @property + def _proxy(self) -> "AccountContext": + """ + Generate an instance context for the instance, the context is capable of + performing various actions. All instance actions are proxied to the context + + :returns: AccountContext for this AccountInstance + """ + if self._context is None: + self._context = AccountContext( + self._version, + organization_sid=self._solution["organization_sid"], + account_sid=self._solution["account_sid"], + ) + return self._context + + def fetch(self) -> "AccountInstance": + """ + Fetch the AccountInstance + + + :returns: The fetched AccountInstance + """ + return self._proxy.fetch() + + async def fetch_async(self) -> "AccountInstance": + """ + Asynchronous coroutine to fetch the AccountInstance + + + :returns: The fetched AccountInstance + """ + return await self._proxy.fetch_async() + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + context = " ".join("{}={}".format(k, v) for k, v in self._solution.items()) + return "".format(context) + + +class AccountContext(InstanceContext): + + def __init__(self, version: Version, organization_sid: str, account_sid: str): + """ + Initialize the AccountContext + + :param version: Version that contains the resource + :param organization_sid: + :param account_sid: + """ + super().__init__(version) + + # Path Solution + self._solution = { + "organization_sid": organization_sid, + "account_sid": account_sid, + } + self._uri = "/{organization_sid}/Accounts/{account_sid}".format( + **self._solution + ) + + def fetch(self) -> AccountInstance: + """ + Fetch the AccountInstance + + + :returns: The fetched AccountInstance + """ + + headers = values.of({}) + + headers["Accept"] = "application/json" + + payload = self._version.fetch(method="GET", uri=self._uri, headers=headers) + + return AccountInstance( + self._version, + payload, + organization_sid=self._solution["organization_sid"], + account_sid=self._solution["account_sid"], + ) + + async def fetch_async(self) -> AccountInstance: + """ + Asynchronous coroutine to fetch the AccountInstance + + + :returns: The fetched AccountInstance + """ + + headers = values.of({}) + + headers["Accept"] = "application/json" + + payload = await self._version.fetch_async( + method="GET", uri=self._uri, headers=headers + ) + + return AccountInstance( + self._version, + payload, + organization_sid=self._solution["organization_sid"], + account_sid=self._solution["account_sid"], + ) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + context = " ".join("{}={}".format(k, v) for k, v in self._solution.items()) + return "".format(context) + + +class AccountPage(Page): + + def get_instance(self, payload: Dict[str, Any]) -> AccountInstance: + """ + Build an instance of AccountInstance + + :param payload: Payload response from the API + """ + return AccountInstance( + self._version, payload, organization_sid=self._solution["organization_sid"] + ) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + return "" + + +class AccountList(ListResource): + + def __init__(self, version: Version, organization_sid: str): + """ + Initialize the AccountList + + :param version: Version that contains the resource + :param organization_sid: + + """ + super().__init__(version) + + # Path Solution + self._solution = { + "organization_sid": organization_sid, + } + self._uri = "/{organization_sid}/Accounts".format(**self._solution) + + def stream( + self, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> Iterator[AccountInstance]: + """ + Streams AccountInstance records from the API as a generator stream. + This operation lazily loads records as efficiently as possible until the limit + is reached. + The results are returned as a generator, so this operation is memory efficient. + + :param limit: Upper limit for the number of records to return. stream() + guarantees to never return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, stream() will attempt to read the + limit with the most efficient page size, i.e. min(limit, 1000) + + :returns: Generator that will yield up to limit results + """ + limits = self._version.read_limits(limit, page_size) + page = self.page(page_size=limits["page_size"]) + + return self._version.stream(page, limits["limit"]) + + async def stream_async( + self, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> AsyncIterator[AccountInstance]: + """ + Asynchronously streams AccountInstance records from the API as a generator stream. + This operation lazily loads records as efficiently as possible until the limit + is reached. + The results are returned as a generator, so this operation is memory efficient. + + :param limit: Upper limit for the number of records to return. stream() + guarantees to never return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, stream() will attempt to read the + limit with the most efficient page size, i.e. min(limit, 1000) + + :returns: Generator that will yield up to limit results + """ + limits = self._version.read_limits(limit, page_size) + page = await self.page_async(page_size=limits["page_size"]) + + return self._version.stream_async(page, limits["limit"]) + + def list( + self, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> List[AccountInstance]: + """ + Lists AccountInstance records from the API as a list. + Unlike stream(), this operation is eager and will load `limit` records into + memory before returning. + + :param limit: Upper limit for the number of records to return. list() guarantees + never to return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, list() will attempt to read the limit + with the most efficient page size, i.e. min(limit, 1000) + + :returns: list that will contain up to limit results + """ + return list( + self.stream( + limit=limit, + page_size=page_size, + ) + ) + + async def list_async( + self, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> List[AccountInstance]: + """ + Asynchronously lists AccountInstance records from the API as a list. + Unlike stream(), this operation is eager and will load `limit` records into + memory before returning. + + :param limit: Upper limit for the number of records to return. list() guarantees + never to return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, list() will attempt to read the limit + with the most efficient page size, i.e. min(limit, 1000) + + :returns: list that will contain up to limit results + """ + return [ + record + async for record in await self.stream_async( + limit=limit, + page_size=page_size, + ) + ] + + def page( + self, + page_token: Union[str, object] = values.unset, + page_number: Union[int, object] = values.unset, + page_size: Union[int, object] = values.unset, + ) -> AccountPage: + """ + Retrieve a single page of AccountInstance records from the API. + Request is executed immediately + + :param page_token: PageToken provided by the API + :param page_number: Page Number, this value is simply for client state + :param page_size: Number of records to return, defaults to 50 + + :returns: Page of AccountInstance + """ + data = values.of( + { + "PageToken": page_token, + "Page": page_number, + "PageSize": page_size, + } + ) + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Accept"] = "application/json" + + response = self._version.page( + method="GET", uri=self._uri, params=data, headers=headers + ) + return AccountPage(self._version, response, self._solution) + + async def page_async( + self, + page_token: Union[str, object] = values.unset, + page_number: Union[int, object] = values.unset, + page_size: Union[int, object] = values.unset, + ) -> AccountPage: + """ + Asynchronously retrieve a single page of AccountInstance records from the API. + Request is executed immediately + + :param page_token: PageToken provided by the API + :param page_number: Page Number, this value is simply for client state + :param page_size: Number of records to return, defaults to 50 + + :returns: Page of AccountInstance + """ + data = values.of( + { + "PageToken": page_token, + "Page": page_number, + "PageSize": page_size, + } + ) + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Accept"] = "application/json" + + response = await self._version.page_async( + method="GET", uri=self._uri, params=data, headers=headers + ) + return AccountPage(self._version, response, self._solution) + + def get_page(self, target_url: str) -> AccountPage: + """ + Retrieve a specific page of AccountInstance records from the API. + Request is executed immediately + + :param target_url: API-generated URL for the requested results page + + :returns: Page of AccountInstance + """ + response = self._version.domain.twilio.request("GET", target_url) + return AccountPage(self._version, response, self._solution) + + async def get_page_async(self, target_url: str) -> AccountPage: + """ + Asynchronously retrieve a specific page of AccountInstance records from the API. + Request is executed immediately + + :param target_url: API-generated URL for the requested results page + + :returns: Page of AccountInstance + """ + response = await self._version.domain.twilio.request_async("GET", target_url) + return AccountPage(self._version, response, self._solution) + + def get(self, organization_sid: str, account_sid: str) -> AccountContext: + """ + Constructs a AccountContext + + :param organization_sid: + :param account_sid: + """ + return AccountContext( + self._version, organization_sid=organization_sid, account_sid=account_sid + ) + + def __call__(self, organization_sid: str, account_sid: str) -> AccountContext: + """ + Constructs a AccountContext + + :param organization_sid: + :param account_sid: + """ + return AccountContext( + self._version, organization_sid=organization_sid, account_sid=account_sid + ) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + return "" diff --git a/twilio/rest/preview_iam/organizations/role_assignment.py b/twilio/rest/preview_iam/organizations/role_assignment.py new file mode 100644 index 0000000000..9608e2b806 --- /dev/null +++ b/twilio/rest/preview_iam/organizations/role_assignment.py @@ -0,0 +1,588 @@ +r""" + This code was generated by + ___ _ _ _ _ _ _ ____ ____ ____ _ ____ ____ _ _ ____ ____ ____ ___ __ __ + | | | | | | | | | __ | | |__| | __ | __ |___ |\ | |___ |__/ |__| | | | |__/ + | |_|_| | |___ | |__| |__| | | | |__] |___ | \| |___ | \ | | | |__| | \ + + Organization Public API + No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator) + + NOTE: This class is auto generated by OpenAPI Generator. + https://openapi-generator.tech + Do not edit the class manually. +""" + +from typing import Any, Dict, List, Optional, Union, Iterator, AsyncIterator +from twilio.base import values +from twilio.base.instance_context import InstanceContext +from twilio.base.instance_resource import InstanceResource +from twilio.base.list_resource import ListResource +from twilio.base.version import Version +from twilio.base.page import Page + + +class RoleAssignmentInstance(InstanceResource): + + class PublicApiCreateRoleAssignmentRequest(object): + """ + :ivar role_sid: Twilio Role Sid representing assigned role + :ivar scope: Twilio Sid representing scope of this assignment + :ivar identity: Twilio Sid representing identity of this assignment + """ + + def __init__(self, payload: Dict[str, Any]): + + self.role_sid: Optional[str] = payload.get("role_sid") + self.scope: Optional[str] = payload.get("scope") + self.identity: Optional[str] = payload.get("identity") + + def to_dict(self): + return { + "role_sid": self.role_sid, + "scope": self.scope, + "identity": self.identity, + } + + """ + :ivar sid: Twilio Role Assignment Sid representing this role assignment + :ivar role_sid: Twilio Role Sid representing assigned role + :ivar scope: Twilio Sid representing identity of this assignment + :ivar identity: Twilio Sid representing scope of this assignment + :ivar code: Twilio-specific error code + :ivar message: Error message + :ivar more_info: Link to Error Code References + :ivar status: HTTP response status code + """ + + def __init__( + self, + version: Version, + payload: Dict[str, Any], + organization_sid: Optional[str] = None, + role_assignment_sid: Optional[str] = None, + ): + super().__init__(version) + + self.sid: Optional[str] = payload.get("sid") + self.role_sid: Optional[str] = payload.get("role_sid") + self.scope: Optional[str] = payload.get("scope") + self.identity: Optional[str] = payload.get("identity") + self.code: Optional[int] = payload.get("code") + self.message: Optional[str] = payload.get("message") + self.more_info: Optional[str] = payload.get("moreInfo") + self.status: Optional[int] = payload.get("status") + + self._solution = { + "organization_sid": organization_sid or self.organization_sid, + "role_assignment_sid": role_assignment_sid or self.role_assignment_sid, + } + self._context: Optional[RoleAssignmentContext] = None + + @property + def _proxy(self) -> "RoleAssignmentContext": + """ + Generate an instance context for the instance, the context is capable of + performing various actions. All instance actions are proxied to the context + + :returns: RoleAssignmentContext for this RoleAssignmentInstance + """ + if self._context is None: + self._context = RoleAssignmentContext( + self._version, + organization_sid=self._solution["organization_sid"], + role_assignment_sid=self._solution["role_assignment_sid"], + ) + return self._context + + def delete(self) -> bool: + """ + Deletes the RoleAssignmentInstance + + + :returns: True if delete succeeds, False otherwise + """ + return self._proxy.delete() + + async def delete_async(self) -> bool: + """ + Asynchronous coroutine that deletes the RoleAssignmentInstance + + + :returns: True if delete succeeds, False otherwise + """ + return await self._proxy.delete_async() + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + context = " ".join("{}={}".format(k, v) for k, v in self._solution.items()) + return "".format( + context + ) + + +class RoleAssignmentContext(InstanceContext): + + class PublicApiCreateRoleAssignmentRequest(object): + """ + :ivar role_sid: Twilio Role Sid representing assigned role + :ivar scope: Twilio Sid representing scope of this assignment + :ivar identity: Twilio Sid representing identity of this assignment + """ + + def __init__(self, payload: Dict[str, Any]): + + self.role_sid: Optional[str] = payload.get("role_sid") + self.scope: Optional[str] = payload.get("scope") + self.identity: Optional[str] = payload.get("identity") + + def to_dict(self): + return { + "role_sid": self.role_sid, + "scope": self.scope, + "identity": self.identity, + } + + def __init__( + self, version: Version, organization_sid: str, role_assignment_sid: str + ): + """ + Initialize the RoleAssignmentContext + + :param version: Version that contains the resource + :param organization_sid: + :param role_assignment_sid: + """ + super().__init__(version) + + # Path Solution + self._solution = { + "organization_sid": organization_sid, + "role_assignment_sid": role_assignment_sid, + } + self._uri = "/{organization_sid}/RoleAssignments/{role_assignment_sid}".format( + **self._solution + ) + + def delete(self) -> bool: + """ + Deletes the RoleAssignmentInstance + + + :returns: True if delete succeeds, False otherwise + """ + + headers = values.of({}) + + headers["Accept"] = "application/scim+json" + + return self._version.delete(method="DELETE", uri=self._uri, headers=headers) + + async def delete_async(self) -> bool: + """ + Asynchronous coroutine that deletes the RoleAssignmentInstance + + + :returns: True if delete succeeds, False otherwise + """ + + headers = values.of({}) + + headers["Accept"] = "application/scim+json" + + return await self._version.delete_async( + method="DELETE", uri=self._uri, headers=headers + ) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + context = " ".join("{}={}".format(k, v) for k, v in self._solution.items()) + return "".format( + context + ) + + +class RoleAssignmentPage(Page): + + def get_instance(self, payload: Dict[str, Any]) -> RoleAssignmentInstance: + """ + Build an instance of RoleAssignmentInstance + + :param payload: Payload response from the API + """ + return RoleAssignmentInstance( + self._version, payload, organization_sid=self._solution["organization_sid"] + ) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + return "" + + +class RoleAssignmentList(ListResource): + + class PublicApiCreateRoleAssignmentRequest(object): + """ + :ivar role_sid: Twilio Role Sid representing assigned role + :ivar scope: Twilio Sid representing scope of this assignment + :ivar identity: Twilio Sid representing identity of this assignment + """ + + def __init__(self, payload: Dict[str, Any]): + + self.role_sid: Optional[str] = payload.get("role_sid") + self.scope: Optional[str] = payload.get("scope") + self.identity: Optional[str] = payload.get("identity") + + def to_dict(self): + return { + "role_sid": self.role_sid, + "scope": self.scope, + "identity": self.identity, + } + + def __init__(self, version: Version, organization_sid: str): + """ + Initialize the RoleAssignmentList + + :param version: Version that contains the resource + :param organization_sid: + + """ + super().__init__(version) + + # Path Solution + self._solution = { + "organization_sid": organization_sid, + } + self._uri = "/{organization_sid}/RoleAssignments".format(**self._solution) + + def create( + self, + public_api_create_role_assignment_request: PublicApiCreateRoleAssignmentRequest, + ) -> RoleAssignmentInstance: + """ + Create the RoleAssignmentInstance + + :param public_api_create_role_assignment_request: + + :returns: The created RoleAssignmentInstance + """ + data = public_api_create_role_assignment_request.to_dict() + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Content-Type"] = "application/json" + + headers["Accept"] = "application/json" + + payload = self._version.create( + method="POST", uri=self._uri, data=data, headers=headers + ) + + return RoleAssignmentInstance( + self._version, payload, organization_sid=self._solution["organization_sid"] + ) + + async def create_async( + self, + public_api_create_role_assignment_request: PublicApiCreateRoleAssignmentRequest, + ) -> RoleAssignmentInstance: + """ + Asynchronously create the RoleAssignmentInstance + + :param public_api_create_role_assignment_request: + + :returns: The created RoleAssignmentInstance + """ + data = public_api_create_role_assignment_request.to_dict() + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Content-Type"] = "application/json" + + headers["Accept"] = "application/json" + + payload = await self._version.create_async( + method="POST", uri=self._uri, data=data, headers=headers + ) + + return RoleAssignmentInstance( + self._version, payload, organization_sid=self._solution["organization_sid"] + ) + + def stream( + self, + identity: Union[str, object] = values.unset, + scope: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> Iterator[RoleAssignmentInstance]: + """ + Streams RoleAssignmentInstance records from the API as a generator stream. + This operation lazily loads records as efficiently as possible until the limit + is reached. + The results are returned as a generator, so this operation is memory efficient. + + :param str identity: + :param str scope: + :param limit: Upper limit for the number of records to return. stream() + guarantees to never return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, stream() will attempt to read the + limit with the most efficient page size, i.e. min(limit, 1000) + + :returns: Generator that will yield up to limit results + """ + limits = self._version.read_limits(limit, page_size) + page = self.page(identity=identity, scope=scope, page_size=limits["page_size"]) + + return self._version.stream(page, limits["limit"]) + + async def stream_async( + self, + identity: Union[str, object] = values.unset, + scope: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> AsyncIterator[RoleAssignmentInstance]: + """ + Asynchronously streams RoleAssignmentInstance records from the API as a generator stream. + This operation lazily loads records as efficiently as possible until the limit + is reached. + The results are returned as a generator, so this operation is memory efficient. + + :param str identity: + :param str scope: + :param limit: Upper limit for the number of records to return. stream() + guarantees to never return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, stream() will attempt to read the + limit with the most efficient page size, i.e. min(limit, 1000) + + :returns: Generator that will yield up to limit results + """ + limits = self._version.read_limits(limit, page_size) + page = await self.page_async( + identity=identity, scope=scope, page_size=limits["page_size"] + ) + + return self._version.stream_async(page, limits["limit"]) + + def list( + self, + identity: Union[str, object] = values.unset, + scope: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> List[RoleAssignmentInstance]: + """ + Lists RoleAssignmentInstance records from the API as a list. + Unlike stream(), this operation is eager and will load `limit` records into + memory before returning. + + :param str identity: + :param str scope: + :param limit: Upper limit for the number of records to return. list() guarantees + never to return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, list() will attempt to read the limit + with the most efficient page size, i.e. min(limit, 1000) + + :returns: list that will contain up to limit results + """ + return list( + self.stream( + identity=identity, + scope=scope, + limit=limit, + page_size=page_size, + ) + ) + + async def list_async( + self, + identity: Union[str, object] = values.unset, + scope: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> List[RoleAssignmentInstance]: + """ + Asynchronously lists RoleAssignmentInstance records from the API as a list. + Unlike stream(), this operation is eager and will load `limit` records into + memory before returning. + + :param str identity: + :param str scope: + :param limit: Upper limit for the number of records to return. list() guarantees + never to return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, list() will attempt to read the limit + with the most efficient page size, i.e. min(limit, 1000) + + :returns: list that will contain up to limit results + """ + return [ + record + async for record in await self.stream_async( + identity=identity, + scope=scope, + limit=limit, + page_size=page_size, + ) + ] + + def page( + self, + identity: Union[str, object] = values.unset, + scope: Union[str, object] = values.unset, + page_token: Union[str, object] = values.unset, + page_number: Union[int, object] = values.unset, + page_size: Union[int, object] = values.unset, + ) -> RoleAssignmentPage: + """ + Retrieve a single page of RoleAssignmentInstance records from the API. + Request is executed immediately + + :param identity: + :param scope: + :param page_token: PageToken provided by the API + :param page_number: Page Number, this value is simply for client state + :param page_size: Number of records to return, defaults to 50 + + :returns: Page of RoleAssignmentInstance + """ + data = values.of( + { + "Identity": identity, + "Scope": scope, + "PageToken": page_token, + "Page": page_number, + "PageSize": page_size, + } + ) + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Accept"] = "application/json" + + response = self._version.page( + method="GET", uri=self._uri, params=data, headers=headers + ) + return RoleAssignmentPage(self._version, response, self._solution) + + async def page_async( + self, + identity: Union[str, object] = values.unset, + scope: Union[str, object] = values.unset, + page_token: Union[str, object] = values.unset, + page_number: Union[int, object] = values.unset, + page_size: Union[int, object] = values.unset, + ) -> RoleAssignmentPage: + """ + Asynchronously retrieve a single page of RoleAssignmentInstance records from the API. + Request is executed immediately + + :param identity: + :param scope: + :param page_token: PageToken provided by the API + :param page_number: Page Number, this value is simply for client state + :param page_size: Number of records to return, defaults to 50 + + :returns: Page of RoleAssignmentInstance + """ + data = values.of( + { + "Identity": identity, + "Scope": scope, + "PageToken": page_token, + "Page": page_number, + "PageSize": page_size, + } + ) + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Accept"] = "application/json" + + response = await self._version.page_async( + method="GET", uri=self._uri, params=data, headers=headers + ) + return RoleAssignmentPage(self._version, response, self._solution) + + def get_page(self, target_url: str) -> RoleAssignmentPage: + """ + Retrieve a specific page of RoleAssignmentInstance records from the API. + Request is executed immediately + + :param target_url: API-generated URL for the requested results page + + :returns: Page of RoleAssignmentInstance + """ + response = self._version.domain.twilio.request("GET", target_url) + return RoleAssignmentPage(self._version, response, self._solution) + + async def get_page_async(self, target_url: str) -> RoleAssignmentPage: + """ + Asynchronously retrieve a specific page of RoleAssignmentInstance records from the API. + Request is executed immediately + + :param target_url: API-generated URL for the requested results page + + :returns: Page of RoleAssignmentInstance + """ + response = await self._version.domain.twilio.request_async("GET", target_url) + return RoleAssignmentPage(self._version, response, self._solution) + + def get( + self, organization_sid: str, role_assignment_sid: str + ) -> RoleAssignmentContext: + """ + Constructs a RoleAssignmentContext + + :param organization_sid: + :param role_assignment_sid: + """ + return RoleAssignmentContext( + self._version, + organization_sid=organization_sid, + role_assignment_sid=role_assignment_sid, + ) + + def __call__( + self, organization_sid: str, role_assignment_sid: str + ) -> RoleAssignmentContext: + """ + Constructs a RoleAssignmentContext + + :param organization_sid: + :param role_assignment_sid: + """ + return RoleAssignmentContext( + self._version, + organization_sid=organization_sid, + role_assignment_sid=role_assignment_sid, + ) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + return "" diff --git a/twilio/rest/preview_iam/organizations/user.py b/twilio/rest/preview_iam/organizations/user.py new file mode 100644 index 0000000000..31ee699422 --- /dev/null +++ b/twilio/rest/preview_iam/organizations/user.py @@ -0,0 +1,1052 @@ +r""" + This code was generated by + ___ _ _ _ _ _ _ ____ ____ ____ _ ____ ____ _ _ ____ ____ ____ ___ __ __ + | | | | | | | | | __ | | |__| | __ | __ |___ |\ | |___ |__/ |__| | | | |__/ + | |_|_| | |___ | |__| |__| | | | |__] |___ | \| |___ | \ | | | |__| | \ + + Organization Public API + No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator) + + NOTE: This class is auto generated by OpenAPI Generator. + https://openapi-generator.tech + Do not edit the class manually. +""" + +from datetime import datetime +from typing import Any, Dict, List, Optional, Union, Iterator, AsyncIterator +from twilio.base import values +from twilio.base.instance_context import InstanceContext +from twilio.base.instance_resource import InstanceResource +from twilio.base.list_resource import ListResource +from twilio.base.version import Version +from twilio.base.page import Page + + +class UserInstance(InstanceResource): + + class ScimEmailAddress(object): + """ + :ivar primary: Indicates if this email address is the primary one + :ivar value: The actual email address value + :ivar type: The type of email address (e.g., work, home, etc.) + """ + + def __init__(self, payload: Dict[str, Any]): + + self.primary: Optional[bool] = payload.get("primary") + self.value: Optional[str] = payload.get("value") + self.type: Optional[str] = payload.get("type") + + def to_dict(self): + return { + "primary": self.primary, + "value": self.value, + "type": self.type, + } + + class ScimMeta(object): + """ + :ivar resource_type: Indicates the type of the resource + :ivar created: The date and time when the resource was created in the system + :ivar last_modified: The date and time when the resource was last modified + :ivar version: A version identifier for the resource. This can be used to manage resource versioning and concurrency control. + """ + + def __init__(self, payload: Dict[str, Any]): + + self.resource_type: Optional[str] = payload.get("resource_type") + self.created: Optional[datetime] = payload.get("created") + self.last_modified: Optional[datetime] = payload.get("last_modified") + self.version: Optional[str] = payload.get("version") + + def to_dict(self): + return { + "resource_type": self.resource_type, + "created": self.created, + "last_modified": self.last_modified, + "version": self.version, + } + + class ScimName(object): + """ + :ivar given_name: The user's first or given name + :ivar family_name: The user's last or family name + """ + + def __init__(self, payload: Dict[str, Any]): + + self.given_name: Optional[str] = payload.get("given_name") + self.family_name: Optional[str] = payload.get("family_name") + + def to_dict(self): + return { + "given_name": self.given_name, + "family_name": self.family_name, + } + + class ScimUser(object): + """ + :ivar id: Unique Twilio user sid + :ivar external_id: External unique resource id defined by provisioning client + :ivar user_name: Unique username, MUST be same as primary email address + :ivar display_name: User friendly display name + :ivar name: + :ivar emails: Email address list of the user. Primary email must be defined if there are more than 1 email. Primary email must match the username. + :ivar active: Indicates whether the user is active + :ivar locale: User's locale + :ivar timezone: User's time zone + :ivar schemas: An array of URIs that indicate the schemas supported for this user resource + :ivar meta: + :ivar detail: A human-readable description of the error + :ivar scim_type: A scimType error code as defined in RFC7644 + :ivar status: Http status code + :ivar code: Twilio-specific error code + :ivar more_info: Link to Error Code References + """ + + def __init__(self, payload: Dict[str, Any]): + + self.id: Optional[str] = payload.get("id") + self.external_id: Optional[str] = payload.get("external_id") + self.user_name: Optional[str] = payload.get("user_name") + self.display_name: Optional[str] = payload.get("display_name") + self.name: Optional[UserList.ScimName] = payload.get("name") + self.emails: Optional[List[UserList.ScimEmailAddress]] = payload.get( + "emails" + ) + self.active: Optional[bool] = payload.get("active") + self.locale: Optional[str] = payload.get("locale") + self.timezone: Optional[str] = payload.get("timezone") + self.schemas: Optional[List[str]] = payload.get("schemas") + self.meta: Optional[UserList.ScimMeta] = payload.get("meta") + self.detail: Optional[str] = payload.get("detail") + self.scim_type: Optional[str] = payload.get("scim_type") + self.status: Optional[str] = payload.get("status") + self.code: Optional[int] = payload.get("code") + self.more_info: Optional[str] = payload.get("more_info") + + def to_dict(self): + return { + "id": self.id, + "externalId": self.external_id, + "userName": self.user_name, + "displayName": self.display_name, + "name": self.name.to_dict() if self.name is not None else None, + "emails": ( + [emails.to_dict() for emails in self.emails] + if self.emails is not None + else None + ), + "active": self.active, + "locale": self.locale, + "timezone": self.timezone, + "schemas": self.schemas, + "meta": self.meta.to_dict() if self.meta is not None else None, + "detail": self.detail, + "scimType": self.scim_type, + "status": self.status, + "code": self.code, + "moreInfo": self.more_info, + } + + """ + :ivar id: Unique Twilio user sid + :ivar external_id: External unique resource id defined by provisioning client + :ivar user_name: Unique username, MUST be same as primary email address + :ivar display_name: User friendly display name + :ivar name: + :ivar emails: Email address list of the user. Primary email must be defined if there are more than 1 email. Primary email must match the username. + :ivar active: Indicates whether the user is active + :ivar locale: User's locale + :ivar timezone: User's time zone + :ivar schemas: An array of URIs that indicate the schemas supported for this user resource + :ivar meta: + :ivar detail: A human-readable description of the error + :ivar scim_type: A scimType error code as defined in RFC7644 + :ivar status: Http status code + :ivar code: Twilio-specific error code + :ivar more_info: Link to Error Code References + """ + + def __init__( + self, + version: Version, + payload: Dict[str, Any], + organization_sid: Optional[str] = None, + user_sid: Optional[str] = None, + ): + super().__init__(version) + + self.id: Optional[str] = payload.get("id") + self.external_id: Optional[str] = payload.get("externalId") + self.user_name: Optional[str] = payload.get("userName") + self.display_name: Optional[str] = payload.get("displayName") + self.name: Optional[UserList.str] = payload.get("name") + self.emails: Optional[List[UserList.str]] = payload.get("emails") + self.active: Optional[bool] = payload.get("active") + self.locale: Optional[str] = payload.get("locale") + self.timezone: Optional[str] = payload.get("timezone") + self.schemas: Optional[List[str]] = payload.get("schemas") + self.meta: Optional[UserList.str] = payload.get("meta") + self.detail: Optional[str] = payload.get("detail") + self.scim_type: Optional[str] = payload.get("scimType") + self.status: Optional[str] = payload.get("status") + self.code: Optional[int] = payload.get("code") + self.more_info: Optional[str] = payload.get("moreInfo") + + self._solution = { + "organization_sid": organization_sid or self.organization_sid, + "user_sid": user_sid or self.user_sid, + } + self._context: Optional[UserContext] = None + + @property + def _proxy(self) -> "UserContext": + """ + Generate an instance context for the instance, the context is capable of + performing various actions. All instance actions are proxied to the context + + :returns: UserContext for this UserInstance + """ + if self._context is None: + self._context = UserContext( + self._version, + organization_sid=self._solution["organization_sid"], + user_sid=self._solution["user_sid"], + ) + return self._context + + def delete(self) -> bool: + """ + Deletes the UserInstance + + + :returns: True if delete succeeds, False otherwise + """ + return self._proxy.delete() + + async def delete_async(self) -> bool: + """ + Asynchronous coroutine that deletes the UserInstance + + + :returns: True if delete succeeds, False otherwise + """ + return await self._proxy.delete_async() + + def fetch(self) -> "UserInstance": + """ + Fetch the UserInstance + + + :returns: The fetched UserInstance + """ + return self._proxy.fetch() + + async def fetch_async(self) -> "UserInstance": + """ + Asynchronous coroutine to fetch the UserInstance + + + :returns: The fetched UserInstance + """ + return await self._proxy.fetch_async() + + def update( + self, scim_user: ScimUser, if_match: Union[str, object] = values.unset + ) -> "UserInstance": + """ + Update the UserInstance + + :param scim_user: + :param if_match: + + :returns: The updated UserInstance + """ + return self._proxy.update( + scim_user=scim_user, + if_match=if_match, + ) + + async def update_async( + self, scim_user: ScimUser, if_match: Union[str, object] = values.unset + ) -> "UserInstance": + """ + Asynchronous coroutine to update the UserInstance + + :param scim_user: + :param if_match: + + :returns: The updated UserInstance + """ + return await self._proxy.update_async( + scim_user=scim_user, + if_match=if_match, + ) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + context = " ".join("{}={}".format(k, v) for k, v in self._solution.items()) + return "".format(context) + + +class UserContext(InstanceContext): + + class ScimEmailAddress(object): + """ + :ivar primary: Indicates if this email address is the primary one + :ivar value: The actual email address value + :ivar type: The type of email address (e.g., work, home, etc.) + """ + + def __init__(self, payload: Dict[str, Any]): + + self.primary: Optional[bool] = payload.get("primary") + self.value: Optional[str] = payload.get("value") + self.type: Optional[str] = payload.get("type") + + def to_dict(self): + return { + "primary": self.primary, + "value": self.value, + "type": self.type, + } + + class ScimMeta(object): + """ + :ivar resource_type: Indicates the type of the resource + :ivar created: The date and time when the resource was created in the system + :ivar last_modified: The date and time when the resource was last modified + :ivar version: A version identifier for the resource. This can be used to manage resource versioning and concurrency control. + """ + + def __init__(self, payload: Dict[str, Any]): + + self.resource_type: Optional[str] = payload.get("resource_type") + self.created: Optional[datetime] = payload.get("created") + self.last_modified: Optional[datetime] = payload.get("last_modified") + self.version: Optional[str] = payload.get("version") + + def to_dict(self): + return { + "resource_type": self.resource_type, + "created": self.created, + "last_modified": self.last_modified, + "version": self.version, + } + + class ScimName(object): + """ + :ivar given_name: The user's first or given name + :ivar family_name: The user's last or family name + """ + + def __init__(self, payload: Dict[str, Any]): + + self.given_name: Optional[str] = payload.get("given_name") + self.family_name: Optional[str] = payload.get("family_name") + + def to_dict(self): + return { + "given_name": self.given_name, + "family_name": self.family_name, + } + + class ScimUser(object): + """ + :ivar id: Unique Twilio user sid + :ivar external_id: External unique resource id defined by provisioning client + :ivar user_name: Unique username, MUST be same as primary email address + :ivar display_name: User friendly display name + :ivar name: + :ivar emails: Email address list of the user. Primary email must be defined if there are more than 1 email. Primary email must match the username. + :ivar active: Indicates whether the user is active + :ivar locale: User's locale + :ivar timezone: User's time zone + :ivar schemas: An array of URIs that indicate the schemas supported for this user resource + :ivar meta: + :ivar detail: A human-readable description of the error + :ivar scim_type: A scimType error code as defined in RFC7644 + :ivar status: Http status code + :ivar code: Twilio-specific error code + :ivar more_info: Link to Error Code References + """ + + def __init__(self, payload: Dict[str, Any]): + + self.id: Optional[str] = payload.get("id") + self.external_id: Optional[str] = payload.get("external_id") + self.user_name: Optional[str] = payload.get("user_name") + self.display_name: Optional[str] = payload.get("display_name") + self.name: Optional[UserList.ScimName] = payload.get("name") + self.emails: Optional[List[UserList.ScimEmailAddress]] = payload.get( + "emails" + ) + self.active: Optional[bool] = payload.get("active") + self.locale: Optional[str] = payload.get("locale") + self.timezone: Optional[str] = payload.get("timezone") + self.schemas: Optional[List[str]] = payload.get("schemas") + self.meta: Optional[UserList.ScimMeta] = payload.get("meta") + self.detail: Optional[str] = payload.get("detail") + self.scim_type: Optional[str] = payload.get("scim_type") + self.status: Optional[str] = payload.get("status") + self.code: Optional[int] = payload.get("code") + self.more_info: Optional[str] = payload.get("more_info") + + def to_dict(self): + return { + "id": self.id, + "externalId": self.external_id, + "userName": self.user_name, + "displayName": self.display_name, + "name": self.name.to_dict() if self.name is not None else None, + "emails": ( + [emails.to_dict() for emails in self.emails] + if self.emails is not None + else None + ), + "active": self.active, + "locale": self.locale, + "timezone": self.timezone, + "schemas": self.schemas, + "meta": self.meta.to_dict() if self.meta is not None else None, + "detail": self.detail, + "scimType": self.scim_type, + "status": self.status, + "code": self.code, + "moreInfo": self.more_info, + } + + def __init__(self, version: Version, organization_sid: str, user_sid: str): + """ + Initialize the UserContext + + :param version: Version that contains the resource + :param organization_sid: + :param user_sid: + """ + super().__init__(version) + + # Path Solution + self._solution = { + "organization_sid": organization_sid, + "user_sid": user_sid, + } + self._uri = "/{organization_sid}/scim/Users/{user_sid}".format(**self._solution) + + def delete(self) -> bool: + """ + Deletes the UserInstance + + + :returns: True if delete succeeds, False otherwise + """ + + headers = values.of({}) + + headers["Accept"] = "application/scim+json" + + return self._version.delete(method="DELETE", uri=self._uri, headers=headers) + + async def delete_async(self) -> bool: + """ + Asynchronous coroutine that deletes the UserInstance + + + :returns: True if delete succeeds, False otherwise + """ + + headers = values.of({}) + + headers["Accept"] = "application/scim+json" + + return await self._version.delete_async( + method="DELETE", uri=self._uri, headers=headers + ) + + def fetch(self) -> UserInstance: + """ + Fetch the UserInstance + + + :returns: The fetched UserInstance + """ + + headers = values.of({}) + + headers["Accept"] = "application/scim+json" + + payload = self._version.fetch(method="GET", uri=self._uri, headers=headers) + + return UserInstance( + self._version, + payload, + organization_sid=self._solution["organization_sid"], + user_sid=self._solution["user_sid"], + ) + + async def fetch_async(self) -> UserInstance: + """ + Asynchronous coroutine to fetch the UserInstance + + + :returns: The fetched UserInstance + """ + + headers = values.of({}) + + headers["Accept"] = "application/scim+json" + + payload = await self._version.fetch_async( + method="GET", uri=self._uri, headers=headers + ) + + return UserInstance( + self._version, + payload, + organization_sid=self._solution["organization_sid"], + user_sid=self._solution["user_sid"], + ) + + def update( + self, scim_user: ScimUser, if_match: Union[str, object] = values.unset + ) -> UserInstance: + """ + Update the UserInstance + + :param scim_user: + :param if_match: + + :returns: The updated UserInstance + """ + data = scim_user.to_dict() + + headers = values.of({}) + + if not ( + if_match is values.unset or (isinstance(if_match, str) and not if_match) + ): + headers["If-Match"] = if_match + + headers["Content-Type"] = "application/json" + + headers["Content-Type"] = "application/scim+json" + + headers["Accept"] = "application/scim+json" + + payload = self._version.update( + method="PUT", uri=self._uri, data=data, headers=headers + ) + + return UserInstance( + self._version, + payload, + organization_sid=self._solution["organization_sid"], + user_sid=self._solution["user_sid"], + ) + + async def update_async( + self, scim_user: ScimUser, if_match: Union[str, object] = values.unset + ) -> UserInstance: + """ + Asynchronous coroutine to update the UserInstance + + :param scim_user: + :param if_match: + + :returns: The updated UserInstance + """ + data = scim_user.to_dict() + + headers = values.of({}) + + if not ( + if_match is values.unset or (isinstance(if_match, str) and not if_match) + ): + headers["If-Match"] = if_match + + headers["Content-Type"] = "application/json" + + headers["Content-Type"] = "application/scim+json" + + headers["Accept"] = "application/scim+json" + + payload = await self._version.update_async( + method="PUT", uri=self._uri, data=data, headers=headers + ) + + return UserInstance( + self._version, + payload, + organization_sid=self._solution["organization_sid"], + user_sid=self._solution["user_sid"], + ) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + context = " ".join("{}={}".format(k, v) for k, v in self._solution.items()) + return "".format(context) + + +class UserPage(Page): + + def get_instance(self, payload: Dict[str, Any]) -> UserInstance: + """ + Build an instance of UserInstance + + :param payload: Payload response from the API + """ + return UserInstance( + self._version, payload, organization_sid=self._solution["organization_sid"] + ) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + return "" + + +class UserList(ListResource): + + class ScimEmailAddress(object): + """ + :ivar primary: Indicates if this email address is the primary one + :ivar value: The actual email address value + :ivar type: The type of email address (e.g., work, home, etc.) + """ + + def __init__(self, payload: Dict[str, Any]): + + self.primary: Optional[bool] = payload.get("primary") + self.value: Optional[str] = payload.get("value") + self.type: Optional[str] = payload.get("type") + + def to_dict(self): + return { + "primary": self.primary, + "value": self.value, + "type": self.type, + } + + class ScimMeta(object): + """ + :ivar resource_type: Indicates the type of the resource + :ivar created: The date and time when the resource was created in the system + :ivar last_modified: The date and time when the resource was last modified + :ivar version: A version identifier for the resource. This can be used to manage resource versioning and concurrency control. + """ + + def __init__(self, payload: Dict[str, Any]): + + self.resource_type: Optional[str] = payload.get("resource_type") + self.created: Optional[datetime] = payload.get("created") + self.last_modified: Optional[datetime] = payload.get("last_modified") + self.version: Optional[str] = payload.get("version") + + def to_dict(self): + return { + "resource_type": self.resource_type, + "created": self.created, + "last_modified": self.last_modified, + "version": self.version, + } + + class ScimName(object): + """ + :ivar given_name: The user's first or given name + :ivar family_name: The user's last or family name + """ + + def __init__(self, payload: Dict[str, Any]): + + self.given_name: Optional[str] = payload.get("given_name") + self.family_name: Optional[str] = payload.get("family_name") + + def to_dict(self): + return { + "given_name": self.given_name, + "family_name": self.family_name, + } + + class ScimUser(object): + """ + :ivar id: Unique Twilio user sid + :ivar external_id: External unique resource id defined by provisioning client + :ivar user_name: Unique username, MUST be same as primary email address + :ivar display_name: User friendly display name + :ivar name: + :ivar emails: Email address list of the user. Primary email must be defined if there are more than 1 email. Primary email must match the username. + :ivar active: Indicates whether the user is active + :ivar locale: User's locale + :ivar timezone: User's time zone + :ivar schemas: An array of URIs that indicate the schemas supported for this user resource + :ivar meta: + :ivar detail: A human-readable description of the error + :ivar scim_type: A scimType error code as defined in RFC7644 + :ivar status: Http status code + :ivar code: Twilio-specific error code + :ivar more_info: Link to Error Code References + """ + + def __init__(self, payload: Dict[str, Any]): + + self.id: Optional[str] = payload.get("id") + self.external_id: Optional[str] = payload.get("external_id") + self.user_name: Optional[str] = payload.get("user_name") + self.display_name: Optional[str] = payload.get("display_name") + self.name: Optional[UserList.ScimName] = payload.get("name") + self.emails: Optional[List[UserList.ScimEmailAddress]] = payload.get( + "emails" + ) + self.active: Optional[bool] = payload.get("active") + self.locale: Optional[str] = payload.get("locale") + self.timezone: Optional[str] = payload.get("timezone") + self.schemas: Optional[List[str]] = payload.get("schemas") + self.meta: Optional[UserList.ScimMeta] = payload.get("meta") + self.detail: Optional[str] = payload.get("detail") + self.scim_type: Optional[str] = payload.get("scim_type") + self.status: Optional[str] = payload.get("status") + self.code: Optional[int] = payload.get("code") + self.more_info: Optional[str] = payload.get("more_info") + + def to_dict(self): + return { + "id": self.id, + "externalId": self.external_id, + "userName": self.user_name, + "displayName": self.display_name, + "name": self.name.to_dict() if self.name is not None else None, + "emails": ( + [emails.to_dict() for emails in self.emails] + if self.emails is not None + else None + ), + "active": self.active, + "locale": self.locale, + "timezone": self.timezone, + "schemas": self.schemas, + "meta": self.meta.to_dict() if self.meta is not None else None, + "detail": self.detail, + "scimType": self.scim_type, + "status": self.status, + "code": self.code, + "moreInfo": self.more_info, + } + + def __init__(self, version: Version, organization_sid: str): + """ + Initialize the UserList + + :param version: Version that contains the resource + :param organization_sid: + + """ + super().__init__(version) + + # Path Solution + self._solution = { + "organization_sid": organization_sid, + } + self._uri = "/{organization_sid}/scim/Users".format(**self._solution) + + def create(self, scim_user: ScimUser) -> UserInstance: + """ + Create the UserInstance + + :param scim_user: + + :returns: The created UserInstance + """ + data = scim_user.to_dict() + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Content-Type"] = "application/json" + + headers["Content-Type"] = "application/scim+json" + + headers["Accept"] = "application/scim+json" + + payload = self._version.create( + method="POST", uri=self._uri, data=data, headers=headers + ) + + return UserInstance( + self._version, payload, organization_sid=self._solution["organization_sid"] + ) + + async def create_async(self, scim_user: ScimUser) -> UserInstance: + """ + Asynchronously create the UserInstance + + :param scim_user: + + :returns: The created UserInstance + """ + data = scim_user.to_dict() + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Content-Type"] = "application/json" + + headers["Content-Type"] = "application/scim+json" + + headers["Accept"] = "application/scim+json" + + payload = await self._version.create_async( + method="POST", uri=self._uri, data=data, headers=headers + ) + + return UserInstance( + self._version, payload, organization_sid=self._solution["organization_sid"] + ) + + def stream( + self, + filter: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> Iterator[UserInstance]: + """ + Streams UserInstance records from the API as a generator stream. + This operation lazily loads records as efficiently as possible until the limit + is reached. + The results are returned as a generator, so this operation is memory efficient. + + :param str filter: + :param limit: Upper limit for the number of records to return. stream() + guarantees to never return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, stream() will attempt to read the + limit with the most efficient page size, i.e. min(limit, 1000) + + :returns: Generator that will yield up to limit results + """ + limits = self._version.read_limits(limit, page_size) + page = self.page(filter=filter, page_size=limits["page_size"]) + + return self._version.stream(page, limits["limit"]) + + async def stream_async( + self, + filter: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> AsyncIterator[UserInstance]: + """ + Asynchronously streams UserInstance records from the API as a generator stream. + This operation lazily loads records as efficiently as possible until the limit + is reached. + The results are returned as a generator, so this operation is memory efficient. + + :param str filter: + :param limit: Upper limit for the number of records to return. stream() + guarantees to never return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, stream() will attempt to read the + limit with the most efficient page size, i.e. min(limit, 1000) + + :returns: Generator that will yield up to limit results + """ + limits = self._version.read_limits(limit, page_size) + page = await self.page_async(filter=filter, page_size=limits["page_size"]) + + return self._version.stream_async(page, limits["limit"]) + + def list( + self, + filter: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> List[UserInstance]: + """ + Lists UserInstance records from the API as a list. + Unlike stream(), this operation is eager and will load `limit` records into + memory before returning. + + :param str filter: + :param limit: Upper limit for the number of records to return. list() guarantees + never to return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, list() will attempt to read the limit + with the most efficient page size, i.e. min(limit, 1000) + + :returns: list that will contain up to limit results + """ + return list( + self.stream( + filter=filter, + limit=limit, + page_size=page_size, + ) + ) + + async def list_async( + self, + filter: Union[str, object] = values.unset, + limit: Optional[int] = None, + page_size: Optional[int] = None, + ) -> List[UserInstance]: + """ + Asynchronously lists UserInstance records from the API as a list. + Unlike stream(), this operation is eager and will load `limit` records into + memory before returning. + + :param str filter: + :param limit: Upper limit for the number of records to return. list() guarantees + never to return more than limit. Default is no limit + :param page_size: Number of records to fetch per request, when not set will use + the default value of 50 records. If no page_size is defined + but a limit is defined, list() will attempt to read the limit + with the most efficient page size, i.e. min(limit, 1000) + + :returns: list that will contain up to limit results + """ + return [ + record + async for record in await self.stream_async( + filter=filter, + limit=limit, + page_size=page_size, + ) + ] + + def page( + self, + filter: Union[str, object] = values.unset, + page_token: Union[str, object] = values.unset, + page_number: Union[int, object] = values.unset, + page_size: Union[int, object] = values.unset, + ) -> UserPage: + """ + Retrieve a single page of UserInstance records from the API. + Request is executed immediately + + :param filter: + :param page_token: PageToken provided by the API + :param page_number: Page Number, this value is simply for client state + :param page_size: Number of records to return, defaults to 50 + + :returns: Page of UserInstance + """ + data = values.of( + { + "filter": filter, + "PageToken": page_token, + "Page": page_number, + "PageSize": page_size, + } + ) + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Accept"] = "application/scim+json" + + response = self._version.page( + method="GET", uri=self._uri, params=data, headers=headers + ) + return UserPage(self._version, response, self._solution) + + async def page_async( + self, + filter: Union[str, object] = values.unset, + page_token: Union[str, object] = values.unset, + page_number: Union[int, object] = values.unset, + page_size: Union[int, object] = values.unset, + ) -> UserPage: + """ + Asynchronously retrieve a single page of UserInstance records from the API. + Request is executed immediately + + :param filter: + :param page_token: PageToken provided by the API + :param page_number: Page Number, this value is simply for client state + :param page_size: Number of records to return, defaults to 50 + + :returns: Page of UserInstance + """ + data = values.of( + { + "filter": filter, + "PageToken": page_token, + "Page": page_number, + "PageSize": page_size, + } + ) + + headers = values.of({"Content-Type": "application/x-www-form-urlencoded"}) + + headers["Accept"] = "application/scim+json" + + response = await self._version.page_async( + method="GET", uri=self._uri, params=data, headers=headers + ) + return UserPage(self._version, response, self._solution) + + def get_page(self, target_url: str) -> UserPage: + """ + Retrieve a specific page of UserInstance records from the API. + Request is executed immediately + + :param target_url: API-generated URL for the requested results page + + :returns: Page of UserInstance + """ + response = self._version.domain.twilio.request("GET", target_url) + return UserPage(self._version, response, self._solution) + + async def get_page_async(self, target_url: str) -> UserPage: + """ + Asynchronously retrieve a specific page of UserInstance records from the API. + Request is executed immediately + + :param target_url: API-generated URL for the requested results page + + :returns: Page of UserInstance + """ + response = await self._version.domain.twilio.request_async("GET", target_url) + return UserPage(self._version, response, self._solution) + + def get(self, organization_sid: str, user_sid: str) -> UserContext: + """ + Constructs a UserContext + + :param organization_sid: + :param user_sid: + """ + return UserContext( + self._version, organization_sid=organization_sid, user_sid=user_sid + ) + + def __call__(self, organization_sid: str, user_sid: str) -> UserContext: + """ + Constructs a UserContext + + :param organization_sid: + :param user_sid: + """ + return UserContext( + self._version, organization_sid=organization_sid, user_sid=user_sid + ) + + def __repr__(self) -> str: + """ + Provide a friendly representation + + :returns: Machine friendly representation + """ + return ""