update: added 'HEAD' method support for HTTP Server
This commit is contained in:
+1
-1
@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
[project]
|
[project]
|
||||||
name = "common"
|
name = "common"
|
||||||
description = "Commonly reusable code"
|
description = "Commonly reusable code"
|
||||||
version = "0.1.18"
|
version = "0.1.19"
|
||||||
requires-python = ">=3.13"
|
requires-python = ">=3.13"
|
||||||
authors = [
|
authors = [
|
||||||
{ name="BreakerBear", email="breakerbear@autistic.men" },
|
{ name="BreakerBear", email="breakerbear@autistic.men" },
|
||||||
|
|||||||
@@ -101,6 +101,7 @@ class History(logging.Handler):
|
|||||||
class Server(ThreadingHTTPServer):
|
class Server(ThreadingHTTPServer):
|
||||||
def __init__(self, *args, **kwargs):
|
def __init__(self, *args, **kwargs):
|
||||||
super().__init__(*args, **kwargs)
|
super().__init__(*args, **kwargs)
|
||||||
|
self.application: str
|
||||||
self.encoding: str
|
self.encoding: str
|
||||||
self.handlers = dict()
|
self.handlers = dict()
|
||||||
|
|
||||||
@@ -132,6 +133,11 @@ class RequestHandler(BaseHTTPRequestHandler):
|
|||||||
self.end_headers()
|
self.end_headers()
|
||||||
self.wfile.write(response)
|
self.wfile.write(response)
|
||||||
|
|
||||||
|
def do_HEAD(self):
|
||||||
|
self.send_response(200)
|
||||||
|
self.send_header('X-Powered-By', app) if (app := self.server.application) else None
|
||||||
|
self.end_headers()
|
||||||
|
|
||||||
def do_POST(self, request=None):
|
def do_POST(self, request=None):
|
||||||
try:
|
try:
|
||||||
size = int(self.headers.get('Content-Length', '0'))
|
size = int(self.headers.get('Content-Length', '0'))
|
||||||
@@ -160,27 +166,26 @@ class ServiceProvider:
|
|||||||
def __init__(self, options):
|
def __init__(self, options):
|
||||||
self.options = dataclasses.asdict(options)
|
self.options = dataclasses.asdict(options)
|
||||||
self.server = Server((self.options['host'], self.options['port']), self.options['handler'])
|
self.server = Server((self.options['host'], self.options['port']), self.options['handler'])
|
||||||
|
self.server.application = self.options['application']
|
||||||
self.server.encoding = self.options['encoding']
|
self.server.encoding = self.options['encoding']
|
||||||
|
|
||||||
@classmethod
|
if not self.options['setup']: return
|
||||||
def default(cls):
|
|
||||||
import sys, _thread as t
|
import sys, _thread as t
|
||||||
logger = logging.getLogger()
|
logger = logging.getLogger()
|
||||||
history = History()
|
history = History()
|
||||||
logger.addHandler(history)
|
logger.addHandler(history)
|
||||||
opts = cls.Options()
|
|
||||||
self = cls(opts)
|
|
||||||
self.set('logs', lambda: history.truncate())
|
self.set('logs', lambda: history.truncate())
|
||||||
self.set('exit', lambda: (t.interrupt_main(), sys.exit(0)))
|
self.set('exit', lambda: (t.interrupt_main(), sys.exit(0)))
|
||||||
return self
|
|
||||||
|
|
||||||
@dataclasses.dataclass
|
@dataclasses.dataclass
|
||||||
class Options:
|
class Options:
|
||||||
host : str = '127.0.0.1'
|
host : str = '127.0.0.1'
|
||||||
port : int = 0
|
port : int = 0
|
||||||
|
setup : bool = True
|
||||||
handler : type = RequestHandler
|
handler : type = RequestHandler
|
||||||
encoding : str = 'UTF-8'
|
encoding : str = 'UTF-8'
|
||||||
interval : float = 0.2
|
interval : float = 0.2
|
||||||
|
application : str = ''
|
||||||
|
|
||||||
def add(self, *handlers: Callable[..., Any], pairs: Iterable[tuple[str, Callable[..., Any]]] = None):
|
def add(self, *handlers: Callable[..., Any], pairs: Iterable[tuple[str, Callable[..., Any]]] = None):
|
||||||
for handler in handlers: self.set(handler.__name__, handler)
|
for handler in handlers: self.set(handler.__name__, handler)
|
||||||
|
|||||||
Reference in New Issue
Block a user