31 Star 231 Fork 52

keijack / python-simple-http-server

Create your Gitee Account
Explore and code with more than 8 million developers,Free private repositories !:)
Sign up
Clone or Download
Sync branch
Notice: Creating folder will generate an empty file .keep, because not support in Git


PyPI version


This is a simple http server, use MVC like design.

Support Python Version

Python 3.7+

Why choose

  • Lightway.
  • Functional programing.
  • Filter chain support.
  • Session support, and can support distributed session by this extention.
  • Spring MVC like request mapping.
  • SSL support.
  • Websocket support
  • Easy to use.
  • Free style controller writing.
  • Easily integraded with WSGI servers.
  • Coroutine mode support.


There are no other dependencies needed to run this project. However, if you want to run the unitests in the tests folder, you need to install websocket via pip:

python3 -m pip install websocket-client

How to use


python3 -m pip install simple_http_server

Minimum code / component requirement setup

Minimum code to get things started should have at least one controller function,
using the route and server modules from simple_http_server

from simple_http_server import route, server
def index():
    return {"hello": "world"}   


Write Controllers

from simple_http_server import request_map
from simple_http_server import Response
from simple_http_server import MultipartFile
from simple_http_server import Parameter
from simple_http_server import Parameters
from simple_http_server import Header
from simple_http_server import JSONBody
from simple_http_server import HttpError
from simple_http_server import StaticFile
from simple_http_server import Headers
from simple_http_server import Cookies
from simple_http_server import Cookie
from simple_http_server import Redirect
from simple_http_server import ModelDict

# request_map has an alias name `route`, you can select the one you familiar with.
def my_ctrl():
    return {"code": 0, "message": "success"}  # You can return a dictionary, a string or a `simple_http_server.simple_http_server.Response` object.

@route("/say_hello", method=["GET", "POST"])
def my_ctrl2(name, name2=Parameter("name", default="KEIJACK"), model=ModelDict()):
    """name and name2 is the same"""
    name == name2 # True
    name == model["name"] # True
    return "<!DOCTYPE html><html><body>hello, %s, %s</body></html>" % (name, name2)

def my_ctrl3():
    return Response(status_code=500)

def exception_ctrl():
    raise HttpError(400, "Exception")

@request_map("/upload", method="GET")
def show_upload():
    root = os.path.dirname(os.path.abspath(__file__))
    return StaticFile("%s/my_dev/my_test_index.html" % root, "text/html; charset=utf-8")

@request_map("/upload", method="POST")
def my_upload(img=MultipartFile("img")):
    root = os.path.dirname(os.path.abspath(__file__))
    img.save_to_file(root + "/my_dev/imgs/" + img.filename)
    return "<!DOCTYPE html><html><body>upload ok!</body></html>"

@request_map("/post_txt", method="POST")
def normal_form_post(txt):
    return "<!DOCTYPE html><html><body>hi, %s</body></html>" % txt

def tuple_results():
    # The order here is not important, we consider the first `int` value as status code,
    # All `Headers` object will be sent to the response
    # And the first valid object whose type in (str, unicode, dict, StaticFile, bytes) will
    # be considered as the body
    return 200, Headers({"my-header": "headers"}), {"success": True}

" Cookie_sc will not be written to response. It's just some kind of default
" value
def tuple_with_cookies(all_cookies=Cookies(), cookie_sc=Cookie("sc")):
    print("=====> cookies ")
    print("=====> cookie sc ")
    import datetime
    expires = datetime.datetime(2018, 12, 31)

    cks = Cookies()
    # cks = cookies.SimpleCookie() # you could also use the build-in cookie objects
    cks["ck1"] = "keijack"request
    cks["ck1"]["path"] = "/"
    cks["ck1"]["expires"] = expires.strftime(Cookies.EXPIRE_DATE_FORMAT)
    # You can ignore status code, headers, cookies even body in this tuple.
    return Header({"xx": "yyy"}), cks, "<html><body>OK</body></html>"

" If you visit /a/b/xyz/x,this controller function will be called, and `path_val` will be `xyz`
def my_path_val_ctr(path_val=PathValue()):
    return f"<html><body>{path_val}</body></html>"

@request_map("/star/*") # /star/c will find this controller, but /star/c/d not.
@request_map("*/star") # /c/star will find this controller, but /c/d/star not.
def star_path(path_val=PathValue()):
    return f"<html><body>{path_val}</body></html>"

@request_map("/star/**") # Both /star/c and /star/c/d will find this controller.
@request_map("**/star") # Both /c/star and /c/d/stars will find this controller.
def star_path(path_val=PathValue()):
    return f"<html><body>{path_val}</body></html>"

def redirect():
    return Redirect("/index")

