A simple Python library for creating web servers and APIs using sockets, supporting openapi and swagger.
Project description
socketwrench
A webserver based on socket.socket
with no dependencies whatsoever other socket (or a substitute you pass in ) and optional standard library dependencies which improve features. See dependencies for more info.
Provides a lightweight quickstart to make an API which supports OpenAPI, Swagger, and more.
NOTE:
this is not a production-ready web server. It is a learning tool and a lightweight way to make a simple API. While I attempted to reduce overhead in calls, I haven't taken time to thoroughly optimize, and I have not implemented any complex features to deal with security, performance, or scalability.
defaults to single-threaded, blocking, synchronous, and single-process
- (recommended) Use
thread=True
to enable using multiple worker threads, which will allow for multiple requests to be processed simultaneously. - (optional) Use
num_connection_threads
to set the number of threads when thread=True. Defaults to None, meaning no limit. - (NOT recommended) Use
run_in_background=True
to run the entire server's main loop in a thread. Just make sure you have something in the main thread to keep the program running.
Project Goals
Part of the goal of this project was to understand how web servers work and to make a simple web server that is easy to use and understand. As learning progressed, features were added, but the code became a bit more complex. To learn more about the basics of web servers and how to develop one from scratch, see learning.md or jump straight into the building blocks of source code in simplestsocketwrench.py => simplesocketwrench.py => socketwrench. If you would prefer to use this library, read on!
Quickstart
Install
pip install socketwrench
Serve a class
from socketwrench import serve, StaticFileHandler
class MyServer:
src = StaticFileHandler(Path(__file__).parent.parent.parent)
def hello(self):
return "world"
if __name__ == '__main__':
serve(MyServer, thread=True)
# OR
# m = MyServer()
# serve(m)
# OR
# serve("my_module.MyServer")
- Go to http://localhost:8080/hello and see the response "world".
- Now go to http://localhost:8080/swagger to see the Swagger UI.
- OR go to http://localhost:8080/api to see a custom api playground.
Features
OpenAPI & Swagger
- Go to http://localhost:8080/swagger (after running
serve
) to see the Swagger UI. - Go to http://localhost:8080/openapi.json (after running
serve
) to see the autogenerated OpenAPI spec which Swagger uses.
Autofilled Parameters
Any of the following parameter names or typehints will get autofilled with the corresponding request data:
available_types = {
"request": Request, # full request object, contains all the other components
"query": Query, # query string
"body": Body, # request body bytes
"headers": Headers, # request headers dict[str, str]
"route": Route, # route string (without query string)
"full_path": FullPath, # full path string (with query string)
"method": Method, # request method str (GET, POST, etc.)
"file": File, # file bytes (essentially the same as body)
"client_addr": ClientAddr, # client ip address string (also contains host and port attributes)
"socket": socket.socket, # the socket object for the client
}
Decorators
from socketwrench import route, methods, get, post, put, patch, delete, private
These decorators do not modify the functions they decorate, they simply tag
the function by adding attributes to the functions.
func.__dict__[key] = value
. This allows the setting function-specific preferences such as which methods to allow.
@tag
simply modifies the function's __dict__
to add the specified attributes.
@tag(do_not_serve=False, methods=["GET", "POST"], error_mode="traceback")
def my_function():
pass
The following decorators set the available_methods
attribute of the function to the specified methods and tells the server to override its default behavior for the function.
@methods("GET", "POST", "DELETE")
: equivalent to@tag(available_methods=["GET", "POST", "DELETE"])
@get
,@post
,@put
,@patch
,@delete
,@private
: self-explanatory
Route Decorator
@route("/a/{c}")
tells the server to use /a/{c} as the route for the function instead of using the function's name as it normally does. This also allows for capturing path parameters.
@get
@post
@route("/a/{c}", error_mode="traceback")
def a(self, b, c=5):
print(f"calling a with {b=}, {c=}")
return f"captured {b=}, {c=}"
Error Modes
"hide"
orErrorModes.HIDE
: returnsb"Internal Server Error"
in the response body when an error occurs.type
orErrorModes.TYPE
: returns the error type only in the response body when an error occurs."short"
orErrorModes.SHORT
: returns the python error message but no traceback in the response body when an error occurs."traceback"
orErrorModes.TRACEBACK
orErrorModes.LONG
orErrorModes.TB
: returns the full traceback in the response body when an error occurs.
To set the default error mode for all functions, use set_default_error_mode
.
from socketwrench import set_default_error_mode, ErrorModes
set_default_error_mode(ErrorModes.TRACEBACK) # equivalent to ErrorModes=ErorModes.TRACEBACK
favicon.ico
No need to use our favicon! pass a str | Path
.ico
filepath to favicon
argument to use your own favicon. Alternatively, tag @route('/favicon.ico')
on a function returning the path.
fallback handler
Add a custom function to handle any requests that don't match any other routes.
Dependencies
Default behavior is to use the standard library only. However, if you do not have the full standard library, socketwrench should still work. This is a work in progress as I am attempting to support micropython, circuitpython, etc. but I have not tested on these environments yet.
How it works:
socket.socket
is the only required dependency, BUT you can pass in a substitute socket object if you want.- The following standard library modules are used, BUT if import fails for any of them we fall back on dump fake versions I made in src/socketwrench/fake_imports which attempt to approximate the same functionality
import socket
import builtins # very niche case use for if a function is typehinted to accept a type, e.g. `def f(x: type):` and you pass in the type name via a string query e.g. `?x=int`
import inspect # used often for getting function signatures, autofilling parameters, etc., spoof version uses `__annotations__` and `__defaults__` of functions
from sys import argv # only used in commandline mode
from argparse import ArgumentParser # only used in commandline mode
from tempfile import TemporaryFile # only used if you attempt to return a folder using a StaticFileHandler
from zipfile import ZipFile # only used if you attempt to return a folder using a StaticFileHandler
from functools import wraps, partial # used regularly but easily replaced
import dataclasses # only used if your python function returns a dataclass which we try to coerce to json
from datetime import datetime # used for Last-Modified header of File responses
from pathlib import Path # used for file responses and static file serving, spoof version works okay
from json import dumps, loads # used for json responses, spoof version works okay
import logging # used for logging, spoof version works okay
from time import sleep # only used if pause_sleep > 0 or accept_sleep > 0, spoof version does not sleep at all
from threading import Event, Thread # only used if you `thread=True` in `serve` function (defaults to False)
from traceback import format_exception # only used if error_mode="traceback"
import importlib # only used if you pass a string into the serve module as the item to be served, e.g. in commandline mode
from sys import modules # only used if you pass a string into the serve module as the item to be served, e.g. in commandline mode
sample
from socketwrench import serve
import socket
class Sample:
def hello(self):
return "world"
if __name__ == '__main__':
serve(Sample, spoof_modules="all", thread=True, socket=socket, port=8123)
Planned Features
- Implement nesting / recursion to serve deeper routes and/or multiple classes
- support default navigation pages to help show links to available routes
- Enforce OpenAPI spec with better error responses
- Serve static folders
- Make a better playground for testing endpoints
- better preview of variadic routes
- Make Response subclass Exception so it can be simply be
raise
d from anywhere in the stack - Simple file browse and upload
- improve docs
- document variadic routes
- document autofilled parameters
- document decorators
- document error modes
- document static file serving
- document favicon
- document fallback handler
- document regexp / match routes
- Make a client-side python proxy object to make API requests from python
Environment Support
- Remove
|
typehints to allow for older python versions :/ (this makes me sad) - Remove standard library dependencies which microcontrollers may not have
- Allow passing in a socket object
- test and support different kinds of sockets and objects pretending to be sockets
- Test on ESP32 and other microcontrollers
- Test in browser-based python environments using pyodide
Other
- Ideas? Let me know!
Other Usage Modes
Serve a module
Using commandline, just specify a filepath or file import path to a module.
# my_module.py
def hello():
return "world"
python -m socketwrench my_module
NOTE: this mode is experimental and less tested than the other modes.
Serve a single function on all routes
from socketwrench import serve
def print_request(request):
s = "<h>You made the following request:</h><br/>"
s += f"<b>Method:</b> {request.method}<br/>"
s += f"<b>Route:</b> {request.path.route()}<br/>"
s += f"<b>Headers:</b><br/> {str(request.headers).replace('\n', '<br>')}<br/>"
s += f"<b>Query:</b> {request.path.query_args()}<br/>"
s += f"<b>Body:</b> {request.body}<br/>"
return s
if __name__ == '__main__':
serve(print_request)
(mostly) Full Feature Sample
import logging
from pathlib import Path
from socketwrench.tags import private, post, put, patch, delete, route, methods
logging.basicConfig(level=logging.DEBUG)
class Sample:
def hello(self):
"""A simple hello world function."""
return "world"
@methods("GET", "POST") # do to the label, this will be accessible by both GET and POST requests
def hello2(self, method):
"""A simple hello world function."""
return "world"
def _unserved(self):
"""This function will not be served."""
return "this will not be served"
@private
def unserved(self):
"""This function will not be served."""
return "this will not be served"
@post
def post(self, name):
"""This function will only be served by POST requests."""
return f"hello {name}"
@put
def put(self, name):
"""This function will only be served by PUT requests."""
return f"hello {name}"
@patch
def patch(self, name):
"""This function will only be served by PATCH requests."""
return f"hello {name}"
@delete
def delete(self, name):
"""This function will only be served by DELETE requests."""
return f"hello {name}"
def echo(self, *args, **kwargs):
"""Echos back any query or body parameters."""
if not args and not kwargs:
return
if args:
if len(args) == 1:
return args[0]
return args
elif kwargs:
return kwargs
return args, kwargs
def string(self) -> str:
"""Returns a string response."""
return "this is a string"
def html(self) -> str:
"""Returns an HTML response."""
return "<h1>hello world</h1><br><p>this is a paragraph</p>"
def json(self) -> dict:
"""Returns a JSON response."""
return {"x": 6, "y": 7}
def file(self) -> Path:
"""Returns sample.py as a file response."""
return Path(__file__)
def add(self, x: int, y: int):
"""Adds two numbers together."""
return x + y
def client_addr(self, client_addr):
"""Returns the client address."""
return client_addr
def headers(self, headers) -> dict:
"""Returns the request headers."""
return headers
def query(self, query, *args, **kwargs) -> str:
"""Returns the query string."""
return query
def body(self, body) -> bytes:
"""Returns the request body."""
return body
def method(self, method) -> str:
"""Returns the method."""
return method
def get_route(self, route) -> str:
"""Returns the route."""
return route
def request(self, request) -> dict:
"""Returns the request object."""
return request
def everything(self, request, client_addr, headers, query, body, method, route, full_path):
d = {
"request": request,
"client_addr": client_addr,
"headers": headers,
"query": query,
"body": body,
"method": method,
"route": route,
"full_path": full_path,
}
for k, v in d.items():
print(k, v)
return d
@route("/a/{c}", error_mode="traceback")
def a(self, b, c=5):
print(f"calling a with {b=}, {c=}")
return f"captured {b=}, {c=}"
if __name__ == '__main__':
from socketwrench import serve
s = Sample()
serve(s)
# OR
# serve(Sample)
# OR
# serve("socketwrench.samples.sample.Sample")
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for socketwrench-2.3.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 93a6f482beb521de5e73db44fbc2862936322ee52f3c3c82c0cd334327a3be6f |
|
MD5 | da2c5f5afe37eb370155a371d6fc74e8 |
|
BLAKE2b-256 | ffce474494f94e039ff52ea55101ef73e7d11004b72c161390d25d7ac6737435 |