-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathserver.py
130 lines (89 loc) · 2.64 KB
/
server.py
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
# -*- coding: utf-8 -*-
'''Just module docstring'''
import json
import random
from datetime import datetime
from functools import wraps
from uuid import uuid4
from flask import Flask, abort, request
class User:
'''Initialize user'''
def __init__(self, name, password):
'''Constructor'''
self.name = name
self.password = password
self.token = None
self.login_time = None
def login(self, name, password):
'''Generates temp token'''
if self.name == name and self.password == password:
self.token = str(uuid4())
self.login_time = datetime.now()
return True
return False
class Task:
'''Initialize Task'''
def __init__(self):
'''Constructor'''
self.state = "waiting"
self.log = ""
self.result = 0
def Run(self):
'''Change task state'''
random.seed()
self.state = random.choice(["ok", "notOk"])
if self.state == "ok":
self.result = random.randint(1, 10)
else:
self.log = "Critical issue"
ADMIN = User("admin", "admin")
TASK = Task()
app = Flask(__name__)
def require_api_token(f):
@wraps(f)
def decorated(*args, **kwargs):
if request.args.get('api_token') and request.args.get('api_token') == ADMIN.token:
if (datetime.now() - ADMIN.login_time).seconds < 300:
return f(*args, **kwargs)
return abort(403)
return decorated
@app.route('/apitoken', methods=['POST'])
def login():
r_user = request.args.get("user")
r_pass = request.args.get("password")
r_data = f"Wrong creds. {r_user}:{r_pass} is wrond user:pass combination"
if ADMIN.login(r_user, r_pass):
r_data = {"token": ADMIN.token}
return app.response_class(
response=json.dumps(r_data),
status=404,
mimetype='application/json'
)
@app.route("/task", methods=['GET'])
@require_api_token
def get_task_status():
if TASK.state == "waiting":
r_status = 404
r_data = "Not found"
elif TASK.state == "ok":
r_status = 200
r_data = f"RESULT:{TASK.result}"
else:
r_status = 500
r_data = f"ERROR:{TASK.log}"
return app.response_class(
response=json.dumps(r_data),
status=r_status,
mimetype='application/json'
)
@app.route("/task/start", methods=['POST'])
@require_api_token
def start_task():
TASK.Run()
return app.response_class(
response=json.dumps({"message": "Task has been started"}),
status=200,
mimetype='application/json'
)
if __name__ == "__main__":
app.run()