• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    公众号

python-simple-http-server: 一个超轻量级的 HTTP Server,支持线程和协程模式,源生 ...

原作者: [db:作者] 来自: 网络 收藏 邀请

开源软件名称:

python-simple-http-server

开源软件地址:

https://gitee.com/keijack/python-simple-http-server

开源软件介绍:

python-simple-http-server

PyPI version

简介

这是一个轻量级编写的 HTTP 服务器,源生支持 websocket,你可以非常容易的搭建一个 Restful API。其中一些请求的转发等参考了 SpringMVC 的设计。

支持的 Python 的版本

Python 3.7+

为什么要选择这个项目?

  • 轻量级
  • 函数化编程
  • 支持 Session 会话,并且可以通过这个扩展支持分布式会话。
  • 支持过滤器链
  • Spring MVC 风格的请求映射配置
  • 简单易用
  • 支持 SSL
  • 支持 websocket
  • 编写风格自由
  • 可嵌入到 WSGI 标准当中
  • 支持协程模式,该模式下,你的所有控制器均运行在一个线程当中。

依赖

这个工程本身并不依赖任何其他的库,但是如果你需要运行在 tests 目录下的单元测试,那么,你需要安装 websocket-client 库:

python3 -m pip install websocket-client

安装

python3 -m pip install simple_http_server

编写控制器

配置路由信息

我们接下来,将处理请求的函数命名为 控制器函数(Controller Function)

类似 Spring MVC,我们使用描述性的方式来将配置请求的路由(在 Java 中,我们会使用标注 Annotation,而在 Python,我们使用 decorator,两者在使用时非常类似)。

基础的配置如下,该例子中,请求 /index 将会路由到当前的方法中。

请注意,其中 request_map 还有另外一个别称 route,你可以选择熟悉的标注使用。

from simple_http_server import request_map@request_map("/index")def your_ctroller_function():    return "<html><body>Hello, World!</body></html>"

@request_map 会接收两个参数,第二个参数会指定当前的控制器函数会处理请求中哪些方法(Method),以下的例子中的方法,仅会处理方法为 GET 的请求。

@request_map("/say_hello", method="GET")def your_ctroller_function():    return "<html><body>Hello, World!</body></html>"

method参数同时也可以是一个列表,以下的例子中,该控制器函数会处理方法为 GET、POST、PUT 的请求

@request_map("/say_hello", method=["GET", "POST", "PUT"])def your_ctroller_function():    return "<html><body>Hello, World!</body></html>"

匹配路由时,除了指定具体的路径之外,你还可以指定路径的某一段是可变的,这部方可变的部分将会作为路径参数放入请求中,如何取得这些路径参数我们将会在 取得请求参数 一节具体说明。@request_map 的配置如下。该配置下,/say/hello/to/world/say/hello/to/jhon 等 url 都能访问该控制器方法。

@request_map("/say_hello/to/{name}", method=["GET", "POST", "PUT"])def your_ctroller_function():    return "<html><body>Hello, World!</body></html>"

你可以给一个控制器函数增加多个 @request_map

@request_map("/")@request_map("/index")def index():    return "<html><body>Hello, World!</body></html>"

你也可以使用协程来编写你的控制器,注意:当你不是用协程模式启动时,你的协程会运行在处理该请求的线程当中。如果以协程模式启动,所有你的控制器,无论你有没有使用async def定义,均通过协程的方式运行在一个线程中。

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)

取的请求中的信息

参考了 Spring MVC 的设计,获取请求方法的方式非常自由。其中最基本的方法是取得 Request 对象,该对象中包含了所有其他方式能够获取的内容。

from simple_http_server import request_mapfrom simple_http_server import Request@request_map("/say_hello/to/{name}", method=["GET", "POST", "PUT"])def your_ctroller_function(req=Request()):    """ 请注意关键字参数传入的默认参数是一个 Request 对象,而不是类本身。 """    ##    # 该请求的方法,可能是 "OPTIONS", "GET", "HEAD", "POST", "PUT", "DELETE", "TRACE", "CONNECT" 中的一个    print(req.method)    ##    # 该请求的路径,就是 /say/hello/to/xxx    print(req.path)    ##    #  一个 dict 对象,包含了所有的头部参数。    print(req.headers)    ##    # 一个 dict 对象,包含了请求的参数,包括了来源于QueryString与 body 中的参数    # 该对象仅存储一个参数,如果同一个参数名传入了多个值,该对象只记录第一个值。    print(req.parameter)    ##    # 一个 dict 对象,类似,req.parameter,但该对象包含了所有的参数    # 该对象的值总是会返回一个列表,即使一个参数名只传入了一个参数。    print(req.parameters)    ##    # 返回 query string    print(req.query_string)    ##    # 返回一个 Cookies 对象,该对象是 http.cookies.SimpleCookie 的子类    print(req.cookies)    ##    # 一个 dict 对象,包含了所有的路径上的参数,本例中是 {"name": "xxx"}    print(req.path_values)    ##    # 请求体部分,在 3.6 中是一个 bytes 对象。2.7 中是一个 str 对象    print(req.body)    ##    # 当你的请求的 Content-Type 是 application/json 时,框架会自动将请求体中的 JSON 对象加载为一个dict对象。    print(req.json)    ##    # Session    session = req.getSession()    ins = session.get_attribute("in-session")    session.set_attribute("in-session", "Hello, Session!")    return "<html><body>Hello, World!</body></html>"

