1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
import requests
import urllib
from config import BACKEND_URL
import logging
import json
from typing import Tuple, Dict
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
def make_request(method: str, url: str, **kwargs) -> Tuple[int, Dict]:
try:
response = requests.request(method, url, **kwargs)
answer = response.json()
except Exception as e:
logger.debug(f"Got exception while making request: {e}")
return 500, {}
logger.debug(
f"Got response from backend: "
f"Status={response.status_code}; "
f"Text={response.text[:200]}..."
)
return response.status_code, answer
def post_request(url: str, **kwargs):
return make_request("post", url, **kwargs)
def put_request(url: str, **kwargs):
return make_request("put", url, **kwargs)
def get_request(url: str, **kwargs):
return make_request("get", url, **kwargs)
def register_user(tg_id: int, username: str, fullname: str) -> Tuple[int, Dict]:
logger.debug(f"Trying to register user with id={tg_id}; username={username}")
return post_request(f"{BACKEND_URL}/profiles/", data={
"tg_id": tg_id,
"username": username,
"fullname": fullname
})
def get_tasks():
logger.debug(f"Trying to retrieve all tasks")
return get_request(f"{BACKEND_URL}/tasks/")
def get_task(title: str) -> Tuple[int, Dict]:
logger.debug(f"Trying to retrieve task with title={title}")
return get_request(f"{BACKEND_URL}/api/get_task/" + urllib.parse.quote(title))
def save_state(last_state: int, tg_id: int, user_data: dict) -> Tuple[int, Dict]:
user_data_dumped = json.dumps(user_data)
logger.debug(f"Trying to save state for user with id={tg_id}; state={last_state}; user_data={user_data_dumped}")
return put_request(f"{BACKEND_URL}/api/state/update/{tg_id}/", data={
"last_state": last_state,
"tg_id": tg_id,
"user_data": user_data_dumped
})
def get_state(tg_id: int) -> Tuple[int, dict]:
logger.debug(f"Trying to get state for user with id={tg_id}")
return get_request(f"{BACKEND_URL}/api/state/get/{tg_id}/")
def create_attempt(tg_id: int, task_title: str, answer: str):
return post_request(f"{BACKEND_URL}/attempts/", data={
"profile": json.dumps({"tg_id": tg_id}),
"task": json.dumps({"title": task_title}),
"answer": answer
})
def get_attempts(tg_id: int=None, task_title: str=None):
data = {}
if tg_id is not None:
data["tg_id"] = tg_id
if task_title is not None:
data["task_title"] = task_title
return get_request(f"{BACKEND_URL}/api/attempts/", data=data)
|
