The pywss from clearave

Pywss - Python Web/WebSocket Server

Pywss 是一个轻量级的 Python Web 框架。

主要特性：

重点版本迭代说明

0.1.10
- 修复部分路由BUG
0.1.9
- 默认支持keep-alive
0.1.7
- 调整json/form解析
0.1.4
- 修复signal无法在子线程注册
0.1.3
- 支持openapi
- 支持swagger ui
0.1.2
- 修复Content-Length丢失问题
0.1.1
- 项目重构

一、快速开始

1、安装 pywss

pip3 install pywss

2、搭建web服务

创建 main.py 文件，写入以下代码：

import pywss

def hello(ctx: pywss.Context):
  ctx.write("hello world")

def main():
    app = pywss.App()
    app.get("/hello", hello)
    app.run()

if __name__ == '__main__':
    main()

启动服务，至此，一个简单的 hello world 服务就启动了。

python3 main.py

1、初始化app

import pywss

# 初始化app
app = pywss.App()

# 启动服务
app.run(port=8080)

参数说明:

host: 默认 0.0.0.0
port: 默认 8080
grace: 退出滞留时间，默认为0

2、绑定路由

import pywss

# 初始化App
app = pywss.App()

# 注册路由 - get/post/head/put/delete/patch/options/any
app.get("/exact/match", lambda ctx: ctx.write({"hello": "world"}))
app.get("/part/{match}", lambda ctx: ctx.write({"hello": ctx.paths["match"]}))
app.get("/blur/match/*", lambda ctx: ctx.write({"hello": "world"}))

# Restful Style
class RouteV1:

    def http_get(self, ctx: pywss.Context):
        ctx.write({"hello": "world"})

    def http_post(self, ctx: pywss.Context):
        ctx.write({"hello": "world"})

app.view("/v1", RouteV1())

app.run(port=8080)

路由处理函数handler仅接收一个参数，就是pywss.Context。

此外，路由支持多种匹配方式：

