forked from fybrik/hello-world-read-module
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathhello-world-module.py
136 lines (115 loc) · 4.51 KB
/
hello-world-module.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
131
132
133
134
135
136
#!/usr/bin/env python3
import os
import yaml
import urllib.request
import csv
import argparse
from http.server import HTTPServer, BaseHTTPRequestHandler
import logging
data_dict = {}
class S(BaseHTTPRequestHandler):
def _set_headers(self):
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
def readurl(self,url):
response = urllib.request.urlopen(url)
lines = [l.decode('utf-8') for l in response.readlines()]
cr = csv.reader(lines)
count = 0
for row in cr:
if count > 10:
break
count += 1
row_str = ", ".join(row)+'\n'
self.wfile.write(row_str.encode("utf8"))
def _html(self, message):
"""This just generates an HTML document that includes `message`
in the body. Override, or re-write this do do more interesting stuff.
"""
content = f"<html><body><h1>{message}</h1></body></html>"
return content.encode("utf8") # NOTE: must return a bytes object!
def do_GET(self):
self._set_headers()
logging.info("path is " + self.path)
data = data_dict.get(self.path[1:], "notfound")
#logging.info("url is " + url)
logging.info("GET request")
self.wfile.write("GET request for {}\n".format(self.path).encode('utf-8'))
if(data != "notfound"):
url = data['endpoint_url']
self.readurl(url)
logging.info("Transformation action is {} for the columns {}\n".format(data['action'],data['transferred_columns']))
else:
self.wfile.write("invalid request\n".encode('utf-8'))
self.wfile.write("The avialable datasets:\n".encode('utf-8'))
for key in data_dict:
self.wfile.write("dataset name: {}\n".format(key).encode('utf-8'))
for k in data_dict[key]:
self.wfile.write(" {}: {}\n".format(k,data_dict[key][k]).encode('utf-8'))
logging.info(data_dict)
def do_HEAD(self):
self._set_headers()
def do_POST(self):
# Doesn't do anything with posted data
self._set_headers()
self.wfile.write(self._html("POST!"))
logging.info("POST request\n")
def run(config_path=None, server_class=HTTPServer, handler_class=S, addr="localhost", port=8000):
logging.basicConfig(level=logging.INFO)
logging.info("\nHello World Read Module!")
with open(config_path, 'r') as stream:
content = yaml.safe_load(stream)
#logging.info(content)
for key,val in content.items():
if "data" in key:
for i in range(len(val)):
data = val[i]
connectionName = data["name"]
name = connectionName.split("/")[1]
format = data["format"]
endpoint_url = data["connection"]["s3"]["endpoint_url"]
transformations = data["transformations"][0]
action = transformations['action']
transferred_columns = transformations['columns']
data_dict[name] = {'format':format, 'endpoint_url':endpoint_url, 'action':action, 'transferred_columns':transferred_columns}
logging.info("The avialable datasets:\n")
for key in data_dict:
logging.info("dataset name: {}\n".format(key))
for k in data_dict[key]:
logging.info(" {}: {}\n".format(k,data_dict[key][k]))
#logging.info(data_dict)
server_address = (addr, port)
httpd = server_class(server_address, handler_class)
logging.info(f"Starting httpd server on {addr}:{port}")
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
httpd.server_close()
logging.info('Stopping httpd...\n')
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Run an HTTP server")
parser.add_argument(
"-l",
"--listen",
default="localhost",
help="Specify the IP address on which the server listens",
)
parser.add_argument(
"-p",
"--port",
type=int,
default=8000,
help="Specify the port on which the server listens",
)
parser.add_argument(
'-c',
'--config',
type=str,
default='../etc/conf/conf.yaml',
help='Path to config file'
)
args = parser.parse_args()
run(config_path=args.config, addr=args.listen, port=args.port)
#main()