diff --git a/.gitignore b/.gitignore index e42a270..dd7c4fc 100644 --- a/.gitignore +++ b/.gitignore @@ -29,3 +29,4 @@ env/ .\#* .projectile +.idea/ \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 5a92322..2c99534 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ ## Unreleased +## [0.0.7] - 2024-10-25 + +### Added + +- Include ERC20 and ERC721 token transfer information into transaction content. +- Support for wallet and address webhooks to trigger based on onchain activities. + ## [0.0.6] - 2024-10-17 ### Added diff --git a/README.md b/README.md index 1bf05fd..9fc8dae 100644 --- a/README.md +++ b/README.md @@ -169,6 +169,31 @@ print(f"Trade successfully completed: {trade}") list(address.trades()) ``` +## Creating a Webhook +A webhook is a way to provide other applications with real-time information from the blockchain. When an event occurs on a blockchain address, it can send a POST request to a URL you specify. You can create a webhook to receive notifications about events that occur in your wallet or crypto address, such as when a user makes a transfer. +```python +from cdp.client.models.webhook import WebhookEventType +from cdp.client.models.webhook import WebhookEventFilter + +wh1 = Webhook.create( + notification_uri="https://your-app.com/callback", + event_type=WebhookEventType.ERC20_TRANSFER, + event_filters=[WebhookEventFilter(from_address="0x71d4d7d5e9ce0f41e6a68bd3a9b43aa597dc0eb0")] +) +print(wh1) +``` + +## Creating a Webhook On A Wallet +A webhook can be attached to an existing wallet to monitor events that occur on the wallet, i.e. all addresses associated with this wallet. A list of supported blockchain events can be found [here](https://docs.cdp.coinbase.com/get-started/docs/webhooks/event-types). +```python +import cdp + +wallet1 = Wallet.create() +wh1 = wallet1.create_webhook("https://your-app.com/callback") +print(wh1) +``` + ## Contributing See [CONTRIBUTING.md](CONTRIBUTING.md) for more information. + diff --git a/cdp/__init__.py b/cdp/__init__.py index 33a4168..1473e30 100644 --- a/cdp/__init__.py +++ b/cdp/__init__.py @@ -16,6 +16,7 @@ from cdp.wallet import Wallet from cdp.wallet_address import WalletAddress from cdp.wallet_data import WalletData +from cdp.webhook import Webhook __all__ = [ "__version__", @@ -24,6 +25,7 @@ "Wallet", "WalletAddress", "WalletData", + "Webhook", "Asset", "Transfer", "Address", diff --git a/cdp/__version__.py b/cdp/__version__.py index 034f46c..6526deb 100644 --- a/cdp/__version__.py +++ b/cdp/__version__.py @@ -1 +1 @@ -__version__ = "0.0.6" +__version__ = "0.0.7" diff --git a/cdp/api_clients.py b/cdp/api_clients.py index b93c943..46c7532 100644 --- a/cdp/api_clients.py +++ b/cdp/api_clients.py @@ -10,6 +10,7 @@ from cdp.client.api.transaction_history_api import TransactionHistoryApi from cdp.client.api.transfers_api import TransfersApi from cdp.client.api.wallets_api import WalletsApi +from cdp.client.api.webhooks_api import WebhooksApi class ApiClients: @@ -21,6 +22,7 @@ class ApiClients: Attributes: _cdp_client (CdpApiClient): The CDP API client used to initialize individual API clients. _wallets (Optional[WalletsApi]): The WalletsApi client instance. + _webhooks (Optional[WebhooksApi]): The WebhooksApi client instance. _addresses (Optional[AddressesApi]): The AddressesApi client instance. _external_addresses (Optional[ExternalAddressesApi]): The ExternalAddressesApi client instance. _transfers (Optional[TransfersApi]): The TransfersApi client instance. @@ -40,6 +42,7 @@ def __init__(self, cdp_client: CdpApiClient) -> None: """ self._cdp_client: CdpApiClient = cdp_client self._wallets: WalletsApi | None = None + self._webhooks: WebhooksApi | None = None self._addresses: AddressesApi | None = None self._external_addresses: ExternalAddressesApi | None = None self._transfers: TransfersApi | None = None @@ -66,6 +69,21 @@ def wallets(self) -> WalletsApi: self._wallets = WalletsApi(api_client=self._cdp_client) return self._wallets + @property + def webhooks(self) -> WebhooksApi: + """Get the WebhooksApi client instance. + + Returns: + WebhooksApi: The WebhooksApi client instance. + + Note: + This property lazily initializes the WebhooksApi client on first access. + + """ + if self._webhooks is None: + self._webhooks = WebhooksApi(api_client=self._cdp_client) + return self._webhooks + @property def addresses(self) -> AddressesApi: """Get the AddressesApi client instance. diff --git a/cdp/client/__init__.py b/cdp/client/__init__.py index 018445d..4fbe28c 100644 --- a/cdp/client/__init__.py +++ b/cdp/client/__init__.py @@ -24,6 +24,7 @@ from cdp.client.api.contract_invocations_api import ContractInvocationsApi from cdp.client.api.external_addresses_api import ExternalAddressesApi from cdp.client.api.networks_api import NetworksApi +from cdp.client.api.onchain_identity_api import OnchainIdentityApi from cdp.client.api.server_signers_api import ServerSignersApi from cdp.client.api.smart_contracts_api import SmartContractsApi from cdp.client.api.stake_api import StakeApi @@ -96,6 +97,9 @@ from cdp.client.models.nft_contract_options import NFTContractOptions from cdp.client.models.network import Network from cdp.client.models.network_identifier import NetworkIdentifier +from cdp.client.models.onchain_name import OnchainName +from cdp.client.models.onchain_name_list import OnchainNameList +from cdp.client.models.onchain_name_text_records_inner import OnchainNameTextRecordsInner from cdp.client.models.payload_signature import PayloadSignature from cdp.client.models.payload_signature_list import PayloadSignatureList from cdp.client.models.read_contract_request import ReadContractRequest diff --git a/cdp/client/api/__init__.py b/cdp/client/api/__init__.py index 28574b9..221b4a3 100644 --- a/cdp/client/api/__init__.py +++ b/cdp/client/api/__init__.py @@ -8,6 +8,7 @@ from cdp.client.api.contract_invocations_api import ContractInvocationsApi from cdp.client.api.external_addresses_api import ExternalAddressesApi from cdp.client.api.networks_api import NetworksApi +from cdp.client.api.onchain_identity_api import OnchainIdentityApi from cdp.client.api.server_signers_api import ServerSignersApi from cdp.client.api.smart_contracts_api import SmartContractsApi from cdp.client.api.stake_api import StakeApi diff --git a/cdp/client/api/onchain_identity_api.py b/cdp/client/api/onchain_identity_api.py new file mode 100644 index 0000000..06b80ac --- /dev/null +++ b/cdp/client/api/onchain_identity_api.py @@ -0,0 +1,346 @@ +# coding: utf-8 + +""" + Coinbase Platform API + + This is the OpenAPI 3.0 specification for the Coinbase Platform APIs, used in conjunction with the Coinbase Platform SDKs. + + The version of the OpenAPI document: 0.0.1-alpha + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + +import warnings +from pydantic import validate_call, Field, StrictFloat, StrictStr, StrictInt +from typing import Any, Dict, List, Optional, Tuple, Union +from typing_extensions import Annotated + +from pydantic import Field, StrictInt, StrictStr +from typing import Optional +from typing_extensions import Annotated +from cdp.client.models.onchain_name_list import OnchainNameList + +from cdp.client.api_client import ApiClient, RequestSerialized +from cdp.client.api_response import ApiResponse +from cdp.client.rest import RESTResponseType + + +class OnchainIdentityApi: + """NOTE: This class is auto generated by OpenAPI Generator + Ref: https://openapi-generator.tech + + Do not edit the class manually. + """ + + def __init__(self, api_client=None) -> None: + if api_client is None: + api_client = ApiClient.get_default() + self.api_client = api_client + + + @validate_call + def resolve_identity_by_address( + self, + network_id: Annotated[StrictStr, Field(description="The ID of the blockchain network")], + address_id: Annotated[StrictStr, Field(description="The ID of the address to fetch the identity for")], + limit: Annotated[Optional[StrictInt], Field(description="A limit on the number of objects to be returned. Limit can range between 1 and 100, and the default is 10.")] = None, + page: Annotated[Optional[Annotated[str, Field(strict=True, max_length=5000)]], Field(description="A cursor for pagination across multiple pages of results. Don't include this parameter on the first call. Use the next_page value returned in a previous response to request subsequent results.")] = None, + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> OnchainNameList: + """Obtains onchain identity for an address on a specific network + + Obtains onchain identity for an address on a specific network + + :param network_id: The ID of the blockchain network (required) + :type network_id: str + :param address_id: The ID of the address to fetch the identity for (required) + :type address_id: str + :param limit: A limit on the number of objects to be returned. Limit can range between 1 and 100, and the default is 10. + :type limit: int + :param page: A cursor for pagination across multiple pages of results. Don't include this parameter on the first call. Use the next_page value returned in a previous response to request subsequent results. + :type page: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._resolve_identity_by_address_serialize( + network_id=network_id, + address_id=address_id, + limit=limit, + page=page, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "OnchainNameList", + } + response_data = self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + response_data.read() + return self.api_client.response_deserialize( + response_data=response_data, + response_types_map=_response_types_map, + ).data + + + @validate_call + def resolve_identity_by_address_with_http_info( + self, + network_id: Annotated[StrictStr, Field(description="The ID of the blockchain network")], + address_id: Annotated[StrictStr, Field(description="The ID of the address to fetch the identity for")], + limit: Annotated[Optional[StrictInt], Field(description="A limit on the number of objects to be returned. Limit can range between 1 and 100, and the default is 10.")] = None, + page: Annotated[Optional[Annotated[str, Field(strict=True, max_length=5000)]], Field(description="A cursor for pagination across multiple pages of results. Don't include this parameter on the first call. Use the next_page value returned in a previous response to request subsequent results.")] = None, + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> ApiResponse[OnchainNameList]: + """Obtains onchain identity for an address on a specific network + + Obtains onchain identity for an address on a specific network + + :param network_id: The ID of the blockchain network (required) + :type network_id: str + :param address_id: The ID of the address to fetch the identity for (required) + :type address_id: str + :param limit: A limit on the number of objects to be returned. Limit can range between 1 and 100, and the default is 10. + :type limit: int + :param page: A cursor for pagination across multiple pages of results. Don't include this parameter on the first call. Use the next_page value returned in a previous response to request subsequent results. + :type page: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._resolve_identity_by_address_serialize( + network_id=network_id, + address_id=address_id, + limit=limit, + page=page, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "OnchainNameList", + } + response_data = self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + response_data.read() + return self.api_client.response_deserialize( + response_data=response_data, + response_types_map=_response_types_map, + ) + + + @validate_call + def resolve_identity_by_address_without_preload_content( + self, + network_id: Annotated[StrictStr, Field(description="The ID of the blockchain network")], + address_id: Annotated[StrictStr, Field(description="The ID of the address to fetch the identity for")], + limit: Annotated[Optional[StrictInt], Field(description="A limit on the number of objects to be returned. Limit can range between 1 and 100, and the default is 10.")] = None, + page: Annotated[Optional[Annotated[str, Field(strict=True, max_length=5000)]], Field(description="A cursor for pagination across multiple pages of results. Don't include this parameter on the first call. Use the next_page value returned in a previous response to request subsequent results.")] = None, + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> RESTResponseType: + """Obtains onchain identity for an address on a specific network + + Obtains onchain identity for an address on a specific network + + :param network_id: The ID of the blockchain network (required) + :type network_id: str + :param address_id: The ID of the address to fetch the identity for (required) + :type address_id: str + :param limit: A limit on the number of objects to be returned. Limit can range between 1 and 100, and the default is 10. + :type limit: int + :param page: A cursor for pagination across multiple pages of results. Don't include this parameter on the first call. Use the next_page value returned in a previous response to request subsequent results. + :type page: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._resolve_identity_by_address_serialize( + network_id=network_id, + address_id=address_id, + limit=limit, + page=page, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "OnchainNameList", + } + response_data = self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + return response_data.response + + + def _resolve_identity_by_address_serialize( + self, + network_id, + address_id, + limit, + page, + _request_auth, + _content_type, + _headers, + _host_index, + ) -> RequestSerialized: + + _host = None + + _collection_formats: Dict[str, str] = { + } + + _path_params: Dict[str, str] = {} + _query_params: List[Tuple[str, str]] = [] + _header_params: Dict[str, Optional[str]] = _headers or {} + _form_params: List[Tuple[str, str]] = [] + _files: Dict[str, Union[str, bytes]] = {} + _body_params: Optional[bytes] = None + + # process the path parameters + if network_id is not None: + _path_params['network_id'] = network_id + if address_id is not None: + _path_params['address_id'] = address_id + # process the query parameters + if limit is not None: + + _query_params.append(('limit', limit)) + + if page is not None: + + _query_params.append(('page', page)) + + # process the header parameters + # process the form parameters + # process the body parameter + + + # set the HTTP header `Accept` + if 'Accept' not in _header_params: + _header_params['Accept'] = self.api_client.select_header_accept( + [ + 'application/json' + ] + ) + + + # authentication setting + _auth_settings: List[str] = [ + ] + + return self.api_client.param_serialize( + method='GET', + resource_path='/v1/networks/{network_id}/addresses/{address_id}/identity', + path_params=_path_params, + query_params=_query_params, + header_params=_header_params, + body=_body_params, + post_params=_form_params, + files=_files, + auth_settings=_auth_settings, + collection_formats=_collection_formats, + _host=_host, + _request_auth=_request_auth + ) + + diff --git a/cdp/client/models/__init__.py b/cdp/client/models/__init__.py index a342db7..6cda9a0 100644 --- a/cdp/client/models/__init__.py +++ b/cdp/client/models/__init__.py @@ -62,6 +62,9 @@ from cdp.client.models.nft_contract_options import NFTContractOptions from cdp.client.models.network import Network from cdp.client.models.network_identifier import NetworkIdentifier +from cdp.client.models.onchain_name import OnchainName +from cdp.client.models.onchain_name_list import OnchainNameList +from cdp.client.models.onchain_name_text_records_inner import OnchainNameTextRecordsInner from cdp.client.models.payload_signature import PayloadSignature from cdp.client.models.payload_signature_list import PayloadSignatureList from cdp.client.models.read_contract_request import ReadContractRequest diff --git a/cdp/client/models/ethereum_token_transfer.py b/cdp/client/models/ethereum_token_transfer.py new file mode 100644 index 0000000..414451e --- /dev/null +++ b/cdp/client/models/ethereum_token_transfer.py @@ -0,0 +1,100 @@ +# coding: utf-8 + +""" + Coinbase Platform API + + This is the OpenAPI 3.0 specification for the Coinbase Platform APIs, used in conjunction with the Coinbase Platform SDKs. + + The version of the OpenAPI document: 0.0.1-alpha + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + + +from __future__ import annotations +import pprint +import re # noqa: F401 +import json + +from pydantic import BaseModel, ConfigDict, Field, StrictInt, StrictStr +from typing import Any, ClassVar, Dict, List, Optional +from cdp.client.models.token_transfer_type import TokenTransferType +from typing import Optional, Set +from typing_extensions import Self + +class EthereumTokenTransfer(BaseModel): + """ + EthereumTokenTransfer + """ # noqa: E501 + contract_address: StrictStr + from_address: StrictStr + to_address: StrictStr + value: Optional[StrictStr] = Field(default=None, description="The value of the transaction in atomic units of the token being transfer for ERC20 or ERC1155 contracts.") + token_id: Optional[StrictStr] = Field(default=None, description="The ID of ERC721 or ERC1155 token being transferred.") + log_index: StrictInt + token_transfer_type: TokenTransferType + __properties: ClassVar[List[str]] = ["contract_address", "from_address", "to_address", "value", "token_id", "log_index", "token_transfer_type"] + + model_config = ConfigDict( + populate_by_name=True, + validate_assignment=True, + protected_namespaces=(), + ) + + + def to_str(self) -> str: + """Returns the string representation of the model using alias""" + return pprint.pformat(self.model_dump(by_alias=True)) + + def to_json(self) -> str: + """Returns the JSON representation of the model using alias""" + # TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead + return json.dumps(self.to_dict()) + + @classmethod + def from_json(cls, json_str: str) -> Optional[Self]: + """Create an instance of EthereumTokenTransfer from a JSON string""" + return cls.from_dict(json.loads(json_str)) + + def to_dict(self) -> Dict[str, Any]: + """Return the dictionary representation of the model using alias. + + This has the following differences from calling pydantic's + `self.model_dump(by_alias=True)`: + + * `None` is only added to the output dict for nullable fields that + were set at model initialization. Other fields with value `None` + are ignored. + """ + excluded_fields: Set[str] = set([ + ]) + + _dict = self.model_dump( + by_alias=True, + exclude=excluded_fields, + exclude_none=True, + ) + return _dict + + @classmethod + def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: + """Create an instance of EthereumTokenTransfer from a dict""" + if obj is None: + return None + + if not isinstance(obj, dict): + return cls.model_validate(obj) + + _obj = cls.model_validate({ + "contract_address": obj.get("contract_address"), + "from_address": obj.get("from_address"), + "to_address": obj.get("to_address"), + "value": obj.get("value"), + "token_id": obj.get("token_id"), + "log_index": obj.get("log_index"), + "token_transfer_type": obj.get("token_transfer_type") + }) + return _obj + + diff --git a/cdp/client/models/ethereum_transaction.py b/cdp/client/models/ethereum_transaction.py index c81e8f8..4ce2d4b 100644 --- a/cdp/client/models/ethereum_transaction.py +++ b/cdp/client/models/ethereum_transaction.py @@ -20,6 +20,7 @@ from datetime import datetime from pydantic import BaseModel, ConfigDict, Field, StrictInt, StrictStr from typing import Any, ClassVar, Dict, List, Optional +from cdp.client.models.ethereum_token_transfer import EthereumTokenTransfer from cdp.client.models.ethereum_transaction_access_list import EthereumTransactionAccessList from cdp.client.models.ethereum_transaction_flattened_trace import EthereumTransactionFlattenedTrace from typing import Optional, Set @@ -43,10 +44,11 @@ class EthereumTransaction(BaseModel): max_priority_fee_per_gas: Optional[StrictInt] = Field(default=None, description="The max priority fee per gas as defined in EIP-1559. https://eips.ethereum.org/EIPS/eip-1559 for more details.") priority_fee_per_gas: Optional[StrictInt] = Field(default=None, description="The confirmed priority fee per gas as defined in EIP-1559. https://eips.ethereum.org/EIPS/eip-1559 for more details.") transaction_access_list: Optional[EthereumTransactionAccessList] = None + token_transfers: Optional[List[EthereumTokenTransfer]] = None flattened_traces: Optional[List[EthereumTransactionFlattenedTrace]] = None block_timestamp: Optional[datetime] = Field(default=None, description="The timestamp of the block in which the event was emitted") mint: Optional[StrictStr] = Field(default=None, description="This is for handling optimism rollup specific EIP-2718 transaction type field.") - __properties: ClassVar[List[str]] = ["from", "gas", "gas_price", "hash", "input", "nonce", "to", "index", "value", "type", "max_fee_per_gas", "max_priority_fee_per_gas", "priority_fee_per_gas", "transaction_access_list", "flattened_traces", "block_timestamp", "mint"] + __properties: ClassVar[List[str]] = ["from", "gas", "gas_price", "hash", "input", "nonce", "to", "index", "value", "type", "max_fee_per_gas", "max_priority_fee_per_gas", "priority_fee_per_gas", "transaction_access_list", "token_transfers", "flattened_traces", "block_timestamp", "mint"] model_config = ConfigDict( populate_by_name=True, @@ -90,6 +92,13 @@ def to_dict(self) -> Dict[str, Any]: # override the default output from pydantic by calling `to_dict()` of transaction_access_list if self.transaction_access_list: _dict['transaction_access_list'] = self.transaction_access_list.to_dict() + # override the default output from pydantic by calling `to_dict()` of each item in token_transfers (list) + _items = [] + if self.token_transfers: + for _item_token_transfers in self.token_transfers: + if _item_token_transfers: + _items.append(_item_token_transfers.to_dict()) + _dict['token_transfers'] = _items # override the default output from pydantic by calling `to_dict()` of each item in flattened_traces (list) _items = [] if self.flattened_traces: @@ -123,6 +132,7 @@ def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: "max_priority_fee_per_gas": obj.get("max_priority_fee_per_gas"), "priority_fee_per_gas": obj.get("priority_fee_per_gas"), "transaction_access_list": EthereumTransactionAccessList.from_dict(obj["transaction_access_list"]) if obj.get("transaction_access_list") is not None else None, + "token_transfers": [EthereumTokenTransfer.from_dict(_item) for _item in obj["token_transfers"]] if obj.get("token_transfers") is not None else None, "flattened_traces": [EthereumTransactionFlattenedTrace.from_dict(_item) for _item in obj["flattened_traces"]] if obj.get("flattened_traces") is not None else None, "block_timestamp": obj.get("block_timestamp"), "mint": obj.get("mint") diff --git a/cdp/client/models/onchain_name.py b/cdp/client/models/onchain_name.py new file mode 100644 index 0000000..08b5d4d --- /dev/null +++ b/cdp/client/models/onchain_name.py @@ -0,0 +1,109 @@ +# coding: utf-8 + +""" + Coinbase Platform API + + This is the OpenAPI 3.0 specification for the Coinbase Platform APIs, used in conjunction with the Coinbase Platform SDKs. + + The version of the OpenAPI document: 0.0.1-alpha + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + + +from __future__ import annotations +import pprint +import re # noqa: F401 +import json + +from pydantic import BaseModel, ConfigDict, Field, StrictStr +from typing import Any, ClassVar, Dict, List, Optional +from cdp.client.models.onchain_name_text_records_inner import OnchainNameTextRecordsInner +from typing import Optional, Set +from typing_extensions import Self + +class OnchainName(BaseModel): + """ + A representation of an onchain stored name from name systems i.e. ENS or Basenames + """ # noqa: E501 + token_id: StrictStr = Field(description="The ID for the NFT related to this name") + owner_address: StrictStr = Field(description="The onchain address of the owner of the name") + manager_address: StrictStr = Field(description="The onchain address of the manager of the name") + primary_address: Optional[StrictStr] = Field(default=None, description="The primary onchain address of the name") + domain: StrictStr = Field(description="The readable format for the name in complete form") + avatar: Optional[StrictStr] = Field(default=None, description="The visual representation attached to this name") + network_id: StrictStr = Field(description="The ID of the blockchain network") + text_records: Optional[List[OnchainNameTextRecordsInner]] = Field(default=None, description="The metadata attached to this name") + __properties: ClassVar[List[str]] = ["token_id", "owner_address", "manager_address", "primary_address", "domain", "avatar", "network_id", "text_records"] + + model_config = ConfigDict( + populate_by_name=True, + validate_assignment=True, + protected_namespaces=(), + ) + + + def to_str(self) -> str: + """Returns the string representation of the model using alias""" + return pprint.pformat(self.model_dump(by_alias=True)) + + def to_json(self) -> str: + """Returns the JSON representation of the model using alias""" + # TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead + return json.dumps(self.to_dict()) + + @classmethod + def from_json(cls, json_str: str) -> Optional[Self]: + """Create an instance of OnchainName from a JSON string""" + return cls.from_dict(json.loads(json_str)) + + def to_dict(self) -> Dict[str, Any]: + """Return the dictionary representation of the model using alias. + + This has the following differences from calling pydantic's + `self.model_dump(by_alias=True)`: + + * `None` is only added to the output dict for nullable fields that + were set at model initialization. Other fields with value `None` + are ignored. + """ + excluded_fields: Set[str] = set([ + ]) + + _dict = self.model_dump( + by_alias=True, + exclude=excluded_fields, + exclude_none=True, + ) + # override the default output from pydantic by calling `to_dict()` of each item in text_records (list) + _items = [] + if self.text_records: + for _item_text_records in self.text_records: + if _item_text_records: + _items.append(_item_text_records.to_dict()) + _dict['text_records'] = _items + return _dict + + @classmethod + def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: + """Create an instance of OnchainName from a dict""" + if obj is None: + return None + + if not isinstance(obj, dict): + return cls.model_validate(obj) + + _obj = cls.model_validate({ + "token_id": obj.get("token_id"), + "owner_address": obj.get("owner_address"), + "manager_address": obj.get("manager_address"), + "primary_address": obj.get("primary_address"), + "domain": obj.get("domain"), + "avatar": obj.get("avatar"), + "network_id": obj.get("network_id"), + "text_records": [OnchainNameTextRecordsInner.from_dict(_item) for _item in obj["text_records"]] if obj.get("text_records") is not None else None + }) + return _obj + + diff --git a/cdp/client/models/onchain_name_list.py b/cdp/client/models/onchain_name_list.py new file mode 100644 index 0000000..8d1629b --- /dev/null +++ b/cdp/client/models/onchain_name_list.py @@ -0,0 +1,101 @@ +# coding: utf-8 + +""" + Coinbase Platform API + + This is the OpenAPI 3.0 specification for the Coinbase Platform APIs, used in conjunction with the Coinbase Platform SDKs. + + The version of the OpenAPI document: 0.0.1-alpha + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + + +from __future__ import annotations +import pprint +import re # noqa: F401 +import json + +from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictInt, StrictStr +from typing import Any, ClassVar, Dict, List, Optional +from cdp.client.models.onchain_name import OnchainName +from typing import Optional, Set +from typing_extensions import Self + +class OnchainNameList(BaseModel): + """ + A list of onchain events with pagination information + """ # noqa: E501 + data: List[OnchainName] = Field(description="A list of onchain name objects") + has_more: Optional[StrictBool] = Field(default=None, description="True if this list has another page of items after this one that can be fetched.") + next_page: StrictStr = Field(description="The page token to be used to fetch the next page.") + total_count: Optional[StrictInt] = Field(default=None, description="The total number of payload signatures for the address.") + __properties: ClassVar[List[str]] = ["data", "has_more", "next_page", "total_count"] + + model_config = ConfigDict( + populate_by_name=True, + validate_assignment=True, + protected_namespaces=(), + ) + + + def to_str(self) -> str: + """Returns the string representation of the model using alias""" + return pprint.pformat(self.model_dump(by_alias=True)) + + def to_json(self) -> str: + """Returns the JSON representation of the model using alias""" + # TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead + return json.dumps(self.to_dict()) + + @classmethod + def from_json(cls, json_str: str) -> Optional[Self]: + """Create an instance of OnchainNameList from a JSON string""" + return cls.from_dict(json.loads(json_str)) + + def to_dict(self) -> Dict[str, Any]: + """Return the dictionary representation of the model using alias. + + This has the following differences from calling pydantic's + `self.model_dump(by_alias=True)`: + + * `None` is only added to the output dict for nullable fields that + were set at model initialization. Other fields with value `None` + are ignored. + """ + excluded_fields: Set[str] = set([ + ]) + + _dict = self.model_dump( + by_alias=True, + exclude=excluded_fields, + exclude_none=True, + ) + # override the default output from pydantic by calling `to_dict()` of each item in data (list) + _items = [] + if self.data: + for _item_data in self.data: + if _item_data: + _items.append(_item_data.to_dict()) + _dict['data'] = _items + return _dict + + @classmethod + def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: + """Create an instance of OnchainNameList from a dict""" + if obj is None: + return None + + if not isinstance(obj, dict): + return cls.model_validate(obj) + + _obj = cls.model_validate({ + "data": [OnchainName.from_dict(_item) for _item in obj["data"]] if obj.get("data") is not None else None, + "has_more": obj.get("has_more"), + "next_page": obj.get("next_page"), + "total_count": obj.get("total_count") + }) + return _obj + + diff --git a/cdp/client/models/onchain_name_text_records_inner.py b/cdp/client/models/onchain_name_text_records_inner.py new file mode 100644 index 0000000..e7779c2 --- /dev/null +++ b/cdp/client/models/onchain_name_text_records_inner.py @@ -0,0 +1,89 @@ +# coding: utf-8 + +""" + Coinbase Platform API + + This is the OpenAPI 3.0 specification for the Coinbase Platform APIs, used in conjunction with the Coinbase Platform SDKs. + + The version of the OpenAPI document: 0.0.1-alpha + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + + +from __future__ import annotations +import pprint +import re # noqa: F401 +import json + +from pydantic import BaseModel, ConfigDict, Field, StrictStr +from typing import Any, ClassVar, Dict, List, Optional +from typing import Optional, Set +from typing_extensions import Self + +class OnchainNameTextRecordsInner(BaseModel): + """ + OnchainNameTextRecordsInner + """ # noqa: E501 + key: Optional[StrictStr] = Field(default=None, description="The key for the text record") + value: Optional[StrictStr] = Field(default=None, description="The value for the text record") + __properties: ClassVar[List[str]] = ["key", "value"] + + model_config = ConfigDict( + populate_by_name=True, + validate_assignment=True, + protected_namespaces=(), + ) + + + def to_str(self) -> str: + """Returns the string representation of the model using alias""" + return pprint.pformat(self.model_dump(by_alias=True)) + + def to_json(self) -> str: + """Returns the JSON representation of the model using alias""" + # TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead + return json.dumps(self.to_dict()) + + @classmethod + def from_json(cls, json_str: str) -> Optional[Self]: + """Create an instance of OnchainNameTextRecordsInner from a JSON string""" + return cls.from_dict(json.loads(json_str)) + + def to_dict(self) -> Dict[str, Any]: + """Return the dictionary representation of the model using alias. + + This has the following differences from calling pydantic's + `self.model_dump(by_alias=True)`: + + * `None` is only added to the output dict for nullable fields that + were set at model initialization. Other fields with value `None` + are ignored. + """ + excluded_fields: Set[str] = set([ + ]) + + _dict = self.model_dump( + by_alias=True, + exclude=excluded_fields, + exclude_none=True, + ) + return _dict + + @classmethod + def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: + """Create an instance of OnchainNameTextRecordsInner from a dict""" + if obj is None: + return None + + if not isinstance(obj, dict): + return cls.model_validate(obj) + + _obj = cls.model_validate({ + "key": obj.get("key"), + "value": obj.get("value") + }) + return _obj + + diff --git a/cdp/client/models/read_smart_contract_request.py b/cdp/client/models/read_smart_contract_request.py new file mode 100644 index 0000000..a644db3 --- /dev/null +++ b/cdp/client/models/read_smart_contract_request.py @@ -0,0 +1,91 @@ +# coding: utf-8 + +""" + Coinbase Platform API + + This is the OpenAPI 3.0 specification for the Coinbase Platform APIs, used in conjunction with the Coinbase Platform SDKs. + + The version of the OpenAPI document: 0.0.1-alpha + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + + +from __future__ import annotations +import pprint +import re # noqa: F401 +import json + +from pydantic import BaseModel, ConfigDict, Field, StrictStr +from typing import Any, ClassVar, Dict, List, Optional +from typing import Optional, Set +from typing_extensions import Self + +class ReadSmartContractRequest(BaseModel): + """ + ReadSmartContractRequest + """ # noqa: E501 + method: StrictStr = Field(description="The name of the contract method to call") + args: List[StrictStr] = Field(description="The arguments to pass to the contract method") + abi: Optional[StrictStr] = Field(default=None, description="The JSON-encoded ABI of the contract method (optional, will use cached ABI if not provided)") + __properties: ClassVar[List[str]] = ["method", "args", "abi"] + + model_config = ConfigDict( + populate_by_name=True, + validate_assignment=True, + protected_namespaces=(), + ) + + + def to_str(self) -> str: + """Returns the string representation of the model using alias""" + return pprint.pformat(self.model_dump(by_alias=True)) + + def to_json(self) -> str: + """Returns the JSON representation of the model using alias""" + # TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead + return json.dumps(self.to_dict()) + + @classmethod + def from_json(cls, json_str: str) -> Optional[Self]: + """Create an instance of ReadSmartContractRequest from a JSON string""" + return cls.from_dict(json.loads(json_str)) + + def to_dict(self) -> Dict[str, Any]: + """Return the dictionary representation of the model using alias. + + This has the following differences from calling pydantic's + `self.model_dump(by_alias=True)`: + + * `None` is only added to the output dict for nullable fields that + were set at model initialization. Other fields with value `None` + are ignored. + """ + excluded_fields: Set[str] = set([ + ]) + + _dict = self.model_dump( + by_alias=True, + exclude=excluded_fields, + exclude_none=True, + ) + return _dict + + @classmethod + def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: + """Create an instance of ReadSmartContractRequest from a dict""" + if obj is None: + return None + + if not isinstance(obj, dict): + return cls.model_validate(obj) + + _obj = cls.model_validate({ + "method": obj.get("method"), + "args": obj.get("args"), + "abi": obj.get("abi") + }) + return _obj + + diff --git a/cdp/client/models/token_transfer_type.py b/cdp/client/models/token_transfer_type.py new file mode 100644 index 0000000..3a86948 --- /dev/null +++ b/cdp/client/models/token_transfer_type.py @@ -0,0 +1,39 @@ +# coding: utf-8 + +""" + Coinbase Platform API + + This is the OpenAPI 3.0 specification for the Coinbase Platform APIs, used in conjunction with the Coinbase Platform SDKs. + + The version of the OpenAPI document: 0.0.1-alpha + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + + +from __future__ import annotations +import json +from enum import Enum +from typing_extensions import Self + + +class TokenTransferType(str, Enum): + """ + The type of the token transfer. + """ + + """ + allowed enum values + """ + ERC20 = 'erc20' + ERC721 = 'erc721' + ERC1155 = 'erc1155' + UNKNOWN = 'unknown' + + @classmethod + def from_json(cls, json_str: str) -> Self: + """Create an instance of TokenTransferType from a JSON string""" + return cls(json.loads(json_str)) + + diff --git a/cdp/wallet.py b/cdp/wallet.py index 4f37ba1..7624ef2 100644 --- a/cdp/wallet.py +++ b/cdp/wallet.py @@ -24,6 +24,7 @@ CreateWalletRequest, CreateWalletRequestWallet, ) +from cdp.client.models.create_wallet_webhook_request import CreateWalletWebhookRequest from cdp.client.models.wallet import Wallet as WalletModel from cdp.client.models.wallet_list import WalletList from cdp.contract_invocation import ContractInvocation @@ -33,6 +34,7 @@ from cdp.trade import Trade from cdp.wallet_address import WalletAddress from cdp.wallet_data import WalletData +from cdp.webhook import Webhook class Wallet: @@ -286,6 +288,26 @@ def create_address(self) -> "WalletAddress": return wallet_address + def create_webhook(self, notification_uri: str) -> "Webhook": + """Create a new webhook for the wallet. + + Args: + notification_uri (str): The notification URI of the webhook. + + Returns: + Webhook: The created webhook object. It can be used to monitor activities happening in the wallet. When they occur, webhook will make a request to the specified URI. + + Raises: + Exception: If there's an error creating the webhook. + + """ + create_wallet_webhook_request = CreateWalletWebhookRequest(notification_uri = notification_uri) + model = Cdp.api_clients.webhooks.create_wallet_webhook( + wallet_id=self.id, create_wallet_webhook_request=create_wallet_webhook_request + ) + + return Webhook(model) + def faucet(self, asset_id: str | None = None) -> FaucetTransaction: """Request faucet funds. diff --git a/cdp/webhook.py b/cdp/webhook.py new file mode 100644 index 0000000..7640490 --- /dev/null +++ b/cdp/webhook.py @@ -0,0 +1,201 @@ +from collections.abc import Iterator + +from cdp.cdp import Cdp +from cdp.client.models.create_webhook_request import CreateWebhookRequest +from cdp.client.models.update_webhook_request import UpdateWebhookRequest +from cdp.client.models.webhook import Webhook as WebhookModel +from cdp.client.models.webhook import WebhookEventFilter, WebhookEventType, WebhookEventTypeFilter +from cdp.client.models.webhook_list import WebhookList + + +class Webhook: + """A class representing a webhook.""" + + def __init__(self, model: WebhookModel) -> None: + """Initialize the Webhook class. + + Args: + model (WebhookModel): The WebhookModel object representing the Webhook. + + """ + self._model = model + + @property + def id(self) -> str: + """Get the ID of the webhook. + + Returns: + str: The ID of the webhook. + + """ + return self._model.id + + @property + def network_id(self) -> str: + """Get the network ID of the webhook. + + Returns: + str: The network ID of the webhook. + + """ + return self._model.network_id + + @property + def notification_uri(self) -> str: + """Get the notification URI of the webhook. + + Returns: + str: The notification URI of the webhook. + + """ + return self._model.notification_uri + + @property + def event_type(self) -> WebhookEventType: + """Get the event type of the webhook. + + Returns: + str: The event type of the webhook. + + """ + return self._model.event_type + + @property + def event_type_filter(self) -> WebhookEventTypeFilter: + """Get the event type filter of the webhook. + + Returns: + str: The event type filter of the webhook. + + """ + return self._model.event_type_filter + + @property + def event_filters(self) -> list[WebhookEventFilter]: + """Get the event filters of the webhook. + + Returns: + str: The event filters of the webhook. + + """ + return self._model.event_filters + + @classmethod + def create( + cls, + notification_uri: str, + event_type: WebhookEventType, + event_type_filter: WebhookEventTypeFilter | None = None, + event_filters: list[WebhookEventFilter] | None = None, + network_id: str = "base-sepolia", + ) -> "Webhook": + """Create a new webhook. + + Args: + notification_uri (str): The URI where notifications should be sent. + event_type (WebhookEventType): The type of event that the webhook listens to. + event_type_filter (WebhookEventTypeFilter): Filter specifically for wallet activity event type. + event_filters (List[WebhookEventTypeFilter]): Filters applied to the events that determine which specific address(es) trigger. + network_id (str): The network ID of the wallet. Defaults to "base-sepolia". + + Returns: + Webhook: The created webhook object. + + """ + create_webhook_request = CreateWebhookRequest( + network_id=network_id, + event_type=event_type, + event_type_filter=event_type_filter, + event_filters=event_filters, + notification_uri=notification_uri, + ) + + model = Cdp.api_clients.webhooks.create_webhook(create_webhook_request) + webhook = cls(model) + + return webhook + + @classmethod + def list(cls) -> Iterator["Webhook"]: + """List webhooks. + + Returns: + Iterator[Webhook]: An iterator of webhook objects. + + """ + while True: + page = None + + response: WebhookList = Cdp.api_clients.webhooks.list_webhooks(limit=100, page=page) + + for webhook_model in response.data: + yield cls(webhook_model) + + if not response.has_more: + break + + page = response.next_page + + @staticmethod + def delete(webhook_id: str) -> None: + """Delete a webhook by its ID. + + Args: + webhook_id (str): The ID of the webhook to delete. + + """ + Cdp.api_clients.webhooks.delete_webhook(webhook_id) + + def update( + self, + notification_uri: str | None = None, + event_type_filter: WebhookEventTypeFilter | None = None + ) -> "Webhook": + """Update the webhook with a new notification URI, and/or a new list of addresses to monitor. + + Args: + notification_uri (str): The new URI for webhook notifications. + event_type_filter (WebhookEventTypeFilter): The new eventTypeFilter that contains a new list (replacement) of addresses to monitor for the webhook. + + Returns: + Webhook: The updated webhook object. + + """ + # Fallback to current properties if no new values are provided + final_notification_uri = notification_uri or self.notification_uri + final_event_type_filter = event_type_filter or self.event_type_filter + + update_webhook_request = UpdateWebhookRequest( + event_type_filter=final_event_type_filter, + event_filters=self.event_filters, + notification_uri=final_notification_uri, + ) + + # Update the webhook via the API client + result = Cdp.api_clients.webhooks.update_webhook( + self.id, + update_webhook_request, + ) + + # Update the internal model with the API response + self._model = result + + return self + + def __str__(self) -> str: + """Return a string representation of the Webhook object. + + Returns: + str: A string representation of the Webhook. + + """ + return f"Webhook: (id: {self.id}, network_id: {self.network_id}, notification_uri: {self.notification_uri}, event_type: {self.event_type}, event_type_filter: {self.event_type_filter}, event_filters: {self.event_filters})" + + def __repr__(self) -> str: + """Return a detailed string representation of the Webhook object. + + Returns: + str: A string that represents the Webhook object. + + """ + return str(self) diff --git a/docs/conf.py b/docs/conf.py index 59a0c90..f6d927c 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -14,7 +14,7 @@ project = 'CDP SDK' author = 'Coinbase Developer Platform' -release = '0.0.6' +release = '0.0.7' # -- General configuration --------------------------------------------------- # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration diff --git a/pyproject.toml b/pyproject.toml index 32bcc20..4445cc8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "cdp-sdk" -version = "0.0.6" +version = "0.0.7" description = "CDP Python SDK" readme = "README.md" authors = [{name = "John Peterson", email = "john.peterson@coinbase.com"}] diff --git a/tests/factories/webhook_factory.py b/tests/factories/webhook_factory.py new file mode 100644 index 0000000..8c75150 --- /dev/null +++ b/tests/factories/webhook_factory.py @@ -0,0 +1,27 @@ +import pytest + +from cdp.webhook import Webhook, WebhookEventType, WebhookModel + + +@pytest.fixture +def webhook_factory(): + """Create and return a factory for Webhook fixtures.""" + def _create_webhook( + webhook_id="webhook-123", + network_id="base-sepolia", + notification_uri="https://example.com/webhook", + event_type=WebhookEventType.WALLET_ACTIVITY, + event_type_filter=None, + event_filters=None + ): + model = WebhookModel( + id=webhook_id, + network_id=network_id, + notification_uri=notification_uri, + event_type=event_type, + event_type_filter=event_type_filter, + event_filters=event_filters or [] + ) + return Webhook(model) + + return _create_webhook diff --git a/tests/test_wallet.py b/tests/test_wallet.py index 41ff3a3..c783ac3 100644 --- a/tests/test_wallet.py +++ b/tests/test_wallet.py @@ -7,6 +7,7 @@ from cdp.client.models.create_address_request import CreateAddressRequest from cdp.client.models.create_wallet_request import CreateWalletRequest, CreateWalletRequestWallet +from cdp.client.models.create_wallet_webhook_request import CreateWalletWebhookRequest from cdp.contract_invocation import ContractInvocation from cdp.payload_signature import PayloadSignature from cdp.smart_contract import SmartContract @@ -14,6 +15,7 @@ from cdp.transfer import Transfer from cdp.wallet import Wallet from cdp.wallet_address import WalletAddress +from cdp.webhook import Webhook @patch("cdp.Cdp.use_server_signer", False) @@ -614,3 +616,32 @@ def test_wallet_deploy_multi_token_with_server_signer(wallet_factory): mock_default_address.deploy_multi_token.assert_called_once_with( "https://example.com/multi-token/{id}.json" ) + +@patch("cdp.Cdp.api_clients") +def test_create_webhook(mock_api_clients, wallet_factory, webhook_factory): + """Test Wallet create_webhook method.""" + mock_api_clients.webhooks.create_wallet_webhook.return_value = webhook_factory() + + # Create a wallet instance using the factory + wallet = wallet_factory() + + # Define the notification URI to pass into the create_webhook method + notification_uri = "https://example.com/webhook" + + # Call the create_webhook method + webhook = wallet.create_webhook(notification_uri) + + # Create the expected request object + expected_request = CreateWalletWebhookRequest(notification_uri=notification_uri) + + # Assert that the API client was called with the correct parameters + mock_api_clients.webhooks.create_wallet_webhook.assert_called_once_with( + wallet_id=wallet.id, + create_wallet_webhook_request=expected_request + ) + + # Assert that the returned webhook is an instance of Webhook + assert isinstance(webhook, Webhook) + + # Additional assertions to check the returned webhook object + assert webhook.notification_uri == notification_uri diff --git a/tests/test_webhook.py b/tests/test_webhook.py new file mode 100644 index 0000000..4bf5272 --- /dev/null +++ b/tests/test_webhook.py @@ -0,0 +1,97 @@ +from unittest.mock import patch + +from cdp.client.models.create_webhook_request import CreateWebhookRequest +from cdp.client.models.update_webhook_request import UpdateWebhookRequest +from cdp.client.models.webhook import WebhookEventFilter, WebhookEventType, WebhookEventTypeFilter +from cdp.webhook import Webhook, WebhookModel + + +@patch("cdp.Cdp.api_clients") +def test_webhook_creation(mock_api_clients, webhook_factory): + """Test Webhook creation method.""" + mock_api_clients.webhooks.create_webhook.return_value = webhook_factory() + + # Define input parameters for the webhook creation + notification_uri = "https://example.com/webhook" + event_type = WebhookEventType.WALLET_ACTIVITY + event_type_filter = WebhookEventTypeFilter() + event_filters = [WebhookEventFilter()] + + expected_request = CreateWebhookRequest( + network_id="base-sepolia", + event_type=event_type, + event_type_filter=event_type_filter, + event_filters=event_filters, + notification_uri=notification_uri + ) + + webhook = Webhook.create( + notification_uri=notification_uri, + event_type=event_type, + event_type_filter=event_type_filter, + event_filters=event_filters, + network_id="base-sepolia" + ) + + mock_api_clients.webhooks.create_webhook.assert_called_once_with(expected_request) + + # Check that the returned object is a Webhook instance + assert isinstance(webhook, Webhook) + assert webhook.notification_uri == notification_uri + assert webhook.event_type == event_type + + +@patch("cdp.Cdp.api_clients") +def test_webhook_delete(mock_api_clients): + """Test Webhook delete method.""" + webhook_id = "webhook-123" + + Webhook.delete(webhook_id) + + mock_api_clients.webhooks.delete_webhook.assert_called_once_with(webhook_id) + + +@patch("cdp.Cdp.api_clients") +def test_webhook_update(mock_api_clients, webhook_factory): + """Test Webhook update method.""" + webhook_model = webhook_factory() + + # Create a Webhook instance + webhook = Webhook(model=webhook_model) + + assert webhook.notification_uri == "https://example.com/webhook" + + # Define new values for the update + new_notification_uri = "https://new.example.com/webhook" + + # Mock the API response for update + mock_api_clients.webhooks.update_webhook.return_value = WebhookModel( + id=webhook.id, + network_id=webhook.network_id, + notification_uri=new_notification_uri, + event_type=webhook.event_type, + event_type_filter=webhook.event_type_filter, + event_filters=webhook.event_filters + ) + + expected_request = UpdateWebhookRequest( + event_type_filter=webhook.event_type_filter, + event_filters=webhook.event_filters, + notification_uri=new_notification_uri + ) + + updated_webhook_model = webhook.update( + notification_uri=new_notification_uri, + ) + + updated_webhook = Webhook(model=updated_webhook_model) + + # Verify that the API client was called with the correct arguments + mock_api_clients.webhooks.update_webhook.assert_called_once_with( + webhook.id, expected_request + ) + + # Assert that the returned object is the updated webhook + assert isinstance(updated_webhook, Webhook) + assert updated_webhook.notification_uri == new_notification_uri + assert updated_webhook.id == webhook.id