在线时间:8:00-16:00
迪恩网络APP
随时随地掌握行业动态
扫描二维码
关注迪恩网络微信公众号
开源软件名称:python-simple-http-server开源软件地址:https://gitee.com/keijack/python-simple-http-server开源软件介绍:python-simple-http-server简介这是一个轻量级编写的 HTTP 服务器,源生支持 websocket,你可以非常容易的搭建一个 Restful API。其中一些请求的转发等参考了 SpringMVC 的设计。 支持的 Python 的版本Python 3.7+ 为什么要选择这个项目?
依赖这个工程本身并不依赖任何其他的库,但是如果你需要运行在 python3 -m pip install websocket-client 安装python3 -m pip install simple_http_server 编写控制器配置路由信息我们接下来,将处理请求的函数命名为 控制器函数(Controller Function)。 类似 Spring MVC,我们使用描述性的方式来将配置请求的路由(在 Java 中,我们会使用标注 Annotation,而在 Python,我们使用 decorator,两者在使用时非常类似)。 基础的配置如下,该例子中,请求 /index 将会路由到当前的方法中。 请注意,其中 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 的配置如下。该配置下, @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 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>" 你也可以通过 @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),不过你更喜欢使用对象的话,你可以将你的 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. "} 如果你想使得控制类为单例的话,你可以使用 @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. "} 你也可以在类上使用 @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(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. "} 从 # 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}" 从 @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,你可以自定义自己的 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()) |
2023-10-27
2022-08-15
2022-08-17
2022-09-23
2022-08-13
请发表评论