|
| 1 | +from typing import Any, Dict, Tuple |
1 | 2 | import requests |
2 | 3 | from datetime import datetime |
3 | | -from typing import Any, Dict, Tuple |
4 | | -from .config import ( |
5 | | - MONO_CURRENCY_URI, |
6 | | - MONO_CLIENT_INFO_URI, |
7 | | - MONO_STATEMENT_URI, |
8 | | - MONO_WEBHOOK_URI, |
9 | | - DAY_UTC, |
10 | | -) |
11 | 4 |
|
12 | 5 |
|
13 | 6 | class MonoManager: |
14 | 7 |
|
15 | 8 | def __init__(self, request): |
16 | 9 | self.request = request |
17 | | - |
| 10 | + |
18 | 11 | session = requests.Session() |
| 12 | + |
| 13 | + _currency_uri = 'https://api.monobank.ua/bank/currency' |
| 14 | + _client_info_uri = 'https://api.monobank.ua/personal/client-info' |
| 15 | + _statement_uri = 'https://api.monobank.ua/personal/statement/0/' |
| 16 | + _webhook_uri = 'https://api.monobank.ua/personal/webhook' |
| 17 | + _day_utc = 86400 # 1 day (UNIX) |
19 | 18 |
|
20 | 19 | @classmethod |
21 | | - @property |
22 | 20 | def get_currency(cls) -> Tuple[int, Dict[str, Any]]: |
23 | 21 | try: |
24 | | - _ = cls.session.get(MONO_CURRENCY_URI) |
25 | | - _.raise_for_status() |
26 | | - return _.json() |
| 22 | + response = cls.session.get(cls._currency_uri) |
| 23 | + response.raise_for_status() |
| 24 | + return response.status_code, response.json() |
| 25 | + except requests.exceptions.HTTPError as exc: |
| 26 | + error_response = { |
| 27 | + "detail": str(exc), |
| 28 | + "code": response.status_code, |
| 29 | + } |
| 30 | + return error_response |
27 | 31 | except Exception as exc: |
28 | | - raise exc |
| 32 | + return { |
| 33 | + "detail": str(exc) |
| 34 | + } |
29 | 35 |
|
30 | 36 | def get_client_info(self, token: str) -> Tuple[int, Dict[str, Any]]: |
31 | 37 | try: |
32 | 38 | headers = {"X-Token": token} |
33 | | - _ = self.session.get( |
34 | | - MONO_CLIENT_INFO_URI, headers=headers |
| 39 | + response = self.session.get( |
| 40 | + self._client_info_uri, |
| 41 | + headers=headers |
35 | 42 | ) |
36 | | - _.raise_for_status() |
37 | | - return _.json() |
| 43 | + response.raise_for_status() |
| 44 | + return response.status_code, response.json() |
| 45 | + except requests.exceptions.HTTPError as exc: |
| 46 | + error_response = { |
| 47 | + "detail": str(exc), |
| 48 | + "code": response.status_code, |
| 49 | + } |
| 50 | + return error_response |
38 | 51 | except Exception as exc: |
39 | | - raise exc |
| 52 | + return { |
| 53 | + "detail": str(exc) |
| 54 | + } |
40 | 55 |
|
41 | 56 | def get_balance(self, token: str) -> Tuple[int, Dict[str, Any]]: |
42 | 57 | try: |
43 | | - payload = self.get_client_info(self, token) |
| 58 | + headers = {"X-Token": token} |
| 59 | + response = self.session.get( |
| 60 | + self._client_info_uri, |
| 61 | + headers=headers |
| 62 | + ) |
| 63 | + response.raise_for_status() |
44 | 64 | balance = { |
45 | | - 'balance': payload["accounts"][0]["balance"] / 100 |
| 65 | + 'balance': response.json()["accounts"][0]["balance"] / 100 |
| 66 | + } |
| 67 | + return response.status_code, balance |
| 68 | + except requests.exceptions.HTTPError as exc: |
| 69 | + error_response = { |
| 70 | + "detail": str(exc), |
| 71 | + "code": response.status_code, |
46 | 72 | } |
47 | | - return balance |
| 73 | + return error_response |
48 | 74 | except Exception as exc: |
49 | | - raise exc |
50 | | - |
| 75 | + return { |
| 76 | + "detail": str(exc) |
| 77 | + } |
| 78 | + |
51 | 79 | def get_statement(self, token: str, period: int) -> Tuple[int, Dict[str, Any]]: |
52 | 80 | try: |
53 | | - time_delta = int(datetime.now().timestamp()) - (period * DAY_UTC) |
| 81 | + time_delta = int(datetime.now().timestamp()) - (period * self._day_utc) |
54 | 82 | headers = {"X-Token": token} |
55 | | - _ = self.session.get( |
56 | | - f"{MONO_STATEMENT_URI}{time_delta}/", |
| 83 | + response = self.session.get( |
| 84 | + f"{self._statement_uri}{time_delta}/", |
57 | 85 | headers=headers |
58 | 86 | ) |
59 | | - _.raise_for_status() |
60 | | - return _.json() |
| 87 | + response.raise_for_status() |
| 88 | + return response.status_code, response.json() |
| 89 | + except requests.exceptions.HTTPError as exc: |
| 90 | + error_response = { |
| 91 | + "detail": str(exc), |
| 92 | + "code": response.status_code, |
| 93 | + } |
| 94 | + return error_response |
61 | 95 | except Exception as exc: |
62 | | - raise exc |
| 96 | + return { |
| 97 | + "detail": str(exc) |
| 98 | + } |
63 | 99 |
|
64 | 100 | def create_webhook(self, token: str, webHookUrl: str) -> Tuple[int, Dict[str, Any]]: |
65 | 101 | try: |
66 | 102 | headers = {"X-Token": token} |
67 | | - _ = self.session.post( |
68 | | - MONO_WEBHOOK_URI, data=webHookUrl, headers=headers |
| 103 | + response = self.session.post( |
| 104 | + self._webhook_uri, |
| 105 | + data=webHookUrl, |
| 106 | + headers=headers |
69 | 107 | ) |
70 | | - _.raise_for_status() |
71 | | - return _.json() |
| 108 | + response.raise_for_status() |
| 109 | + return response.status_code, response.json() |
| 110 | + except requests.exceptions.HTTPError as exc: |
| 111 | + error_response = { |
| 112 | + "detail": str(exc), |
| 113 | + "code": response.status_code, |
| 114 | + } |
| 115 | + return error_response |
72 | 116 | except Exception as exc: |
73 | | - raise exc |
| 117 | + return { |
| 118 | + "detail": str(exc) |
| 119 | + } |
0 commit comments