Skip to content

Commit

Permalink
SW-3741
Browse files Browse the repository at this point in the history
  • Loading branch information
nostalgiaCode committed Feb 12, 2025
1 parent c7dedac commit 460e711
Show file tree
Hide file tree
Showing 2 changed files with 124 additions and 7 deletions.
51 changes: 44 additions & 7 deletions whatsapp_api_client_python/API.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,8 @@
receiving,
sending,
serviceMethods,
webhooks
webhooks,
partner
)


Expand Down Expand Up @@ -78,18 +79,14 @@ def request(
url = url.replace("{{idInstance}}", self.idInstance)
url = url.replace("{{apiTokenInstance}}", self.apiTokenInstance)

headers = {
'User-Agent': 'GREEN-API_SDK_PY/1.0'
}

try:
if not files:
response = self.session.request(
method=method, url=url, json=payload, timeout=self.host_timeout, headers=headers
method=method, url=url, json=payload, timeout=self.host_timeout
)
else:
response = self.session.request(
method=method, url=url, data=payload, files=files, timeout=self.media_timeout, headers=headers
method=method, url=url, data=payload, files=files, timeout=self.media_timeout
)
except Exception as error:
error_message = f"Request was failed with error: {error}."
Expand Down Expand Up @@ -182,3 +179,43 @@ class GreenAPI(GreenApi):

class GreenAPIError(Exception):
pass

class GreenApiPartner(GreenApi):
def __init__(
self,
partnerToken: str,
email: str = None,
debug_mode: bool = False,
raise_errors: bool = False,
host: str = "https://api.green-api.com",
media: str = "https://media.green-api.com",
host_timeout: float = 180,
media_timeout: float = 10800
):

super().__init__(
idInstance="",
apiTokenInstance="",
debug_mode=debug_mode,
raise_errors=raise_errors,
host=host,
media=media,
host_timeout=host_timeout,
media_timeout=media_timeout
)

self.partnerToken = partnerToken
self.email = email
self.partner = partner.Partner(self)

def request(
self,
method: str,
url: str,
payload: Optional[dict] = None,
files: Optional[dict] = None
) -> GreenAPIResponse:

url = url.replace("{{partnerToken}}", self.partnerToken)

return super().request(method, url, payload, files)
80 changes: 80 additions & 0 deletions whatsapp_api_client_python/tools/partner.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
from typing import Dict, List, Optional, TYPE_CHECKING, Union

Check failure on line 1 in whatsapp_api_client_python/tools/partner.py

View workflow job for this annotation

GitHub Actions / 3.8

Ruff (F401)

whatsapp_api_client_python/tools/partner.py:1:20: F401 `typing.Dict` imported but unused

Check failure on line 1 in whatsapp_api_client_python/tools/partner.py

View workflow job for this annotation

GitHub Actions / 3.8

Ruff (F401)

whatsapp_api_client_python/tools/partner.py:1:26: F401 `typing.List` imported but unused

Check failure on line 1 in whatsapp_api_client_python/tools/partner.py

View workflow job for this annotation

GitHub Actions / 3.8

Ruff (F401)

whatsapp_api_client_python/tools/partner.py:1:57: F401 `typing.Union` imported but unused

Check failure on line 1 in whatsapp_api_client_python/tools/partner.py

View workflow job for this annotation

GitHub Actions / 3.10

Ruff (F401)

whatsapp_api_client_python/tools/partner.py:1:20: F401 `typing.Dict` imported but unused

Check failure on line 1 in whatsapp_api_client_python/tools/partner.py

View workflow job for this annotation

GitHub Actions / 3.10

Ruff (F401)

whatsapp_api_client_python/tools/partner.py:1:26: F401 `typing.List` imported but unused

Check failure on line 1 in whatsapp_api_client_python/tools/partner.py

View workflow job for this annotation

GitHub Actions / 3.10

Ruff (F401)

whatsapp_api_client_python/tools/partner.py:1:57: F401 `typing.Union` imported but unused

Check failure on line 1 in whatsapp_api_client_python/tools/partner.py

View workflow job for this annotation

GitHub Actions / 3.11

Ruff (F401)

whatsapp_api_client_python/tools/partner.py:1:20: F401 `typing.Dict` imported but unused

Check failure on line 1 in whatsapp_api_client_python/tools/partner.py

View workflow job for this annotation

GitHub Actions / 3.11

Ruff (F401)

whatsapp_api_client_python/tools/partner.py:1:26: F401 `typing.List` imported but unused

Check failure on line 1 in whatsapp_api_client_python/tools/partner.py

View workflow job for this annotation

GitHub Actions / 3.11

Ruff (F401)

whatsapp_api_client_python/tools/partner.py:1:57: F401 `typing.Union` imported but unused

from ..response import Response

if TYPE_CHECKING:
from ..API import GreenApiPartner

class Partner:
def __init__(self, api: "GreenApiPartner"):
self.api = api

def getInstances(
self
) -> Response:

request_body = self.__handle_parameters(locals())

return self.api.request(
"GET", (
"{{host}}/partner/"
"getInstances/{{partnerToken}}"
), request_body
) ### GOOD

def createInstance(
self,
name: Optional[str] = None,
webhookUrl: Optional[str] = None,
webhookUrlToken: Optional[str] = None,
delaySendMessagesMilliseconds: Optional[int] = None,
markIncomingMessagesReaded: Optional[str] = None,
markIncomingMessagesReadedOnReply: Optional[str] = None,
outgoingWebhook: Optional[str] = None,
outgoingMessageWebhook: Optional[str] = None,
outgoingAPIMessageWebhook: Optional[str] = None,
stateWebhook: Optional[str] = None,
incomingWebhook: Optional[str] = None,
deviceWebhook: Optional[str] = None,
keepOnlineStatus: Optional[str] = None,
pollMessageWebhook: Optional[str] = None,
incomingBlockWebhook: Optional[str] = None,
incomingCallWebhook: Optional[str] = None,
editedMessageWebhook: Optional[str] = None,
deletedMessageWebhook: Optional[str] = None
) -> Response:

request_body = self.__handle_parameters(locals())

return self.api.request(
"POST", (
"{{host}}/partner/"
"createInstance/{{partnerToken}}"
), request_body
) ### NOT FINISHED!!!!

def deleteInstanceAccount(
self,
idInstance: int
) -> Response:

request_body = self.__handle_parameters(locals())

return self.api.request(
"POST", (
"{{host}}/partner/"
"deleteInstanceAccount/{{partnerToken}}"
), request_body
) ###GOOD

@classmethod
def __handle_parameters(cls, parameters: dict) -> dict:
handled_parameters = parameters.copy()

handled_parameters.pop("self")

for key, value in parameters.items():
if value is None:
handled_parameters.pop(key)

return handled_parameters

0 comments on commit 460e711

Please sign in to comment.