/hello/world：精确匹配
/hello/{world}：局部匹配（注意：对应路径参数可通过ctx.paths["world"]获取`）
/hello/*：模糊匹配（注意：路由最后一位必须是*）

3、创建子路由

pywss 支持通过app.party来实现丰富的路由管理

import pywss

def hello(ctx: pywss.Context):
    ctx.write({"hello": ctx.path})

app = pywss.App()

v1 = app.party("/api/v1")
v1.get("/hello", lambda ctx: ctx.write({"hello": "v1"}))
v1.post("/hello/{name}", hello)

v2 = app.party("/api/v2")
v2.get("/hello", lambda ctx: ctx.write({"hello": "v2"}))
v2.post("/hello/{name}", hello)

app.run(port=8080)

在终端界面执行：

$ curl localhost:8080/api/v1/hello
{"hello": "v1"}

$ curl -X POST localhost:8080/api/v1/hello/pywss
{"hello": "/api/v1/hello/pywss"}

$ curl localhost:8080/api/v2/hello
{"hello": "v2"}

$ curl -X POST localhost:8080/api/v2/hello/pywss
{"hello": "/api/v2/hello/pywss"}

4、使用中间件

pywss 支持通过use注册全局中间件，也支持单个路由绑定中间件。

使用中间件时，注意需要调用ctx.next()才能继续往后执行，否则会中断此次请求。

import pywss, time

# 日志中间件，单次请求结束后输出cost耗时 - 根据响应码判断输出不同级别日志
def log_handler(ctx: pywss.Context):  
    start = time.time()
    ctx.next()  # 调用 next 进入到下一个 handler
    cost = time.time() - start
    if ctx.response_status_code < 300:
        ctx.log.info(cost)
    elif ctx.response_status_code < 400:
        ctx.log.warning(cost)
    else:
        ctx.log.error(cost)

# 认证中间件
def auth_handler(ctx: pywss.Context):  
    # 校验请求参数
    if ctx.paths["name"] != "pywss":  
        ctx.set_status_code(pywss.StatusUnauthorized)
        return
    ctx.next()

app = pywss.App()

# 注册全局中间件
app.use(log_handler)  

# 中间件也可以直接路由处注册
app.get("/hello/{name}", auth_handler, lambda ctx: ctx.write({"hello": "world"}))  

app.run()

5、升级WebSocket

WebSocket 本质基于 HTTP GET 升级实现，Pywss 则通过WebSocketUpgrade完成此处升级。

import pywss

def websocket(ctx: pywss.Context):
    # 升级 WebSocket
    err = pywss.WebSocketUpgrade(ctx)
    if err:
        ctx.log.error(err)
        ctx.set_status_code(pywss.StatusBadRequest)
        return
    # 轮询获取消息，实际使用场景建议引入心跳/探活机制
    while True:
        data = ctx.ws_read()
        ctx.log.info(data)
        ctx.ws_write(b"hello")


app = pywss.App()

app.get("/websocket", websocket)

app.run()

测试需要打开浏览器 -> F12 -> 控制台，输入以下代码：

ws = new WebSocket("ws://127.0.0.1:8080/websocket");
ws.onmessage = function (ev) {
    console.log(ev.data);
}
ws.onclose = function (ev) {
    console.log('Connect Closed')
}
ws.onopen = function() {
    if (ws.readyState === WebSocket.OPEN) {
        ws.send('hello??')
    }
}

其他具体使用场景/用例，可以参考多人在线协同编辑luckysheet、多人聊天室

6、openapi & swagger ui

import pywss

@pywss.openapi.docs(
    summary="此处是接口摘要 - 可选",
    description="此处是接口描述 - 可选",
    params={
        "page_size": "此处是参数说明",
        "username:query": "可以指定参数属于query(默认参数)",
        "name:path,required": "path表示为路径参数，required表示必填参数",
        "Auth:header,required": "参数支持：query、path、header、cookie",
    },
    request={"请求说明": "此处是请求示例"},
    response={"响应说明": "此处是响应示例"},
)
def hello(ctx: pywss.Context):
    ctx.write({
        "hello": "world",
        "page_size": ctx.params.get("page_size", 10),
        "username": ctx.params.get("username", "username"),
        "name": ctx.paths.get("name", "name"),
        "Auth": ctx.headers.get("Auth", "Auth"),
    })

app = pywss.App()

# 开启 openapi
app.openapi(  
    title="OpenAPI",
    version="0.0.1",
    openapi_json_route="/openapi.json",
    openapi_ui_route="/docs",
)

app.post("/hello/{name}", hello)

app.run()

打开浏览器，访问 localhost:8080/docs

7、静态文件服务器

import pywss

app = pywss.App()

# 注册静态资源，需要指定文件根目录
app.static("/static", rootDir="/rootDir") 

app.run()

假设已注册目录/rootDir结构如下，则可以通过 localhost:8080/static/index.html 进行访问

- rootDir
    - index.html
    - 200.html
    - 500.html

8、单元测试

import pywss

app = pywss.App()

app.get("/test", lambda ctx: ctx.set_status_code(204))

# 基于app创建HttpRequest
req = pywss.HttpTestRequest(app)  

# 发起Get请求，获取resp
resp = req.get("/test")  

assert resp.status_code == 204

可以参考 pywss单元测试

8、命令行启动

如果你只是想快速且简单的起一个服务，那么你还可以通过命令pywss的方式：

查看帮助指令

pywss -h

启动静态文件服务：
- --static表示本地路径 : 路由前缀，即将本地路径下的文件映射到指定路由
- --port表示端口号

pywss --static=".:/" --port=8080

通过http://localhost:8080/访问

启动web服务：
- --route表示method : route : code : body，即指定响应信息

pywss --route="GET:/hello:200:hello, world" --route="GET:/ok:204:" --port=8080

通过http://localhost:8080/hello访问

三、参数说明

1、请求参数

Context
- ctx.app: app
- ctx.fd: socket.socket类型，一般用于写操作
- ctx.rfd: socket.makefile类型，一般用于读操作
- ctx.method: str类型，请求方法，如 GET/POST/PUT/DELETE
- ctx.path: str类型，请求路径，如 /api/v1/query
- ctx.paths: dict类型，请求路径参数，如 /api/v1/query/{name}
- ctx.route: str类型，匹配路由的路径，如 GET/api/v1/query
- ctx.cookies: dict类型，用于存储请求cookies数据
- ctx.body(): bytes类型，获取用户请求报文body
- ctx.json(): 解析用户请求，等同于json.loads(self.body())，需要遵循一定json格式
- ctx.form(): 解析用户请求，需要遵循一定form格式
- ctx.params: dict类型，用于存储解析的query参数
  - http://github.com/czasg/pywss?code=1&callback=2->{"code": "1", "callback": "2"}
  - http://github.com/czasg/pywss?code=1&code=2->{"code": ["1", "2"]}
- ctx.headers: dict类型，用于存储解析的header参数

2、响应参数

Context
- ctx.set_status_code: 设置响应状态码
- ctx.set_header: 设置响应头
- ctx.set_content_type: 设置响应类型
- ctx.set_cookie: 设置响应cookie
- ctx.write: 用于写请求
- ctx.write_text: 同ctx.write
- ctx.write_json: 同ctx.write
- ctx.write_file: 同ctx.write
- ctx.ws_read: WebSocket 读请求，需要pywss.WebSocketUpgrade升级后使用
- ctx.ws_write: WebSocket 写请求，需要pywss.WebSocketUpgrade升级后使用
- ctx.flush: 一般不需要自己调用

clearave / pywss Goto Github PK

pywss's Introduction