你也可以通过 ModelDict 来直接取得 request 里的参数。

@request_map("/say_hello/to/{name}", method=["GET", "POST", "PUT"])def your_ctroller_function(model=ModelDict(), name=PathValue()):    # 如果你访问 http://.../say_hello/to/keijack?a=1&b=2&b=3    print(name) # 输出 keijack    print(model["a"]) # 输出 1    print(model["b"]) # 取得一个列表,输出 ["2,", "3"]    return "<html><body>Hello, World!</body></html>"

我们还可以通过更直接的参数和关键字参数来获取请求中的信息,使得编码更加简洁和方便。

from simple_http_server import request_map@request_map("/say_hello/to/{name}", method=["GET", "POST", "PUT"])def your_ctroller_function(        user_name, # 传入 req.parameter["user_name"],如果该参数为空,则会响应为 400 参数错误        password, # 传入 req.parameter["password"],如果参数为空,则会响应为 400 参数错误        nickName="", # 传入 req.parameter["nickName"],如果参数为空,则会传入 ""        age=16, # 传入 int(req.parameter["age"]),如果传入空则传入 16,如果传入的不是数字类型,会返回 400 参数错误        male=True, # 传入 0, false, 空字符串 为 False,其他均为 True,如果不传入,传入这里指定的默认值        skills=[], # 传入 req.parameters["skills"],会返回一个数组,如果没有传入任何的内容,则返回这里指定的数组        extra={} # 传入 json.loads(req.parameter["extra"]),如果不传入则传入这里指定的 dict 对象,如果传入的字符串不是 JSON 格式,则响应为 400 参数错误    ):    return "<html><body>Hello, World!</body></html>"

以上的是基础类型的获取,实施上,我们还提供了几个类,通过这些类,你还能快速地获取一些在请求头中,Cookies 中,请求体,路径中的信息。以下是一些代码实例:

from simple_http_server import request_mapfrom simple_http_server import Parameterfrom simple_http_server import Parametersfrom simple_http_server import Headersfrom simple_http_server import Headerfrom simple_http_server import Cookiesfrom simple_http_server import Cookiefrom simple_http_server import PathValuefrom simple_http_server import Session@request_map("/say_hello/to/{name}", method=["GET", "POST", "PUT"])def your_ctroller_function(        user_name=Parameter("userName", required=True), # 传入 req.parameter["userName"],如果该参数为空,则会响应为 400 参数错误        password=Parameter("password", required=True), # 传入 req.parameter["password"],如果参数为空,则会响应为 400 参数错误        nickName=Parameter(default=""), # 传入 req.parameter["nickName"],如果参数为空,则会传入 "",参数名和        skills=Parameters(required=True), # 传入 req.parameters["skills"],会返回一个数组,如果没有传入任何的内容,则响应为 400 参数错误        all_headers=Headers(), # 传入 req.headers        user_token=Header(name="userToken", required=True), # 传入 req.headers["userToken"],如果请求头中没有 "userToken" 字段,则响应为 400 参数错误        all_cookies=Cookies(), # 传入 req.cookies,返回所有当前请求的 cookies        user_info=Cookie("userInfo", required=False), # 传入 req.cookies["userInfo"],如果没有该 cookie,则响应为 400 参数错误        name=PathValue("name"), # 传入 req.path_values["name"],返回路径中你路由配置中匹配 {name} 的字符串,        session=Session() # 传入 req.getSession(True),取得当前 request 的 Session 会话,如果没有会创建一个。    ):    return "<html><body>Hello, World!</body></html>"

从上述的例子我们看出,这些类中的参数均有默认值,即使不传入,也能返回正确的数据。除了上述的这些例子之外,我们还有一些额外的情况。例如请求的 Content-Type 是 application/json,然后我们将 JSON 字符串直接写入请求体中,我们可以这样获取信息:

from simple_http_server import request_mapfrom simple_http_server import JSONBody@request_map("/from_json_bldy", method=["post", "put", "delete"])def your_json_body_controller_function(data=JSONBody()):    ##    #  JSONBody 是 dict 的子类,你可以直接其是一个 dict 来使用    print(data["some_key"])    return "<html><body>Hello, World!</body></html>"

我们也支持使用 multipart/form-data 上传文件,你可以这样获取文件:

from simple_http_server import request_mapfrom simple_http_server import MultipartFile@request_map("/upload", method="POST")def upload(        img=MultipartFile("img", required=True) # 如果没有传入 img 参数,或者该参数不是一个文件,均响应为 400 参数错误        ):    root = os.path.dirname(os.path.abspath(__file__))    ##    # 获取上传文件的 content-type    print(img.content_type)    ##    # MultipartFile.content 在 3.6 中为 bytes 类型,在 2.7 中为字符串    print(img.content)    ##    # 还可以通过内置的 save_to_file 将内容直接写入到某个文件中    img.save_to_file(root + "/uploads/imgs/" + img.filename)    return "<!DOCTYPE html><html><body>upload ok!</body></html>"

你也可以使用 Python3 的函数变量标注(variable annotation) 来定义你需要取得的数据类型。

@request_map("/say_hello/to/{name}", method=["GET", "POST", "PUT"])def your_ctroller_function(        user_name: str, # 传入 req.parameter["user_name"],如果该参数为空,则会响应为 400 参数错误        password: str, # 传入 req.parameter["password"],如果参数为空,则会响应为 400 参数错误        skills: list, # 传入 req.parameters["skills"],会返回一个数组,如果没有传入任何的内容,则响应为 400 参数错误        all_headers: Headers, # 传入 req.headers        user_token: Header, # 传入 req.headers["user_token"],如果请求头中没有 "userToken" 字段,则响应为 400 参数错误        all_cookies: Cookies, # 传入 req.cookies,返回所有当前请求的 cookies        user_info: Cookie, # 传入 req.cookies["user_info"],如果没有该 cookie,则响应为 400 参数错误        name: PathValue, # 传入 req.path_values["name"],返回路径中你路由配置中匹配 {name} 的字符串,        session: Session # 传入 req.getSession(True),取得当前 request 的 Session 会话,如果没有会创建一个。    ):    return "<html><body>Hello, World!</body></html>"

我们建议使用函数式编程来编写你的控制器(Controller),不过你更喜欢使用对象的话,你可以将你的@request_map 用在类方法上,下面的例子中,每一个请求进来之后,系统会自动创建一个对象来调用该方法。:

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. "}

如果你想使得控制类为单例的话,你可以使用 @controller 装饰器。

@controllerclass 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. "}

你也可以在类上使用 @request_map,类上的路径会作为访问路径的一部分。

@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. "}

如果你的类有初始化变量,你也可以在 @controller 中设定。

@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. "}

0.7.0 开始 @request_map 还支持正则表达式。

# url `/reg/abcef/aref/xxx` 能匹配到以下控制器:@route(regexp="^(reg/(.+))$", method="GET")def my_reg_ctr(reg_groups: RegGroups, reg_group: RegGroup = RegGroup(1)):    print(reg_groups) # 输出 ("reg/abcef/aref/xxx", "abcef/aref/xxx")    print(reg_group) # 输出 "abcef/aref/xxx"    return f"{self._name}, {reg_group.group},{reg_group}"

0.14.0 开始,@request_map 还支持通配符。你只能使用前置通配符或者时后知通配符其中一项,不能同时配置。如有复杂需求则请使用正则表达式来进行匹配。

@request_map("/star/*") # /star/c 可以匹配,但是 /star/c/d 不能。@request_map("*/star") # /c/star 可以匹配,但是 /c/d/star 不能。def star_path(path_val=PathValue()): # 你可以通过 PathValue 取得 url 中 匹配 * 通配符的内容。    return f"<html><body>{path_val}</body></html>"@request_map("/star/**") # /star/c 和 /star/c/d 均可以匹配。@request_map("**/star") # /c/star 和 /c/d/stars 均可匹配。def star_path(path_val=PathValue()): # 你可以通过 PathValue 通配符取得 url 中 匹配 ** 通配符的内容。    return f"<html><body>{path_val}</body></html>"

在控制器类中使用正则表达式的 request_map:

@controller(args=["ctr_name"], kwargs={"desc": "this is a key word argument"})@request_map("/obj", method="GET") # 请不要在这里配置 regexp,因为不工作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/(.+))$") # 从类装饰器来的 `/obj` 会被无视, 但 `method`(GET) 则依然有用    def my_ctrl_mth(self, name: str):        return {"message": f"hello, {name}, {self._name} says. "}

Session

默认情况下,Session 中的数据会存储到本地,如果你需要做分布式 Session,例如将 Session内容存储在 Redis 或者 Memcache,你可以自定义自己的 SessionSessionFactory,然后创建一个你定义的 SessionFactory 对象通过 simple_http_server.set_session_factory 设置到框架中。

from simple_http_server import Session, SessionFactory, set_session_factoryclass 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 implementationclass 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()) 

鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap