forked from digitalocean/sample-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathserver.py
More file actions
86 lines (76 loc) · 3.45 KB
/
server.py
File metadata and controls
86 lines (76 loc) · 3.45 KB
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
import os
import http.server
import socketserver
from http import HTTPStatus
import uuid
import random
import json
import datetime
def generate_json():
data = {
"id": str(uuid.uuid4()),
"timestampStarted": (datetime.datetime.now() - datetime.timedelta(days=random.randint(0, 30))).isoformat() + "Z",
"timestampEnded": (datetime.datetime.now() - datetime.timedelta(days=random.randint(0, 30))).isoformat() + "Z",
# "metric": random.choice(["M_REACHABILITY", "M_LATENCY", "M_BANDWIDTH", "M_PACKET_LOSS", "M_TCP_RETRANSMISSIONS"]),
"metric": "M_REACHABILITY",
"causes": [
{
"id": str(uuid.uuid4()),
"type": random.choice(["ISP", "CDN", "CLOUD"]),
"sourceAsns": [
str(random.randint(1000, 9000))
],
"h3Indexes": [
str(uuid.uuid4())[:15]
],
#limit network_id to 202 - 210
"networkIds": [
str(random.randint(200, 210)),
str(random.randint(202, 210)),
str(random.randint(202, 210))
],
"causeAsn": str(random.randint(1000, 9000)),
"timestampStarted": (datetime.datetime.now() - datetime.timedelta(days=random.randint(0, 30))).isoformat() + "Z",
"measurementCount": str(random.randint(0, 50)),
"alertsMatched": [
{
"id": str(uuid.uuid4()),
"timestampStarted": (datetime.datetime.now() - datetime.timedelta(days=random.randint(0, 30))).isoformat() + "Z",
"timestampEnded": (datetime.datetime.now() - datetime.timedelta(days=random.randint(0, 30))).isoformat() + "Z",
"sourceAsn": str(random.randint(1000, 9000)),
"sourceGeospatialH3Index": str(uuid.uuid4())[:15],
#change network_id to 202-210
"destinationNetworkId": str(random.randint(202, 210)),
#"metric": random.choice(["M_REACHABILITY", "M_LATENCY", "M_BANDWIDTH", "M_PACKET_LOSS", "M_TCP_RETRANSMISSIONS"]),
"metric": random.choice(["M_REACHABILITY"]),
"value": random.uniform(0, 80),
#"valueUnit": random.choice(["ms", "bps", "pct"]),
"valueUnit": "pct",
#"valueBaseline": random.uniform(0, 1000),
"valueBaseline": "97",
"measurementCount": str(random.randint(0, 50))
}
]
}
]
}
return data
# Generate 10 additional records
records = [generate_json() for i in range(10)]
# Write records to a JSON
# print(records)
# print(json.dumps(records, indent = 3))
class Handler(http.server.SimpleHTTPRequestHandler):
def do_GET(self):
self.send_response(HTTPStatus.OK)
# Should http headers be here?
self.end_headers()
# msg = 'Hello! you requested %s' % (self.path)
msg = (json.dumps(records, indent = 3))
self.wfile.write(msg.encode())
r = requests.get('https://x8ki-letl-twmt.n7.xano.io/api:qTxuLGUC/my_first_endpoint')
self.wfile.write(r.encode())
port = int(os.getenv('PORT', 80))
print('Listening on port %s' % (port))
httpd = socketserver.TCPServer(('', port), Handler)
httpd.serve_forever()