| Name | simple-http-server JSON |
| Version |
0.24.1
JSON |
| download |
| home_page | |
| Summary | This is a simple http server, use MVC like design. |
| upload_time | 2024-03-05 11:47:55 |
| maintainer | |
| docs_url | None |
| author | |
| requires_python | >=3.7 |
| license | MIT License Copyright (c) 2018 Keijack Wu Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
| keywords |
http-server
websocket
http
web
web-server
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# python-simple-http-server
[](https://badge.fury.io/py/simple-http-server)
## Discription
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](https://github.com/keijack/python-simple-http-server-redis-session).
* You can use [this extention](https://github.com/keijack/python-simple-http-server-jinja) to support `jinja` views.
* Spring MVC like request mapping.
* SSL support.
* Gzip support.
* Websocket support
* Easy to use.
* Free style controller writing.
* Easily integraded with WSGI servers.
* Easily integraded with ASGI servers. Websocket will be supported when ASGI server enable websocket functions.
* Coroutine mode support.
## Dependencies
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:
```shell
python3 -m pip install websocket-client
```
## How to use
### Install
```shell
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,<br />
using the route and server modules from simple_http_server
```python
from simple_http_server import route, server
@route("/")
def index():
return {"hello": "world"}
server.start(port=9090)
```
### Write Controllers
```python
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.
@request_map("/index")
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)
@request_map("/error")
def my_ctrl3():
return Response(status_code=500)
@request_map("/exception")
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
@request_map("/tuple")
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
"""
@request_map("tuple_cookie")
def tuple_with_cookies(all_cookies=Cookies(), cookie_sc=Cookie("sc")):
print("=====> cookies ")
print(all_cookies)
print("=====> cookie sc ")
print(cookie_sc)
print("======<")
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`
"""
@request_map("/a/b/{path_val}/x")
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>"
@request_map("/redirect")
def redirect():
return Redirect("/index")
@request_map("session")
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)
session.invalidate()
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}"
@request_map("/中文/coroutine")
async def coroutine_ctrl(hey: str = "Hey!"):
return await say(hey)
@route("/res/write/bytes")
def res_writer(response: Response):
response.status_code = 200
response.add_header("Content-Type", "application/octet-stream")
response.write_bytes(b'abcd')
response.write_bytes(bytearray(b'efg'))
response.close()
```
Beside using the default values, you can also use variable annotations to specify your controller function's variables.
```python
@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.
```python
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.
```python
@controller
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.
```python
@controller
@request_map("/obj", method="GET")
class MyController:
def __init__(self) -> None:
self._name = "ctr object"
@request_map
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.
```python
@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}"
@request_map
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.
```python
# 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:
```python
@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}"
@request_map
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. "}
```
### Model binding
If the inner binding can not satisfied you, you can use `@model_binding` and `@default_model_binding` to define your own binding logic.
You can use `@model_binding` to define the binding of specified types.
```python
from typing import Any
from simple_http_server.models.model_bindings import ModelBinding
from simple_http_server import model_binding
from simple_http_server import HttpError, route
class Person:
def __init__(self, name: str = "", sex: int = "", age: int = 0) -> None:
self.name = name
self.sex = sex
self.age = age
@model_binding(Person)
class PersonModelBinding(ModelBinding):
async def bind(self) -> Any:
name = self.request.get_parameter("name", "no-one")
sex = self.request.get_parameter("sex", "secret")
try:
age = int(self.request.get_parameter("age", ""))
except:
raise HttpError(400, "Age is required, and must be an integer")
return Person(name, sex, age)
# Now, you can use `Person` in your controller attributes.
@route("/model_binding/person")
def test_model_binding(person: Person):
return {
"name": person.name,
"sex": person.sex,
"age": person.age,
}
```
You can also use `@default_model_binding` to handle all the types that not defined in inner and the `@model_binding` configurations.
```python
from simple_http_server.models.model_bindings import ModelBinding
from simple_http_server import default_model_binding
from simple_http_server import HttpError, route
class Dog:
def __init__(self, name="a dog") -> None:
self.name = name
def wang(self):
return self.name
@default_model_binding
class SetAttrModelBinding(ModelBinding):
def bind(self) -> Any:
# You can define `bind` method in normal or async ways.
try:
obj = self.arg_type()
for k, v in self.request.parameter.items():
setattr(obj, k, v)
return obj
except Exception as e:
_logger.warning(
f"Cannot create Object with given type {self.arg_type}. ", stack_info=True)
return self.default_value
@route("/model_binding/dog")
def test_model_binding_dog(dog: Dog):
return {
"name": dog.wang()
}
```
### Session
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)
```python
from simple_http_server import Session, SessionFactory, set_session_factory
class MySessionImpl(Session):
def __init__(self):
super().__init__()
# your own implementation
@property
def id(self) -> str:
# your own implementation
@property
def creation_time(self) -> float:
# your own implementation
@property
def last_accessed_time(self) -> float:
# your own implementation
@property
def is_new(self) -> bool:
# your own implementation
@property
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):
super().__init__()
# your own implementation
def get_session(self, session_id: str, create: bool = False) -> Session:
# your own implementation
return MySessionImpl()
set_session_factory(MySessionFacImpl())
```
There is an offical Redis implementation here: https://github.com/keijack/python-simple-http-server-redis-session.git
### Websocket
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.
```python
from simple_http_server import WebsocketHandler, WebsocketRequest,WebsocketSession, websocket_handler
@websocket_handler(endpoint="/ws/{path_val}")
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.
"
"""
session.send_pone(message)
def on_pong_message(self, session: WebsocketSession = None, message: bytes = ""):
"""
"
" Will be called when receive a pong message.
"
"""
pass
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}")
session.send(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.
"
"""
pass
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
"""
```
But if you want to only handle one event, you can also use a function to handle it.
```python
from simple_http_server import WebsocketCloseReason, WebsocketHandler, WebsocketRequest, WebsocketSession, websocket_message, websocket_handshake, websocket_open, websocket_close, WEBSOCKET_MESSAGE_TEXT
@websocket_handshake(endpoint="/ws-fun/{path_val}")
def ws_handshake(request: WebsocketRequest):
return 0, {}
@websocket_open(endpoint="/ws-fun/{path_val}")
def ws_open(session: WebsocketSession):
_logger.info(f">>{session.id}<< open! {session.request.path_values}")
@websocket_close(endpoint="/ws-fun/{path_val}")
def ws_close(session: WebsocketSession, reason: WebsocketCloseReason):
_logger.info(
f">>{session.id}<< close::{reason.message}-{reason.code}-{reason.reason}")
@websocket_message(endpoint="/ws-fun/{path_val}", message_type=WEBSOCKET_MESSAGE_TEXT)
# You can define a function in a sync or async way.
async def ws_text(session: WebsocketSession, message: str):
_logger.info(f">>{session.id}<< on text message: {message}")
session.send(f"{session.request.path_values['path_val']}-{message}")
if message == "close":
session.close()
```
### Error pages
You can use `@error_message` to specify your own error page. See:
```python
from simple_http_server import error_message
# map specified codes
@error_message("403", "404")
def my_40x_page(message: str, explain=""):
return f"""
<html>
<head>
<title>发生错误！</title>
<head>
<body>
message: {message}, explain: {explain}
</body>
</html>
"""
# map specified code rangs
@error_message("40x", "50x")
def my_error_message(code, message, explain=""):
return f"{code}-{message}-{explain}"
# map all error page
@error_message
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.
```python
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:
ctx.response.send_redirect("/index")
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")
else:
# you should always use do_chain method to go to the next
ctx.do_chain()
```
### Start your server
```python
# 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.*")
server.start()
if __name__ == "__main__":
main()
```
If you want to specify the host and port:
```python
server.start(host="", port=8080)
```
If you want to specify the resources path:
```python
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:
```python
server.start(host="",
port=8443,
ssl=True,
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.
keyfile="/path/to/your/keyfile.key",
certfile="/path/to/your/certfile.cert",
keypass="", # Optional, your private key's password
)
```
### Coroutine
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.
```python
server.start(prefer_coroutine=True)
```
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.
```python
await server.start_async(prefer_coroutine=True)
```
### Gzip
If you want to the response return gzip data:
```python
server.start(host="",
port=8080,
gzip_content_types={"text/html", "text/plain", "text/css", "application/json", "text/javascript"},
gzip_compress_level=9)
```
## Logger
The default logger is try to write logs to the screen, you can specify the logger handler to write it to a file.
```python
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)
_handler.setFormatter(_formatter)
_handler.setLevel("INFO")
logger.set_handler(_handler)
```
If you want to add a handler rather than replace the inner one, you can use:
```python
logger.add_handler(_handler)
```
If you want to change the logger level:
```python
logger.set_level("DEBUG")
```
You can get a stand alone logger which is independent from the framework one via a new class `logger.LoggerFactory`.
```python
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")
log.info(...)
```
## WSGI Support
You can use this module in WSGI apps.
```python
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
@request_map("/hello_wsgi")
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)
```
## ASGI Support
You can use this module in ASGI server, take `uvicorn` fro example:
```python
import asyncio
import uvicorn
import simple_http_server.server as server
from simple_http_server.server import ASGIProxy
asgi_proxy: ASGIProxy = None
init_asgi_proxy_lock: asyncio.Lock = asyncio.Lock()
async def init_asgi_proxy():
global asgi_proxy
if asgi_proxy == None:
async with init_asgi_proxy_lock:
if asgi_proxy == None:
server.scan(base_dir="tests/ctrls", regx=r'.*controllers.*')
asgi_proxy = server.init_asgi_proxy(resources={"/public/*": "tests/static"})
async def app(scope, receive, send):
await init_asgi_proxy()
await asgi_proxy.app_proxy(scope, receive, send)
def main():
config = uvicorn.Config("main:app", host="0.0.0.0", port=9090, log_level="info")
asgi_server = uvicorn.Server(config)
asgi_server.run()
if __name__ == "__main__":
main()
```
## Thanks
The code that process websocket comes from the following project: https://github.com/Pithikos/python-websocket-server
Raw data
{
"_id": null,
"home_page": "",
"name": "simple-http-server",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": "",
"keywords": "http-server,websocket,http,web,web-server",
"author": "",
"author_email": "keijack <keijack.wu@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/f2/31/a939d6773d95cc6415fdf18c79bbdf1a46906809321cbf3b2aa5756d36f5/simple_http_server-0.24.1.tar.gz",
"platform": null,
"description": "# python-simple-http-server\n\n[](https://badge.fury.io/py/simple-http-server)\n\n## Discription\n\nThis is a simple http server, use MVC like design.\n\n## Support Python Version\n\nPython 3.7+\n\n## Why choose\n\n* Lightway.\n* Functional programing.\n* Filter chain support.\n* Session support, and can support distributed session by [this extention](https://github.com/keijack/python-simple-http-server-redis-session).\n* You can use [this extention](https://github.com/keijack/python-simple-http-server-jinja) to support `jinja` views.\n* Spring MVC like request mapping.\n* SSL support.\n* Gzip support.\n* Websocket support\n* Easy to use.\n* Free style controller writing.\n* Easily integraded with WSGI servers. \n* Easily integraded with ASGI servers. Websocket will be supported when ASGI server enable websocket functions.\n* Coroutine mode support.\n\n## Dependencies\n\nThere 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:\n\n```shell\npython3 -m pip install websocket-client\n```\n\n## How to use\n\n### Install\n\n```shell\npython3 -m pip install simple_http_server\n```\n\n### Minimum code / component requirement setup\n\nMinimum code to get things started should have at least one controller function,<br /> \nusing the route and server modules from simple_http_server\n\n```python\nfrom simple_http_server import route, server\n \n@route(\"/\")\ndef index():\n return {\"hello\": \"world\"} \n\nserver.start(port=9090)\n```\n\n### Write Controllers\n\n```python\n\nfrom simple_http_server import request_map\nfrom simple_http_server import Response\nfrom simple_http_server import MultipartFile\nfrom simple_http_server import Parameter\nfrom simple_http_server import Parameters\nfrom simple_http_server import Header\nfrom simple_http_server import JSONBody\nfrom simple_http_server import HttpError\nfrom simple_http_server import StaticFile\nfrom simple_http_server import Headers\nfrom simple_http_server import Cookies\nfrom simple_http_server import Cookie\nfrom simple_http_server import Redirect\nfrom simple_http_server import ModelDict\n\n# request_map has an alias name `route`, you can select the one you familiar with.\n@request_map(\"/index\")\ndef my_ctrl():\n return {\"code\": 0, \"message\": \"success\"} # You can return a dictionary, a string or a `simple_http_server.simple_http_server.Response` object.\n\n\n@route(\"/say_hello\", method=[\"GET\", \"POST\"])\ndef my_ctrl2(name, name2=Parameter(\"name\", default=\"KEIJACK\"), model=ModelDict()):\n \"\"\"name and name2 is the same\"\"\"\n name == name2 # True\n name == model[\"name\"] # True\n return \"<!DOCTYPE html><html><body>hello, %s, %s</body></html>\" % (name, name2)\n\n\n@request_map(\"/error\")\ndef my_ctrl3():\n return Response(status_code=500)\n\n\n@request_map(\"/exception\")\ndef exception_ctrl():\n raise HttpError(400, \"Exception\")\n\n@request_map(\"/upload\", method=\"GET\")\ndef show_upload():\n root = os.path.dirname(os.path.abspath(__file__))\n return StaticFile(\"%s/my_dev/my_test_index.html\" % root, \"text/html; charset=utf-8\")\n\n\n@request_map(\"/upload\", method=\"POST\")\ndef my_upload(img=MultipartFile(\"img\")):\n root = os.path.dirname(os.path.abspath(__file__))\n img.save_to_file(root + \"/my_dev/imgs/\" + img.filename)\n return \"<!DOCTYPE html><html><body>upload ok!</body></html>\"\n\n\n@request_map(\"/post_txt\", method=\"POST\")\ndef normal_form_post(txt):\n return \"<!DOCTYPE html><html><body>hi, %s</body></html>\" % txt\n\n@request_map(\"/tuple\")\ndef tuple_results():\n # The order here is not important, we consider the first `int` value as status code,\n # All `Headers` object will be sent to the response\n # And the first valid object whose type in (str, unicode, dict, StaticFile, bytes) will\n # be considered as the body\n return 200, Headers({\"my-header\": \"headers\"}), {\"success\": True}\n\n\"\"\"\n\" Cookie_sc will not be written to response. It's just some kind of default\n\" value\n\"\"\"\n@request_map(\"tuple_cookie\")\ndef tuple_with_cookies(all_cookies=Cookies(), cookie_sc=Cookie(\"sc\")):\n print(\"=====> cookies \")\n print(all_cookies)\n print(\"=====> cookie sc \")\n print(cookie_sc)\n print(\"======<\")\n import datetime\n expires = datetime.datetime(2018, 12, 31)\n\n cks = Cookies()\n # cks = cookies.SimpleCookie() # you could also use the build-in cookie objects\n cks[\"ck1\"] = \"keijack\"request\n cks[\"ck1\"][\"path\"] = \"/\"\n cks[\"ck1\"][\"expires\"] = expires.strftime(Cookies.EXPIRE_DATE_FORMAT)\n # You can ignore status code, headers, cookies even body in this tuple.\n return Header({\"xx\": \"yyy\"}), cks, \"<html><body>OK</body></html>\"\n\n\"\"\"\n\" If you visit /a/b/xyz/x\uff0cthis controller function will be called, and `path_val` will be `xyz`\n\"\"\"\n@request_map(\"/a/b/{path_val}/x\")\ndef my_path_val_ctr(path_val=PathValue()):\n return f\"<html><body>{path_val}</body></html>\"\n\n@request_map(\"/star/*\") # /star/c will find this controller, but /star/c/d not.\n@request_map(\"*/star\") # /c/star will find this controller, but /c/d/star not.\ndef star_path(path_val=PathValue()):\n return f\"<html><body>{path_val}</body></html>\"\n\n@request_map(\"/star/**\") # Both /star/c and /star/c/d will find this controller.\n@request_map(\"**/star\") # Both /c/star and /c/d/stars will find this controller.\ndef star_path(path_val=PathValue()):\n return f\"<html><body>{path_val}</body></html>\"\n\n@request_map(\"/redirect\")\ndef redirect():\n return Redirect(\"/index\")\n\n@request_map(\"session\")\ndef test_session(session=Session(), invalid=False):\n ins = session.get_attribute(\"in-session\")\n if not ins:\n session.set_attribute(\"in-session\", \"Hello, Session!\")\n\n __logger.info(\"session id: %s\" % session.id)\n if invalid:\n __logger.info(\"session[%s] is being invalidated. \" % session.id)\n session.invalidate()\n return \"<!DOCTYPE html><html><body>%s</body></html>\" % str(ins)\n\n# use coroutine, these controller functions will work both in a coroutine mode or threading mode.\n\nasync def say(sth: str = \"\"):\n _logger.info(f\"Say: {sth}\")\n return f\"Success! {sth}\"\n\n@request_map(\"/\u4e2d\u6587/coroutine\")\nasync def coroutine_ctrl(hey: str = \"Hey!\"):\n return await say(hey)\n\n@route(\"/res/write/bytes\")\ndef res_writer(response: Response):\n response.status_code = 200\n response.add_header(\"Content-Type\", \"application/octet-stream\")\n response.write_bytes(b'abcd')\n response.write_bytes(bytearray(b'efg'))\n response.close()\n```\n\nBeside using the default values, you can also use variable annotations to specify your controller function's variables.\n\n```python\n@request_map(\"/say_hello/to/{name}\", method=[\"GET\", \"POST\", \"PUT\"])\ndef your_ctroller_function(\n user_name: str, # req.parameter[\"user_name\"]\uff0c400 error will raise when there's no such parameter in the query string.\n password: str, # req.parameter[\"password\"]\uff0c400 error will raise when there's no such parameter in the query string.\n skills: list, # req.parameters[\"skills\"]\uff0c400 error will raise when there's no such parameter in the query string.\n all_headers: Headers, # req.headers\n user_token: Header, # req.headers[\"user_token\"]\uff0c400 error will raise when there's no such parameter in the quest headers.\n all_cookies: Cookies, # req.cookies, return all cookies\n user_info: Cookie, # req.cookies[\"user_info\"]\uff0c400 error will raise when there's no such parameter in the cookies.\n name: PathValue, # req.path_values[\"name\"]\uff0cget the {name} value from your path.\n session: Session # req.getSession(True)\uff0cget the session, if there is no sessions, create one.\n ):\n return \"<html><body>Hello, World!</body></html>\"\n\n# you can use `params` to narrow the controller mapping, the following examples shows only the `params` mapping, ignoring the \n# `headers` examples for the usage is almost the same as the `params`. \n@request(\"/exact_params\", method=\"GET\", params=\"a=b\")\ndef exact_params(a: str):\n print(f\"{a}\") # b\n return {\"result\": \"ok\"}\n\n@request(\"/exact_params\", method=\"GET\", params=\"a!=b\")\ndef exact_not_params(a: str):\n print(f\"{a}\") # b\n return {\"result\": \"ok\"}\n\n@request(\"/exact_params\", method=\"GET\", params=\"a^=b\")\ndef exact_startwith_params(a: str):\n print(f\"{a}\") # b\n return {\"result\": \"ok\"}\n\n@request(\"/exact_params\", method=\"GET\", params=\"!a\")\ndef no_params():\n return {\"result\": \"ok\"}\n\n@request(\"/exact_params\", method=\"GET\", params=\"a\")\ndef must_has_params():\n return {\"result\": \"ok\"}\n\n# If multiple expressions are set, all expressions must be matched to enter this controller function.\n@request(\"/exact_params\", method=\"GET\", params=[\"a=b\", \"c!=d\"])\ndef multipul_params():\n return {\"result\": \"ok\"}\n\n# 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.\n@request(\"/exact_params\", method=\"GET\", params=[\"a=b\", \"c!=d\"], match_all_params_expressions=False)\ndef multipul_params():\n return {\"result\": \"ok\"}\n```\n\nWe 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.\n\n```python\n\nclass MyController:\n\n def __init__(self) -> None:\n self._name = \"ctr object\"\n\n @request_map(\"/obj/say_hello\", method=\"GET\")\n def my_ctrl_mth(self, name: str):\n return {\"message\": f\"hello, {name}, {self._name} says. \"}\n\n```\n\nIf you want a singleton, you can add a `@controller` decorator to the class.\n\n```python\n\n@controller\nclass MyController:\n\n def __init__(self) -> None:\n self._name = \"ctr object\"\n\n @request_map(\"/obj/say_hello\", method=\"GET\")\n def my_ctrl_mth(self, name: str):\n return {\"message\": f\"hello, {name}, {self._name} says. \"}\n\n```\n\nYou can also add the `@request_map` to your class, this will be as the part of the url.\n\n```python\n\n@controller\n@request_map(\"/obj\", method=\"GET\")\nclass MyController:\n\n def __init__(self) -> None:\n self._name = \"ctr object\"\n\n @request_map\n def my_ctrl_default_mth(self, name: str):\n return {\"message\": f\"hello, {name}, {self._name} says. \"}\n\n @request_map(\"/say_hello\", method=(\"GET\", \"POST\"))\n def my_ctrl_mth(self, name: str):\n return {\"message\": f\"hello, {name}, {self._name} says. \"}\n\n```\n\nYou can specify the `init` variables in `@controller` decorator. \n\n```python\n\n@controller(args=[\"ctr_name\"], kwargs={\"desc\": \"this is a key word argument\"})\n@request_map(\"/obj\", method=\"GET\")\nclass MyController:\n\n def __init__(self, name, desc=\"\") -> None:\n self._name = f\"ctr[{name}] - {desc}\"\n\n @request_map\n def my_ctrl_default_mth(self, name: str):\n return {\"message\": f\"hello, {name}, {self._name} says. \"}\n\n @request_map(\"/say_hello\", method=(\"GET\", \"POST\"))\n def my_ctrl_mth(self, name: str):\n return {\"message\": f\"hello, {name}, {self._name} says. \"}\n\n```\n\nFrom `0.7.0`, `@request_map` support regular expression mapping. \n\n```python\n# url `/reg/abcef/aref/xxx` can map the flowing controller:\n@route(regexp=\"^(reg/(.+))$\", method=\"GET\")\ndef my_reg_ctr(reg_groups: RegGroups, reg_group: RegGroup = RegGroup(1)):\n print(reg_groups) # will output (\"reg/abcef/aref/xxx\", \"abcef/aref/xxx\")\n print(reg_group) # will output \"abcef/aref/xxx\"\n return f\"{self._name}, {reg_group.group},{reg_group}\"\n```\nRegular expression mapping a class:\n\n```python\n@controller(args=[\"ctr_name\"], kwargs={\"desc\": \"this is a key word argument\"})\n@request_map(\"/obj\", method=\"GET\") # regexp do not work here, method will still available\nclass MyController:\n\n def __init__(self, name, desc=\"\") -> None:\n self._name = f\"ctr[{name}] - {desc}\"\n\n @request_map\n def my_ctrl_default_mth(self, name: str):\n return {\"message\": f\"hello, {name}, {self._name} says. \"}\n\n @route(regexp=\"^(reg/(.+))$\") # prefix `/obj` from class decorator will be ignored, but `method`(GET in this example) from class decorator will still work.\n def my_ctrl_mth(self, name: str):\n return {\"message\": f\"hello, {name}, {self._name} says. \"}\n\n```\n\n### Model binding\n\nIf the inner binding can not satisfied you, you can use `@model_binding` and `@default_model_binding` to define your own binding logic.\n\nYou can use `@model_binding` to define the binding of specified types.\n\n```python\nfrom typing import Any\nfrom simple_http_server.models.model_bindings import ModelBinding\nfrom simple_http_server import model_binding\nfrom simple_http_server import HttpError, route\n\nclass Person:\n\n def __init__(self, name: str = \"\", sex: int = \"\", age: int = 0) -> None:\n self.name = name\n self.sex = sex\n self.age = age\n\n@model_binding(Person)\nclass PersonModelBinding(ModelBinding):\n\n async def bind(self) -> Any:\n name = self.request.get_parameter(\"name\", \"no-one\")\n sex = self.request.get_parameter(\"sex\", \"secret\")\n try:\n age = int(self.request.get_parameter(\"age\", \"\"))\n except:\n raise HttpError(400, \"Age is required, and must be an integer\")\n return Person(name, sex, age)\n\n# Now, you can use `Person` in your controller attributes.\n@route(\"/model_binding/person\")\ndef test_model_binding(person: Person):\n return {\n \"name\": person.name,\n \"sex\": person.sex,\n \"age\": person.age,\n }\n\n```\n\nYou can also use `@default_model_binding` to handle all the types that not defined in inner and the `@model_binding` configurations.\n\n```python\nfrom simple_http_server.models.model_bindings import ModelBinding\nfrom simple_http_server import default_model_binding\nfrom simple_http_server import HttpError, route\n\nclass Dog:\n\n def __init__(self, name=\"a dog\") -> None:\n self.name = name\n\n def wang(self):\n return self.name\n\n@default_model_binding\nclass SetAttrModelBinding(ModelBinding):\n\n def bind(self) -> Any:\n # You can define `bind` method in normal or async ways. \n try:\n obj = self.arg_type()\n for k, v in self.request.parameter.items():\n setattr(obj, k, v)\n return obj\n except Exception as e:\n _logger.warning(\n f\"Cannot create Object with given type {self.arg_type}. \", stack_info=True)\n return self.default_value\n\n@route(\"/model_binding/dog\")\ndef test_model_binding_dog(dog: Dog):\n return {\n \"name\": dog.wang()\n }\n```\n\n\n### Session\n\nDefaultly, 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)\n\n```python\nfrom simple_http_server import Session, SessionFactory, set_session_factory\n\nclass MySessionImpl(Session):\n\n def __init__(self):\n super().__init__()\n # your own implementation\n\n @property\n def id(self) -> str:\n # your own implementation\n\n @property\n def creation_time(self) -> float:\n # your own implementation\n\n @property\n def last_accessed_time(self) -> float:\n # your own implementation\n\n @property\n def is_new(self) -> bool:\n # your own implementation\n\n @property\n def attribute_names(self) -> Tuple:\n # your own implementation\n\n def get_attribute(self, name: str) -> Any:\n # your own implementation\n\n def set_attribute(self, name: str, value: Any) -> None:\n # your own implementation\n\n def invalidate(self) -> None:\n # your own implementation\n\nclass MySessionFacImpl(SessionFactory):\n\n def __init__(self):\n super().__init__()\n # your own implementation\n\n \n def get_session(self, session_id: str, create: bool = False) -> Session:\n # your own implementation\n return MySessionImpl()\n\nset_session_factory(MySessionFacImpl())\n\n```\n\nThere is an offical Redis implementation here: https://github.com/keijack/python-simple-http-server-redis-session.git\n\n### Websocket\n\nTo handle a websocket session, you should handle multiple events, so it's more reasonable to use a class rather than functions to do it. \n\nIn 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.\n\nYou 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.\n\n```python\nfrom simple_http_server import WebsocketHandler, WebsocketRequest,WebsocketSession, websocket_handler\n\n@websocket_handler(endpoint=\"/ws/{path_val}\")\nclass WSHandler(WebsocketHandler):\n\n def on_handshake(self, request: WebsocketRequest):\n \"\"\"\n \"\n \" You can get path/headers/path_values/cookies/query_string/query_parameters from request.\n \" \n \" You should return a tuple means (http_status_code, headers)\n \"\n \" If status code in (0, None, 101), the websocket will be connected, or will return the status you return. \n \"\n \" All headers will be send to client\n \"\n \"\"\"\n _logger.info(f\">>{session.id}<< open! {request.path_values}\")\n return 0, {}\n\n def on_open(self, session: WebsocketSession):\n \"\"\"\n \" \n \" Will be called when the connection opened.\n \"\n \"\"\"\n _logger.info(f\">>{session.id}<< open! {session.request.path_values}\")\n\n def on_close(self, session: WebsocketSession, reason: str):\n \"\"\"\n \"\n \" Will be called when the connection closed.\n \"\n \"\"\"\n _logger.info(f\">>{session.id}<< close::{reason}\")\n\n def on_ping_message(self, session: WebsocketSession = None, message: bytes = b''):\n \"\"\"\n \"\n \" Will be called when receive a ping message. Will send all the message bytes back to client by default.\n \"\n \"\"\"\n session.send_pone(message)\n\n def on_pong_message(self, session: WebsocketSession = None, message: bytes = \"\"):\n \"\"\"\n \"\n \" Will be called when receive a pong message.\n \"\n \"\"\"\n pass\n\n def on_text_message(self, session: WebsocketSession, message: str):\n \"\"\"\n \"\n \" Will be called when receive a text message.\n \"\n \"\"\"\n _logger.info(f\">>{session.id}<< on text message: {message}\")\n session.send(message)\n\n def on_binary_message(self, session: WebsocketSession = None, message: bytes = b''):\n \"\"\"\n \"\n \" Will be called when receive a binary message if you have not consumed all the bytes in `on_binary_frame` \n \" method.\n \"\n \"\"\"\n pass\n\n def on_binary_frame(self, session: WebsocketSession = None, fin: bool = False, frame_payload: bytes = b''):\n \"\"\"\n \"\n \" If you are sending a continuation binary message to server, this will be called every time a frame is \n \" received, you can consumed all the bytes in this method, e.g. save all bytes to a file. By doing so, \n \" you should not return and value in this method. \n \"\n \" If you does not implement this method or return a True in this method, all the bytes will be caced in\n \" memory and be sent to your `on_binary_message` method.\n \"\n \"\"\"\n return True\n\n@websocket_handler(regexp=\"^/ws-reg/([a-zA-Z0-9]+)$\", singleton=False)\nclass WSHandler(WebsocketHandler):\n\n \"\"\"\n \" You code here\n \"\"\"\n\n```\n\nBut if you want to only handle one event, you can also use a function to handle it. \n\n```python\n\nfrom simple_http_server import WebsocketCloseReason, WebsocketHandler, WebsocketRequest, WebsocketSession, websocket_message, websocket_handshake, websocket_open, websocket_close, WEBSOCKET_MESSAGE_TEXT\n\n@websocket_handshake(endpoint=\"/ws-fun/{path_val}\")\ndef ws_handshake(request: WebsocketRequest):\n return 0, {}\n\n\n@websocket_open(endpoint=\"/ws-fun/{path_val}\")\ndef ws_open(session: WebsocketSession):\n _logger.info(f\">>{session.id}<< open! {session.request.path_values}\")\n\n\n@websocket_close(endpoint=\"/ws-fun/{path_val}\")\ndef ws_close(session: WebsocketSession, reason: WebsocketCloseReason):\n _logger.info(\n f\">>{session.id}<< close::{reason.message}-{reason.code}-{reason.reason}\")\n\n\n@websocket_message(endpoint=\"/ws-fun/{path_val}\", message_type=WEBSOCKET_MESSAGE_TEXT)\n# You can define a function in a sync or async way.\nasync def ws_text(session: WebsocketSession, message: str): \n _logger.info(f\">>{session.id}<< on text message: {message}\")\n session.send(f\"{session.request.path_values['path_val']}-{message}\")\n if message == \"close\":\n session.close()\n```\n\n### Error pages\n\nYou can use `@error_message` to specify your own error page. See:\n\n```python\nfrom simple_http_server import error_message\n# map specified codes\n@error_message(\"403\", \"404\")\ndef my_40x_page(message: str, explain=\"\"):\n return f\"\"\"\n <html>\n <head>\n <title>\u53d1\u751f\u9519\u8bef\uff01</title>\n <head>\n <body>\n message: {message}, explain: {explain}\n </body>\n </html>\n \"\"\"\n\n# map specified code rangs\n@error_message(\"40x\", \"50x\")\ndef my_error_message(code, message, explain=\"\"):\n return f\"{code}-{message}-{explain}\"\n\n# map all error page\n@error_message\ndef my_error_message(code, message, explain=\"\"):\n return f\"{code}-{message}-{explain}\"\n```\n\n### Write filters\n\nThis server support filters, you can use `request_filter` decorator to define your filters.\n\n```python\nfrom simple_http_server import request_filter\n\n@request_filter(\"/tuple/**\") # use wildcard\n@request_filter(regexp=\"^/tuple\") # use regular expression\ndef filter_tuple(ctx):\n print(\"---------- through filter ---------------\")\n # add a header to request header\n ctx.request.headers[\"filter-set\"] = \"through filter\"\n if \"user_name\" not in ctx.request.parameter:\n ctx.response.send_redirect(\"/index\")\n elif \"pass\" not in ctx.request.parameter:\n ctx.response.send_error(400, \"pass should be passed\")\n # you can also raise a HttpError\n # raise HttpError(400, \"pass should be passed\")\n else:\n # you should always use do_chain method to go to the next\n ctx.do_chain()\n```\n\n### Start your server\n\n```python\n# If you place the controllers method in the other files, you should import them here.\n\nimport simple_http_server.server as server\nimport my_test_ctrl\n\n\ndef main(*args):\n # The following method can import several controller files once.\n server.scan(\"my_ctr_pkg\", r\".*controller.*\")\n server.start()\n\nif __name__ == \"__main__\":\n main()\n```\n\nIf you want to specify the host and port:\n\n```python\n server.start(host=\"\", port=8080)\n```\n\nIf you want to specify the resources path: \n\n```python \n server.start(resources={\"/path_prefix/*\", \"/absolute/dir/root/path\", # Match the files in the given folder with a special path prefix.\n \"/path_prefix/**\", \"/absolute/dir/root/path\", # Match all the files in the given folder and its sub-folders with a special path prefix.\n \"*.suffix\", \"/absolute/dir/root/path\", # Match the specific files in the given folder.\n \"**.suffix\", \"/absolute/dir/root/path\", # Match the specific files in the given folder and its sub-folders.\n })\n```\n\nIf you want to use ssl:\n\n```python\n server.start(host=\"\", \n port=8443,\n ssl=True,\n 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. \n 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.\n keyfile=\"/path/to/your/keyfile.key\",\n certfile=\"/path/to/your/certfile.cert\",\n keypass=\"\", # Optional, your private key's password\n )\n```\n\n### Coroutine\n\nFrom `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. \n\n```python\n server.start(prefer_coroutine=True)\n```\n\nFrom `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.\n\nIf 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. \n\n```python\n await server.start_async(prefer_coroutine=True)\n```\n\n### Gzip\n\nIf you want to the response return gzip data:\n\n```python\nserver.start(host=\"\", \n port=8080, \n gzip_content_types={\"text/html\", \"text/plain\", \"text/css\", \"application/json\", \"text/javascript\"}, \n gzip_compress_level=9)\n```\n\n## Logger\n\nThe default logger is try to write logs to the screen, you can specify the logger handler to write it to a file.\n\n```python\nimport simple_http_server.logger as logger\nimport logging\n\n_formatter = logging.Formatter(fmt='[%(asctime)s]-[%(name)s]-%(levelname)-4s: %(message)s')\n_handler = logging.TimedRotatingFileHandler(\"/var/log/simple_http_server.log\", when=\"midnight\", backupCount=7)\n_handler.setFormatter(_formatter)\n_handler.setLevel(\"INFO\")\n\nlogger.set_handler(_handler)\n```\n\nIf you want to add a handler rather than replace the inner one, you can use:\n\n```python\nlogger.add_handler(_handler)\n```\n\nIf you want to change the logger level:\n\n```python\nlogger.set_level(\"DEBUG\")\n```\n\nYou can get a stand alone logger which is independent from the framework one via a new class `logger.LoggerFactory`. \n\n```python\nimport simple_http_server.logger as logger\n\nlog = logger.get_logger(\"my_service\", \"my_log_fac\")\n\n# If you want to set a different log level to this logger factory: \n\nlog_fac = logger.get_logger_factory(\"my_log_fac\")\nlog_fac.log_level = \"DEBUG\"\nlog = log_fac.get_logger(\"my_service\")\n\nlog.info(...)\n\n```\n\n\n## WSGI Support\n\nYou can use this module in WSGI apps. \n\n```python\nimport simple_http_server.server as server\nimport os\nfrom simple_http_server import request_map\n\n\n# scan all your controllers\nserver.scan(\"tests/ctrls\", r'.*controllers.*')\n# or define a new controller function here\n@request_map(\"/hello_wsgi\")\ndef my_controller(name: str):\n return 200, \"Hello, WSGI!\"\n# resources is optional\nwsgi_proxy = server.init_wsgi_proxy(resources={\"/public/*\": f\"/you/static/files/path\"})\n\n# wsgi app entrance. \ndef simple_app(environ, start_response):\n return wsgi_proxy.app_proxy(environ, start_response)\n\n# If your entrance is async:\nasync def simple_app(envion, start_response):\n return await wsgi_proxy.async_app_proxy(environ, start_response)\n```\n\n## ASGI Support\n\nYou can use this module in ASGI server, take `uvicorn` fro example:\n\n```python\n\nimport asyncio\nimport uvicorn\nimport simple_http_server.server as server\nfrom simple_http_server.server import ASGIProxy\n\n\nasgi_proxy: ASGIProxy = None\ninit_asgi_proxy_lock: asyncio.Lock = asyncio.Lock()\n\n\nasync def init_asgi_proxy():\n global asgi_proxy\n if asgi_proxy == None:\n async with init_asgi_proxy_lock:\n if asgi_proxy == None:\n server.scan(base_dir=\"tests/ctrls\", regx=r'.*controllers.*')\n asgi_proxy = server.init_asgi_proxy(resources={\"/public/*\": \"tests/static\"})\n\nasync def app(scope, receive, send):\n await init_asgi_proxy()\n await asgi_proxy.app_proxy(scope, receive, send)\n\ndef main():\n config = uvicorn.Config(\"main:app\", host=\"0.0.0.0\", port=9090, log_level=\"info\")\n asgi_server = uvicorn.Server(config)\n asgi_server.run()\n\nif __name__ == \"__main__\":\n main()\n\n```\n\n## Thanks\n\nThe code that process websocket comes from the following project: https://github.com/Pithikos/python-websocket-server\n",
"bugtrack_url": null,
"license": "MIT License Copyright (c) 2018 Keijack Wu Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.",
"summary": "This is a simple http server, use MVC like design.",
"version": "0.24.1",
"project_urls": {
"homepage": "https://github.com/keijack/python-simple-http-server",
"repository": "https://github.com/keijack/python-simple-http-server"
},
"split_keywords": [
"http-server",
"websocket",
"http",
"web",
"web-server"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "fb59de563f91d1a70c80e8f00b676672364486419ea58d1ce2c430c69d543e8b",
"md5": "739075dcd4a7d8a05942ac7c45a63185",
"sha256": "e3ce5e9c30352ae3be441135695eefefc81bf9461c1ba17d221b7e1a499579c5"
},
"downloads": -1,
"filename": "simple_http_server-0.24.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "739075dcd4a7d8a05942ac7c45a63185",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 63639,
"upload_time": "2024-03-05T11:47:53",
"upload_time_iso_8601": "2024-03-05T11:47:53.271579Z",
"url": "https://files.pythonhosted.org/packages/fb/59/de563f91d1a70c80e8f00b676672364486419ea58d1ce2c430c69d543e8b/simple_http_server-0.24.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "f231a939d6773d95cc6415fdf18c79bbdf1a46906809321cbf3b2aa5756d36f5",
"md5": "febce2e8d50924fa37ae7ebe9a765f0d",
"sha256": "a2cc923ae2cd8188382b6f685671c20166168d2e3f978a1f6933926085247c81"
},
"downloads": -1,
"filename": "simple_http_server-0.24.1.tar.gz",
"has_sig": false,
"md5_digest": "febce2e8d50924fa37ae7ebe9a765f0d",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 60721,
"upload_time": "2024-03-05T11:47:55",
"upload_time_iso_8601": "2024-03-05T11:47:55.915508Z",
"url": "https://files.pythonhosted.org/packages/f2/31/a939d6773d95cc6415fdf18c79bbdf1a46906809321cbf3b2aa5756d36f5/simple_http_server-0.24.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-03-05 11:47:55",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "keijack",
"github_project": "python-simple-http-server",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "simple-http-server"
}