def test_session(session=Session(), invalid=False):
    ins = session.get_attribute("in-session")
    if not ins:
        session.set_attribute("in-session", "Hello, Session!")

    __logger.info("session id: %s" % session.id)
    if invalid:
        __logger.info("session[%s] is being invalidated. " % session.id)
    return "<!DOCTYPE html><html><body>%s</body></html>" % str(ins)

# use coroutine, these controller functions will work both in a coroutine mode or threading mode.

async def say(sth: str = ""):
    _logger.info(f"Say: {sth}")
    return f"Success! {sth}"

async def coroutine_ctrl(hey: str = "Hey!"):
    return await say(hey)

def res_writer(response: Response):
    response.status_code = 200
    response.add_header("Content-Type", "application/octet-stream")

Beside using the default values, you can also use variable annotations to specify your controller function's variables.

@request_map("/say_hello/to/{name}", method=["GET", "POST", "PUT"])
def your_ctroller_function(
        user_name: str, # req.parameter["user_name"],400 error will raise when there's no such parameter in the query string.
        password: str, # req.parameter["password"],400 error will raise when there's no such parameter in the query string.
        skills: list, # req.parameters["skills"],400 error will raise when there's no such parameter in the query string.
        all_headers: Headers, # req.headers
        user_token: Header, # req.headers["user_token"],400 error will raise when there's no such parameter in the quest headers.
        all_cookies: Cookies, # req.cookies, return all cookies
        user_info: Cookie, # req.cookies["user_info"],400 error will raise when there's no such parameter in the cookies.
        name: PathValue, # req.path_values["name"],get the {name} value from your path.
        session: Session # req.getSession(True),get the session, if there is no sessions, create one.
    return "<html><body>Hello, World!</body></html>"

# you can use `params` to narrow the controller mapping, the following examples shows only the `params` mapping, ignoring the 
# `headers` examples for the usage is almost the same as the `params`. 
@request("/exact_params", method="GET", params="a=b")
def exact_params(a: str):
    print(f"{a}") # b
    return {"result": "ok"}

@request("/exact_params", method="GET", params="a!=b")
def exact_not_params(a: str):
    print(f"{a}") # b
    return {"result": "ok"}

@request("/exact_params", method="GET", params="a^=b")
def exact_startwith_params(a: str):
    print(f"{a}") # b
    return {"result": "ok"}

@request("/exact_params", method="GET", params="!a")
def no_params():
    return {"result": "ok"}

@request("/exact_params", method="GET", params="a")
def must_has_params():
    return {"result": "ok"}

# If multiple expressions are set, all expressions must be matched to enter this controller function.
@request("/exact_params", method="GET", params=["a=b", "c!=d"])
def multipul_params():
    return {"result": "ok"}

# You can set `match_all_params_expressions` to False to make that the url can enter this controller function even only one expression is matched.
@request("/exact_params", method="GET", params=["a=b", "c!=d"], match_all_params_expressions=False)
def multipul_params():
    return {"result": "ok"}

We recommend using functional programing to write controller functions. but if you realy want to use Object, you can use @request_map in a class method. For doing this, every time a new request comes, a new MyController object will be created.

class MyController:

    def __init__(self) -> None:
        self._name = "ctr object"

    @request_map("/obj/say_hello", method="GET")
    def my_ctrl_mth(self, name: str):
        return {"message": f"hello, {name}, {self._name} says. "}

If you want a singleton, you can add a @controller decorator to the class.

class MyController:

    def __init__(self) -> None:
        self._name = "ctr object"

    @request_map("/obj/say_hello", method="GET")
    def my_ctrl_mth(self, name: str):
        return {"message": f"hello, {name}, {self._name} says. "}

You can also add the @request_map to your class, this will be as the part of the url.

@request_map("/obj", method="GET")
class MyController:

    def __init__(self) -> None:
        self._name = "ctr object"

    def my_ctrl_default_mth(self, name: str):
        return {"message": f"hello, {name}, {self._name} says. "}

    @request_map("/say_hello", method=("GET", "POST"))
    def my_ctrl_mth(self, name: str):
        return {"message": f"hello, {name}, {self._name} says. "}

You can specify the init variables in @controller decorator.

@controller(args=["ctr_name"], kwargs={"desc": "this is a key word argument"})
@request_map("/obj", method="GET")
class MyController:

    def __init__(self, name, desc="") -> None:
        self._name = f"ctr[{name}] - {desc}"

    def my_ctrl_default_mth(self, name: str):
        return {"message": f"hello, {name}, {self._name} says. "}

    @request_map("/say_hello", method=("GET", "POST"))
    def my_ctrl_mth(self, name: str):
        return {"message": f"hello, {name}, {self._name} says. "}

From 0.7.0, @request_map support regular expression mapping.

# url `/reg/abcef/aref/xxx` can map the flowing controller:
@route(regexp="^(reg/(.+))$", method="GET")
def my_reg_ctr(reg_groups: RegGroups, reg_group: RegGroup = RegGroup(1)):
    print(reg_groups) # will output ("reg/abcef/aref/xxx", "abcef/aref/xxx")
    print(reg_group) # will output "abcef/aref/xxx"
    return f"{self._name}, {reg_group.group},{reg_group}"

Regular expression mapping a class:

@controller(args=["ctr_name"], kwargs={"desc": "this is a key word argument"})
@request_map("/obj", method="GET") # regexp do not work here, method will still available
class MyController:

    def __init__(self, name, desc="") -> None:
        self._name = f"ctr[{name}] - {desc}"

    def my_ctrl_default_mth(self, name: str):
        return {"message": f"hello, {name}, {self._name} says. "}

    @route(regexp="^(reg/(.+))$") # prefix `/obj`  from class decorator will be ignored, but `method`(GET in this example) from class decorator will still work.
    def my_ctrl_mth(self, name: str):
        return {"message": f"hello, {name}, {self._name} says. "}


Defaultly, the session is stored in local, you can extend SessionFactory and Session classes to implement your own session storage requirement (like store all data in redis or memcache)

from simple_http_server import Session, SessionFactory, set_session_factory

class MySessionImpl(Session):

    def __init__(self):
        # your own implementation

    def id(self) -> str:
        # your own implementation

    def creation_time(self) -> float:
        # your own implementation

    def last_accessed_time(self) -> float:
        # your own implementation

    def is_new(self) -> bool:
        # your own implementation

    def attribute_names(self) -> Tuple:
        # your own implementation

    def get_attribute(self, name: str) -> Any:
        # your own implementation

    def set_attribute(self, name: str, value: Any) -> None:
        # your own implementation

    def invalidate(self) -> None:
        # your own implementation

class MySessionFacImpl(SessionFactory):

    def __init__(self):
        # your own implementation

    def get_session(self, session_id: str, create: bool = False) -> Session:
        # your own implementation
        return MySessionImpl()


There is an offical Redis implementation here: https://github.com/keijack/python-simple-http-server-redis-session.git


To handle a websocket session, you should handle multiple events, so it's more reasonable to use a class rather than functions to do it.

In this framework, you should use @websocket_handler to decorate the class you want to handle websocket session. Specific event listener methods should be defined in a fixed way. However, the easiest way to do it is to inherit simple_http_server.WebsocketHandler class, and choose the event you want to implement. But this inheritance is not compulsory.

You can configure endpoit or regexp in @websocket_handler to setup which url the class should handle. Alongside, there is a singleton field, which is set to True by default. Which means that all connections are handle by ONE object of this class. If this field is set to False, objects will be created when every WebsocketSession try to connect.

from simple_http_server import WebsocketHandler, WebsocketRequest,WebsocketSession, websocket_handler

class WSHandler(WebsocketHandler):

    def on_handshake(self, request: WebsocketRequest):
        " You can get path/headers/path_values/cookies/query_string/query_parameters from request.
        " You should return a tuple means (http_status_code, headers)
        " If status code in (0, None, 101), the websocket will be connected, or will return the status you return. 
        " All headers will be send to client
        _logger.info(f">>{session.id}<< open! {request.path_values}")
        return 0, {}

    def on_open(self, session: WebsocketSession):
        " Will be called when the connection opened.
        _logger.info(f">>{session.id}<< open! {session.request.path_values}")

    def on_close(self, session: WebsocketSession, reason: str):
        " Will be called when the connection closed.
        _logger.info(f">>{session.id}<< close::{reason}")

    def on_ping_message(self, session: WebsocketSession = None, message: bytes = b''):
        " Will be called when receive a ping message. Will send all the message bytes back to client by default.

    def on_pong_message(self, session: WebsocketSession = None, message: bytes = ""):
        " Will be called when receive a pong message.

    def on_text_message(self, session: WebsocketSession, message: str):
        " Will be called when receive a text message.
        _logger.info(f">>{session.id}<< on text message: {message}")

    def on_binary_message(self, session: WebsocketSession = None, message: bytes = b''):
        " Will be called when receive a binary message if you have not consumed all the bytes in `on_binary_frame` 
        " method.

    def on_binary_frame(self, session: WebsocketSession = None, fin: bool = False, frame_payload: bytes = b''):
        " If you are sending a continuation binary message to server, this will be called every time a frame is 
        " received, you can consumed all the bytes in this method, e.g. save all bytes to a file. By doing so, 
        " you should not return and value in this method. 
        " If you does not implement this method or return a True in this method, all the bytes will be caced in
        " memory and be sent to your `on_binary_message` method.
        return True

@websocket_handler(regexp="^/ws-reg/([a-zA-Z0-9]+)$", singleton=False)
class WSHandler(WebsocketHandler):

    " You code here

Error pages

You can use @error_message to specify your own error page. See:

from simple_http_server import error_message
# map specified codes
@error_message("403", "404")
def my_40x_page(message: str, explain=""):
    return f"""
            message: {message}, explain: {explain}

# map specified code rangs
@error_message("40x", "50x")
def my_error_message(code, message, explain=""):
    return f"{code}-{message}-{explain}"

# map all error page
def my_error_message(code, message, explain=""):
    return f"{code}-{message}-{explain}"

Write filters

This server support filters, you can use request_filter decorator to define your filters.

from simple_http_server import request_filter

@request_filter("/tuple/**") # use wildcard
@request_filter(regexp="^/tuple") # use regular expression
def filter_tuple(ctx):
    print("---------- through filter ---------------")
    # add a header to request header
    ctx.request.headers["filter-set"] = "through filter"
    if "user_name" not in ctx.request.parameter:
    elif "pass" not in ctx.request.parameter:
        ctx.response.send_error(400, "pass should be passed")
        # you can also raise a HttpError
        # raise HttpError(400, "pass should be passed")
        # you should always use do_chain method to go to the next

Start your server

# If you place the controllers method in the other files, you should import them here.

import simple_http_server.server as server
import my_test_ctrl

def main(*args):
    # The following method can import several controller files once.
    server.scan("my_ctr_pkg", r".*controller.*")

if __name__ == "__main__":

If you want to specify the host and port:

    server.start(host="", port=8080)

If you want to specify the resources path:

    server.start(resources={"/path_prefix/*", "/absolute/dir/root/path", # Match the files in the given folder with a special path prefix.
                            "/path_prefix/**", "/absolute/dir/root/path", # Match all the files in the given folder and its sub-folders with a special path prefix.
                            "*.suffix", "/absolute/dir/root/path", # Match the specific files in the given folder.
                            "**.suffix", "/absolute/dir/root/path", # Match the specific files in the given folder and its sub-folders.

If you want to use ssl:

                 ssl_protocol=ssl.PROTOCOL_TLS_SERVER, # Optional, default is ssl.PROTOCOL_TLS_SERVER, which will auto detect the highted protocol version that both server and client support. 
                 ssl_check_hostname=False, #Optional, if set to True, if the hostname is not match the certificat, it cannot establish the connection, default is False.
                 keypass="", # Optional, your private key's password


From 0.12.0, you can use coroutine tasks than threads to handle requests, you can set the prefer_coroutine parameter in start method to enable the coroutine mode.


From 0.13.0, coroutine mode uses the coroutine server, that means all requests will use the async I/O rather than block I/O. So you can now use async def to define all your controllers including the Websocket event callback methods.

If you call the server starting in a async function, you can all its async version, by doing this, there sever will use the same event loop with your other async functions.

    await server.start_async(prefer_coroutine=True)


The default logger is try to write logs to the screen, you can specify the logger handler to write it to a file.

import simple_http_server.logger as logger
import logging

_formatter = logging.Formatter(fmt='[%(asctime)s]-[%(name)s]-%(levelname)-4s: %(message)s')
_handler = logging.TimedRotatingFileHandler("/var/log/simple_http_server.log", when="midnight", backupCount=7)


If you want to add a handler rather than replace the inner one, you can use:


If you want to change the logger level:


You can get a stand alone logger which is independent from the framework one via a new class logger.LoggerFactory.

import simple_http_server.logger as logger

log = logger.get_logger("my_service", "my_log_fac")

# If you want to set a different log level to this logger factory: 

log_fac = logger.get_logger_factory("my_log_fac")
log_fac.log_level = "DEBUG"
log = log_fac.get_logger("my_service")


WSGI Support

You can use this module in WSGI apps.

import simple_http_server.server as server
import os
from simple_http_server import request_map

# scan all your controllers
server.scan("tests/ctrls", r'.*controllers.*')
# or define a new controller function here
def my_controller(name: str):
    return 200, "Hello, WSGI!"
# resources is optional
wsgi_proxy = server.init_wsgi_proxy(resources={"/public/*": f"/you/static/files/path"})

# wsgi app entrance. 
def simple_app(environ, start_response):
    return wsgi_proxy.app_proxy(environ, start_response)

# If your entrance is async:
async def simple_app(envion, start_response):
    return await wsgi_proxy.async_app_proxy(environ, start_response)


The code that process websocket comes from the following project: https://github.com/Pithikos/python-websocket-server


一个超轻量级的 HTTP Server,支持线程和协程模式,源生支持 websocket 哦!你也可以非常容易的将其嵌入到 WSGI 服务器里。并且支持分布式 Session! expand collapse


No release





Load More
can not load any more


10d9f8b4 4838521 8bde8327 4838521