MisaghM/HttpServer.py

## HttpServer.py
import sys
import time
from http import HTTPStatus
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
from functools import cached_property, partial


IP = "127.0.0.1"
PORT = 18018

SIMPLE_PAGE = """
<!DOCTYPE html>
<html lang="en">
    <head>
        <meta charset="UTF-8">
        <title>{title}</title>
    </head>
    <body>
        <h1>{header}</h1>
        <p>{par}</p>
    </body>
</html>
"""


class System:
    """
    Sample backend logic with a method that multiplies its input.
    """

    def __init__(self, factor: int):
        self.factor = factor

    def calc(self, num: int):
        return self.factor * num


class TokenBucket:
    """
    How many 'tokens' (requests) can happen per 'per' seconds.
    """

    def __init__(self, tokens: int, per: int):
        self.tokens = tokens
        self.per = per
        self.bucket = float(tokens)
        self.last_check = time.time()

    def handle(self, curr_time: float) -> bool:
        time_passed = curr_time - self.last_check
        self.last_check = curr_time
        self.bucket += time_passed * (self.tokens / self.per)

        if self.bucket > self.tokens:
            self.bucket = float(self.tokens)
        if self.bucket >= 1.0:
            self.bucket -= 1.0
            return True
        return False


class ReqHandler(BaseHTTPRequestHandler):
    rate_limiter = TokenBucket(2, 10)  # two requests per 10 seconds

    def __init__(self, system: System, *args, **kwargs):
        self.system = system
        super().__init__(*args, **kwargs)

    @cached_property
    def url(self):
        return urlparse(self.path)

    @cached_property
    def query(self):
        return parse_qs(self.url.query)

    def do_GET(self):
        if self.rate_limit():
            return
        if self.url.path == '/':
            self.handler_default()
            return
        if self.url.path == '/calc':
            self.handler_calc()
            return
        self.handler_404()

    def handler_default(self):
        self.send_response(HTTPStatus.OK)
        self.send_header('content-type', 'text/html')
        self.end_headers()
        page = SIMPLE_PAGE.format(
            title='Welcome',
            header=f'Status: {HTTPStatus.OK}',
            par='You can query like "/calc?num=20" to get the system\'s results.'
        )
        self.wfile.write(page.encode())

    def handler_calc(self):
        if not self.query:
            self.send_response(HTTPStatus.BAD_REQUEST)
            page = SIMPLE_PAGE.format(
                title='Bad Request',
                header=f'Status: {HTTPStatus.BAD_REQUEST}',
                par='No query given.'
            )
        else:
            self.send_response(HTTPStatus.OK)
            try:
                num = int(self.query['num'][0])
            except:
                num = 0
            res = self.system.calc(num)
            page = SIMPLE_PAGE.format(
                title='Result',
                header=res,
                par='This is the answer.'
            )
        self.send_header('content-type', 'text/html')
        self.end_headers()
        self.wfile.write(page.encode())

    def handler_404(self):
        self.send_response(HTTPStatus.NOT_FOUND)
        self.send_header('content-type', 'text/html')
        self.end_headers()
        page = SIMPLE_PAGE.format(
            title='Not Found',
            header=f'Status: {HTTPStatus.NOT_FOUND}',
            par='The requested page was not found.'
        )
        self.wfile.write(page.encode())

    def rate_limit(self):
        if self.rate_limiter.handle(time.time()):
            return False
        self.send_response(HTTPStatus.TOO_MANY_REQUESTS)
        self.send_header('content-type', 'text/html')
        self.end_headers()
        page = SIMPLE_PAGE.format(
            title='Too Many Requests',
            header=f'Status: {HTTPStatus.TOO_MANY_REQUESTS}',
            par='Please wait before using the website.'
        )
        self.wfile.write(page.encode())
        return True


def main(argv: list[str] = None):
    if argv is None:
        argv = sys.argv[1:]
    if not argv:
        (ip, port) = (IP, PORT)
    elif len(argv) > 1:
        (ip, port) = (argv[0], int(argv[1]))
    else:
        print('Enter both ip and port in the command-line arguments.')
        return

    system = System(factor=2)
    with HTTPServer((ip, port), partial(ReqHandler, system)) as server:
        print(f'Server running on {ip}:{port}')
        try:
            server.serve_forever()
        except KeyboardInterrupt:
            pass
    print('Server closed.')


if __name__ == '__main__':
    main()
	import sys
	import time
	from http import HTTPStatus
	from http.server import HTTPServer, BaseHTTPRequestHandler
	from urllib.parse import urlparse, parse_qs
	from functools import cached_property, partial


	IP = "127.0.0.1"
	PORT = 18018

	SIMPLE_PAGE = """
	<!DOCTYPE html>
	<html lang="en">
	<head>
	<meta charset="UTF-8">
	<title>{title}</title>
	</head>
	<body>
	<h1>{header}</h1>
	<p>{par}</p>
	</body>
	</html>
	"""


	class System:
	"""
	Sample backend logic with a method that multiplies its input.
	"""

	def __init__(self, factor: int):
	self.factor = factor

	def calc(self, num: int):
	return self.factor * num


	class TokenBucket:
	"""
	How many 'tokens' (requests) can happen per 'per' seconds.
	"""

	def __init__(self, tokens: int, per: int):
	self.tokens = tokens
	self.per = per
	self.bucket = float(tokens)
	self.last_check = time.time()

	def handle(self, curr_time: float) -> bool:
	time_passed = curr_time - self.last_check
	self.last_check = curr_time
	self.bucket += time_passed * (self.tokens / self.per)

	if self.bucket > self.tokens:
	self.bucket = float(self.tokens)
	if self.bucket >= 1.0:
	self.bucket -= 1.0
	return True
	return False


	class ReqHandler(BaseHTTPRequestHandler):
	rate_limiter = TokenBucket(2, 10) # two requests per 10 seconds

	def __init__(self, system: System, args, *kwargs):
	self.system = system
	super().__init__(args, *kwargs)

	@cached_property
	def url(self):
	return urlparse(self.path)

	@cached_property
	def query(self):
	return parse_qs(self.url.query)

	def do_GET(self):
	if self.rate_limit():
	return
	if self.url.path == '/':
	self.handler_default()
	return
	if self.url.path == '/calc':
	self.handler_calc()
	return
	self.handler_404()

	def handler_default(self):
	self.send_response(HTTPStatus.OK)
	self.send_header('content-type', 'text/html')
	self.end_headers()
	page = SIMPLE_PAGE.format(
	title='Welcome',
	header=f'Status: {HTTPStatus.OK}',
	par='You can query like "/calc?num=20" to get the system\'s results.'
	)
	self.wfile.write(page.encode())

	def handler_calc(self):
	if not self.query:
	self.send_response(HTTPStatus.BAD_REQUEST)
	page = SIMPLE_PAGE.format(
	title='Bad Request',
	header=f'Status: {HTTPStatus.BAD_REQUEST}',
	par='No query given.'
	)
	else:
	self.send_response(HTTPStatus.OK)
	try:
	num = int(self.query['num'][0])
	except:
	num = 0
	res = self.system.calc(num)
	page = SIMPLE_PAGE.format(
	title='Result',
	header=res,
	par='This is the answer.'
	)
	self.send_header('content-type', 'text/html')
	self.end_headers()
	self.wfile.write(page.encode())

	def handler_404(self):
	self.send_response(HTTPStatus.NOT_FOUND)
	self.send_header('content-type', 'text/html')
	self.end_headers()
	page = SIMPLE_PAGE.format(
	title='Not Found',
	header=f'Status: {HTTPStatus.NOT_FOUND}',
	par='The requested page was not found.'
	)
	self.wfile.write(page.encode())

	def rate_limit(self):
	if self.rate_limiter.handle(time.time()):
	return False
	self.send_response(HTTPStatus.TOO_MANY_REQUESTS)
	self.send_header('content-type', 'text/html')
	self.end_headers()
	page = SIMPLE_PAGE.format(
	title='Too Many Requests',
	header=f'Status: {HTTPStatus.TOO_MANY_REQUESTS}',
	par='Please wait before using the website.'
	)
	self.wfile.write(page.encode())
	return True


	def main(argv: list[str] = None):
	if argv is None:
	argv = sys.argv[1:]
	if not argv:
	(ip, port) = (IP, PORT)
	elif len(argv) > 1:
	(ip, port) = (argv[0], int(argv[1]))
	else:
	print('Enter both ip and port in the command-line arguments.')
	return

	system = System(factor=2)
	with HTTPServer((ip, port), partial(ReqHandler, system)) as server:
	print(f'Server running on {ip}:{port}')
	try:
	server.serve_forever()
	except KeyboardInterrupt:
	pass
	print('Server closed.')


	if __name__ == '__main__':
	main()